import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
InMemoryPlan(QueueMetrics queueMetrics, SharingPolicy policy,
    ReservationAgent agent, Resource totalCapacity, long step,
    ResourceCalculator resCalc, Resource minAlloc, Resource maxAlloc,
    String queueName, Planner replanner, boolean getMoveOnExpiry, Clock clock) {
  this.queueMetrics = queueMetrics;
  this.policy = policy;
  this.agent = agent;
  this.step = step;
  this.totalCapacity = totalCapacity;
  this.resCalc = resCalc;
  this.minAlloc = minAlloc;
  this.maxAlloc = maxAlloc;
  this.rleSparseVector = new RLESparseResourceAllocation(resCalc, minAlloc);
  this.queueName = queueName;
  this.replanner = replanner;
  this.getMoveOnExpiry = getMoveOnExpiry;
  this.clock = clock;
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
protected Plan initializePlan(String planQueueName) throws YarnException {
  String planQueuePath = getPlanQueuePath(planQueueName);
  SharingPolicy adPolicy = getAdmissionPolicy(planQueuePath);
  adPolicy.init(planQueuePath, getReservationSchedulerConfiguration());
  // Calculate the max plan capacity
  Resource minAllocation = getMinAllocation();
  Resource maxAllocation = getMaxAllocation();
  ResourceCalculator rescCalc = getResourceCalculator();
  Resource totCap = getPlanQueueCapacity(planQueueName);
  Plan plan =
      new InMemoryPlan(getRootQueueMetrics(), adPolicy,
          getAgent(planQueuePath), totCap, planStepSize, rescCalc,
          minAllocation, maxAllocation, planQueueName,
          getReplanner(planQueuePath), getReservationSchedulerConfiguration()
          .getMoveOnExpiry(planQueuePath));
  LOG.info("Intialized plan {0} based on reservable queue {1}",
      plan.toString(), planQueueName);
  return plan;
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
InMemoryReservationAllocation(ReservationId reservationID,
    ReservationDefinition contract, String user, String planName,
    long startTime, long endTime,
    Map<ReservationInterval, ReservationRequest> allocationRequests,
    ResourceCalculator calculator, Resource minAlloc) {
  this.contract = contract;
  this.startTime = startTime;
  this.endTime = endTime;
  this.reservationID = reservationID;
  this.user = user;
  this.allocationRequests = allocationRequests;
  this.planName = planName;
  resourcesOverTime = new RLESparseResourceAllocation(calculator, minAlloc);
  for (Map.Entry<ReservationInterval, ReservationRequest> r : allocationRequests
      .entrySet()) {
    resourcesOverTime.addInterval(r.getKey(), r.getValue());
    if (r.getValue().getConcurrency() > 1) {
      hasGang = true;
    }
  }
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
/**
 * This method produces an Allocation that includes the current view
 * of the resources that will be allocated to and preempted from this
 * application.
 *
 * @param rc
 * @param clusterResource
 * @param minimumAllocation
 * @return an allocation
 */
public synchronized Allocation getAllocation(ResourceCalculator rc,
    Resource clusterResource, Resource minimumAllocation) {

  Set<ContainerId> currentContPreemption = Collections.unmodifiableSet(
      new HashSet<ContainerId>(containersToPreempt));
  containersToPreempt.clear();
  Resource tot = Resource.newInstance(0, 0, 0);
  for(ContainerId c : currentContPreemption){
    Resources.addTo(tot,
        liveContainers.get(c).getContainer().getResource());
  }
  int numCont = (int) Math.ceil(
      Resources.divide(rc, clusterResource, tot, minimumAllocation));
  ResourceRequest rr = ResourceRequest.newInstance(
      Priority.UNDEFINED, ResourceRequest.ANY,
      minimumAllocation, numCont);
  ContainersAndNMTokensAllocation allocation =
      pullNewlyAllocatedContainersAndNMTokens();
  Resource headroom = getHeadroom();
  setApplicationHeadroomForMetrics(headroom);
  return new Allocation(allocation.getContainerList(), headroom, null,
    currentContPreemption, Collections.singletonList(rr),
    allocation.getNMTokenList());
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
@Before
public void setUp() {
  clock = mock(Clock.class);
  plan = mock(Plan.class);
  rSystem = mock(ReservationSystem.class);
  plans.put(PLAN_NAME, plan);
  rrValidator = new ReservationInputValidator(clock);
  when(clock.getTime()).thenReturn(1L);
  ResourceCalculator rCalc = new DefaultResourceCalculator();
  Resource resource = Resource.newInstance(10240, 10, 10);
  when(plan.getResourceCalculator()).thenReturn(rCalc);
  when(plan.getTotalCapacity()).thenReturn(resource);
  when(rSystem.getQueueForReservation(any(ReservationId.class))).thenReturn(
      PLAN_NAME);
  when(rSystem.getPlan(PLAN_NAME)).thenReturn(plan);
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
private void checkCSLeafQueue(MockRM rm,
    SchedulerApplication<SchedulerApplicationAttempt> app,
    Resource clusterResource, Resource queueResource, Resource usedResource,
    int numContainers) {
  LeafQueue leafQueue = (LeafQueue) app.getQueue();
  // assert queue used resources.
  assertEquals(usedResource, leafQueue.getUsedResources());
  assertEquals(numContainers, leafQueue.getNumContainers());

  ResourceCalculator calc =
      ((CapacityScheduler) rm.getResourceScheduler()).getResourceCalculator();
  float usedCapacity =
      Resources.divide(calc, clusterResource, usedResource, queueResource);
  // assert queue used capacity
  assertEquals(usedCapacity, leafQueue.getUsedCapacity(), 1e-8);
  float absoluteUsedCapacity =
      Resources.divide(calc, clusterResource, usedResource, clusterResource);
  // assert queue absolute capacity
  assertEquals(absoluteUsedCapacity, leafQueue.getAbsoluteUsedCapacity(),
    1e-8);
  // assert user consumed resources.
  assertEquals(usedResource, leafQueue.getUser(app.getUser())
    .getUsed());
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
private static YarnScheduler mockYarnScheduler() {
  YarnScheduler yarnScheduler = mock(YarnScheduler.class);
  when(yarnScheduler.getMinimumResourceCapability()).thenReturn(
      Resources.createResource(
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB));
  when(yarnScheduler.getMaximumResourceCapability()).thenReturn(
      Resources.createResource(
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB));
  when(yarnScheduler.getAppsInQueue(QUEUE_1)).thenReturn(
      Arrays.asList(getApplicationAttemptId(101), getApplicationAttemptId(102)));
  when(yarnScheduler.getAppsInQueue(QUEUE_2)).thenReturn(
      Arrays.asList(getApplicationAttemptId(103)));
  ApplicationAttemptId attemptId = getApplicationAttemptId(1);
  when(yarnScheduler.getAppResourceUsageReport(attemptId)).thenReturn(null);
  ResourceCalculator rc = new DefaultResourceCalculator();
  when(yarnScheduler.getResourceCalculator()).thenReturn(rc);
  return yarnScheduler;
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
public InMemoryPlan(QueueMetrics queueMetrics, SharingPolicy policy,
    ReservationAgent agent, Resource totalCapacity, long step,
    ResourceCalculator resCalc, Resource minAlloc, Resource maxAlloc,
    String queueName, Planner replanner, boolean getMoveOnExpiry,
    RMContext rmContext, Clock clock) {
  this.queueMetrics = queueMetrics;
  this.policy = policy;
  this.agent = agent;
  this.step = step;
  this.totalCapacity = totalCapacity;
  this.resCalc = resCalc;
  this.minAlloc = minAlloc;
  this.maxAlloc = maxAlloc;
  this.rleSparseVector = new RLESparseResourceAllocation(resCalc);
  this.queueName = queueName;
  this.replanner = replanner;
  this.getMoveOnExpiry = getMoveOnExpiry;
  this.clock = clock;
  this.rmStateStore = rmContext.getStateStore();
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
protected double calcCostOfInterval(long startTime, long endTime,
    Map<Long, Resource> planLoads,
    RLESparseResourceAllocation planModifications, Resource capacity,
    ResourceCalculator resCalc, long step) {

  // Sum costs in the interval [startTime,endTime)
  double totalCost = 0.0;
  for (long t = startTime; t < endTime; t += step) {
    totalCost += calcCostOfTimeSlot(t, planLoads, planModifications, capacity,
        resCalc);
  }

  // Return sum
  return totalCost;

}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
protected Plan initializePlan(String planQueueName) throws YarnException {
  String planQueuePath = getPlanQueuePath(planQueueName);
  SharingPolicy adPolicy = getAdmissionPolicy(planQueuePath);
  adPolicy.init(planQueuePath, getReservationSchedulerConfiguration());
  // Calculate the max plan capacity
  Resource minAllocation = getMinAllocation();
  Resource maxAllocation = getMaxAllocation();
  ResourceCalculator rescCalc = getResourceCalculator();
  Resource totCap = getPlanQueueCapacity(planQueueName);
  Plan plan =
      new InMemoryPlan(getRootQueueMetrics(), adPolicy,
          getAgent(planQueuePath), totCap, planStepSize, rescCalc,
          minAllocation, maxAllocation, planQueueName,
          getReplanner(planQueuePath), getReservationSchedulerConfiguration()
          .getMoveOnExpiry(planQueuePath), rmContext);
  LOG.info("Intialized plan {} based on reservable queue {}",
      plan.toString(), planQueueName);
  return plan;
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
public InMemoryReservationAllocation(ReservationId reservationID,
    ReservationDefinition contract, String user, String planName,
    long startTime, long endTime,
    Map<ReservationInterval, Resource> allocations,
    ResourceCalculator calculator, Resource minAlloc, boolean hasGang) {
  this.contract = contract;
  this.startTime = startTime;
  this.endTime = endTime;
  this.reservationID = reservationID;
  this.user = user;
  this.allocationRequests = allocations;
  this.planName = planName;
  this.hasGang = hasGang;
  resourcesOverTime = new RLESparseResourceAllocation(calculator);
  for (Map.Entry<ReservationInterval, Resource> r : allocations
      .entrySet()) {
    resourcesOverTime.addInterval(r.getKey(), r.getValue());
  }
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
public synchronized ApplicationResourceUsageReport getResourceUsageReport() {
  AggregateAppResourceUsage runningResourceUsage =
      getRunningAggregateAppResourceUsage();
  Resource usedResourceClone =
      Resources.clone(attemptResourceUsage.getAllUsed());
  Resource reservedResourceClone =
      Resources.clone(attemptResourceUsage.getReserved());
  Resource cluster = rmContext.getScheduler().getClusterResource();
  ResourceCalculator calc = rmContext.getScheduler().getResourceCalculator();
  float queueUsagePerc = calc.divide(cluster, usedResourceClone, Resources
      .multiply(cluster, queue.getQueueInfo(false, false).getCapacity()))
      * 100;
  float clusterUsagePerc =
      calc.divide(cluster, usedResourceClone, cluster) * 100;
  return ApplicationResourceUsageReport.newInstance(liveContainers.size(),
      reservedContainers.size(), usedResourceClone, reservedResourceClone,
      Resources.add(usedResourceClone, reservedResourceClone),
      runningResourceUsage.getMemorySeconds(),
      runningResourceUsage.getVcoreSeconds(),
      queueUsagePerc, clusterUsagePerc);
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
@Before
public void setUp() {
  clock = mock(Clock.class);
  plan = mock(Plan.class);
  rSystem = mock(ReservationSystem.class);
  plans.put(PLAN_NAME, plan);
  rrValidator = new ReservationInputValidator(clock);
  when(clock.getTime()).thenReturn(1L);
  ResourceCalculator rCalc = new DefaultResourceCalculator();
  Resource resource = Resource.newInstance(10240, 10);
  when(plan.getResourceCalculator()).thenReturn(rCalc);
  when(plan.getTotalCapacity()).thenReturn(resource);
  when(rSystem.getQueueForReservation(any(ReservationId.class))).thenReturn(
      PLAN_NAME);
  when(rSystem.getPlan(PLAN_NAME)).thenReturn(plan);
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
private static Resource getNonPartitionedMaxAvailableResourceToQueue(
    final ResourceCalculator rc, Resource totalNonPartitionedResource,
    CSQueue queue) {
  Resource queueLimit = Resources.none();
  Resource usedResources = queue.getUsedResources();

  if (Resources.greaterThan(rc, totalNonPartitionedResource,
      totalNonPartitionedResource, Resources.none())) {
    queueLimit =
        Resources.multiply(totalNonPartitionedResource,
            queue.getAbsoluteCapacity());
  }

  Resource available = Resources.subtract(queueLimit, usedResources);
  return Resources.max(rc, totalNonPartitionedResource, available,
      Resources.none());
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
private static YarnScheduler mockYarnScheduler() {
  YarnScheduler yarnScheduler = mock(YarnScheduler.class);
  when(yarnScheduler.getMinimumResourceCapability()).thenReturn(
      Resources.createResource(
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB));
  when(yarnScheduler.getMaximumResourceCapability()).thenReturn(
      Resources.createResource(
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB));
  when(yarnScheduler.getAppsInQueue(QUEUE_1)).thenReturn(
      Arrays.asList(getApplicationAttemptId(101), getApplicationAttemptId(102)));
  when(yarnScheduler.getAppsInQueue(QUEUE_2)).thenReturn(
      Arrays.asList(getApplicationAttemptId(103)));
  ApplicationAttemptId attemptId = getApplicationAttemptId(1);
  when(yarnScheduler.getAppResourceUsageReport(attemptId)).thenReturn(null);

  ResourceCalculator rs = mock(ResourceCalculator.class);
  when(yarnScheduler.getResourceCalculator()).thenReturn(rs);

  return yarnScheduler;
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
/**
 * This method produces an Allocation that includes the current view
 * of the resources that will be allocated to and preempted from this
 * application.
 *
 * @param rc
 * @param clusterResource
 * @param minimumAllocation
 * @return an allocation
 */
public synchronized Allocation getAllocation(ResourceCalculator rc,
    Resource clusterResource, Resource minimumAllocation) {

  Set<ContainerId> currentContPreemption = Collections.unmodifiableSet(
      new HashSet<ContainerId>(containersToPreempt));
  containersToPreempt.clear();
  Resource tot = Resource.newInstance(0, 0);
  for(ContainerId c : currentContPreemption){
    Resources.addTo(tot,
        liveContainers.get(c).getContainer().getResource());
  }
  int numCont = (int) Math.ceil(
      Resources.divide(rc, clusterResource, tot, minimumAllocation));
  ResourceRequest rr = ResourceRequest.newInstance(
      Priority.UNDEFINED, ResourceRequest.ANY,
      minimumAllocation, numCont);
  ContainersAndNMTokensAllocation allocation =
      pullNewlyAllocatedContainersAndNMTokens();
  Resource headroom = getHeadroom();
  setApplicationHeadroomForMetrics(headroom);
  return new Allocation(allocation.getContainerList(), headroom, null,
    currentContPreemption, Collections.singletonList(rr),
    allocation.getNMTokenList());
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
@Override
protected Resource getPlanQueueCapacity(String planQueueName) {
  Resource minAllocation = getMinAllocation();
  ResourceCalculator rescCalc = getResourceCalculator();
  CSQueue planQueue = capScheduler.getQueue(planQueueName);
  return rescCalc.multiplyAndNormalizeDown(capScheduler.getClusterResource(),
      planQueue.getAbsoluteCapacity(), minAllocation);
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
/**
 * Utility method to normalize a list of resource requests, by insuring that
 * the memory for each request is a multiple of minMemory and is not zero.
 */
public static void normalizeRequests(
  List<ResourceRequest> asks,
  ResourceCalculator resourceCalculator,
  Resource clusterResource,
  Resource minimumResource,
  Resource maximumResource) {
  for (ResourceRequest ask : asks) {
    normalizeRequest(
      ask, resourceCalculator, clusterResource, minimumResource,
      maximumResource, minimumResource);
  }
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
/**
 * Utility method to normalize a resource request, by insuring that the
 * requested memory is a multiple of minMemory and is not zero.
 */
public static void normalizeRequest(
  ResourceRequest ask,
  ResourceCalculator resourceCalculator,
  Resource clusterResource,
  Resource minimumResource,
  Resource maximumResource) {
  Resource normalized =
    Resources.normalize(
      resourceCalculator, ask.getCapability(), minimumResource,
      maximumResource, minimumResource);
  ask.setCapability(normalized);
}
 

import org.apache.hadoop.yarn.util.resource.ResourceCalculator; //导入依赖的package包/类
synchronized public NodeId getNodeIdToUnreserve(Priority priority,
    Resource resourceNeedUnreserve, ResourceCalculator rc,
    Resource clusterResource) {

  // first go around make this algorithm simple and just grab first
  // reservation that has enough resources
  Map<NodeId, RMContainer> reservedContainers = this.reservedContainers
      .get(priority);

  if ((reservedContainers != null) && (!reservedContainers.isEmpty())) {
    for (Map.Entry<NodeId, RMContainer> entry : reservedContainers.entrySet()) {
      NodeId nodeId = entry.getKey();
      Resource containerResource = entry.getValue().getContainer().getResource();
      
      // make sure we unreserve one with at least the same amount of
      // resources, otherwise could affect capacity limits
      if (Resources.lessThanOrEqual(rc, clusterResource,
          resourceNeedUnreserve, containerResource)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("unreserving node with reservation size: "
              + containerResource
              + " in order to allocate container with size: " + resourceNeedUnreserve);
        }
        return nodeId;
      }
    }
  }
  return null;
}