import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private int assignNodeLocalContainers(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority) {
  int assignedContainers = 0;
  ResourceRequest request = 
    application.getResourceRequest(priority, node.getNodeName());
  if (request != null) {
    // Don't allocate on this node if we don't need containers on this rack
    ResourceRequest rackRequest =
        application.getResourceRequest(priority, 
            node.getRMNode().getRackName());
    if (rackRequest == null || rackRequest.getNumContainers() <= 0) {
      return 0;
    }
    
    int assignableContainers = 
      Math.min(
          getMaxAllocatableContainers(application, priority, node, 
              NodeType.NODE_LOCAL), 
              request.getNumContainers());
    assignedContainers = 
      assignContainer(node, application, priority, 
          assignableContainers, request, NodeType.NODE_LOCAL);
  }
  return assignedContainers;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private int assignRackLocalContainers(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority) {
  int assignedContainers = 0;
  ResourceRequest request = 
    application.getResourceRequest(priority, node.getRMNode().getRackName());
  if (request != null) {
    // Don't allocate on this rack if the application doens't need containers
    ResourceRequest offSwitchRequest =
        application.getResourceRequest(priority, ResourceRequest.ANY);
    if (offSwitchRequest.getNumContainers() <= 0) {
      return 0;
    }
    
    int assignableContainers = 
      Math.min(
          getMaxAllocatableContainers(application, priority, node, 
              NodeType.RACK_LOCAL), 
              request.getNumContainers());
    assignedContainers = 
      assignContainer(node, application, priority, 
          assignableContainers, request, NodeType.RACK_LOCAL);
  }
  return assignedContainers;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private synchronized CSAssignment assignReservedContainer(
    FiCaSchedulerApp application, FiCaSchedulerNode node,
    RMContainer rmContainer, Resource clusterResource) {
  // Do we still need this reservation?
  Priority priority = rmContainer.getReservedPriority();
  if (application.getTotalRequiredResources(priority) == 0) {
    // Release
    return new CSAssignment(application, rmContainer);
  }

  // Try to assign if we have sufficient resources
  assignContainersOnNode(clusterResource, node, application, priority, 
      rmContainer, new ResourceLimits(Resources.none()));
  
  // Doesn't matter... since it's already charged for at time of reservation
  // "re-reservation" is *free*
  return new CSAssignment(Resources.none(), NodeType.NODE_LOCAL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
public synchronized void assign(List<Container> containers) 
throws IOException, YarnException {
  
  int numContainers = containers.size();
  // Schedule in priority order
  for (Priority priority : requests.keySet()) {
    assign(priority, NodeType.NODE_LOCAL, containers);
    assign(priority, NodeType.RACK_LOCAL, containers);
    assign(priority, NodeType.OFF_SWITCH, containers);

    if (containers.isEmpty()) { 
      break;
    }
  }
  
  int assignedContainers = numContainers - containers.size();
  LOG.info("Application " + applicationId + " assigned " + 
      assignedContainers + "/" + numContainers);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
@Test
/**
 * Ensure that when negative paramaters are given (signaling delay scheduling
 * no tin use), the least restrictive locality level is returned.
 */
public void testLocalityLevelWithoutDelays() {
  FSLeafQueue queue = Mockito.mock(FSLeafQueue.class);
  Priority prio = Mockito.mock(Priority.class);
  Mockito.when(prio.getPriority()).thenReturn(1);

  RMContext rmContext = resourceManager.getRMContext();
  ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1);
  FSAppAttempt schedulerApp =
      new FSAppAttempt(scheduler, applicationAttemptId, "user1", queue ,
          null, rmContext);
  assertEquals(NodeType.OFF_SWITCH, schedulerApp.getAllowedLocalityLevel(
      prio, 10, -1.0, -1.0));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private CSAssignment createReservedIncreasedCSAssignment(
    SchedContainerChangeRequest request) {
  CSAssignment assignment =
      new CSAssignment(request.getDeltaCapacity(), NodeType.NODE_LOCAL, null,
          application, false, false);
  Resources.addTo(assignment.getAssignmentInformation().getReserved(),
      request.getDeltaCapacity());
  assignment.getAssignmentInformation().incrReservations();
  assignment.getAssignmentInformation().addReservationDetails(
      request.getContainerId(), application.getCSLeafQueue().getQueuePath());
  assignment.setIncreasedAllocation(true);
  
  LOG.info("Reserved increase container request:" + request.toString());
  
  return assignment;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private CSAssignment createSuccessfullyIncreasedCSAssignment(
    SchedContainerChangeRequest request, boolean fromReservation) {
  CSAssignment assignment =
      new CSAssignment(request.getDeltaCapacity(), NodeType.NODE_LOCAL, null,
          application, false, fromReservation);
  Resources.addTo(assignment.getAssignmentInformation().getAllocated(),
      request.getDeltaCapacity());
  assignment.getAssignmentInformation().incrAllocations();
  assignment.getAssignmentInformation().addAllocationDetails(
      request.getContainerId(), application.getCSLeafQueue().getQueuePath());
  assignment.setIncreasedAllocation(true);
  
  // notify application
  application
      .getCSLeafQueue()
      .getOrderingPolicy()
      .containerAllocated(application,
          application.getRMContainer(request.getContainerId()));

  LOG.info("Approved increase container request:" + request.toString()
      + " fromReservation=" + fromReservation);    
  
  return assignment;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private synchronized CSAssignment assignReservedContainer(
    FiCaSchedulerApp application, FiCaSchedulerNode node,
    RMContainer rmContainer, Resource clusterResource) {
  // Do we still need this reservation?
  Priority priority = rmContainer.getReservedPriority();
  //优先级是否已经达到要求以分配reservation 的 containers
  if (application.getTotalRequiredResources(priority) == 0) {
    // Release
    return new CSAssignment(application, rmContainer);
  }

  // Try to assign if we have sufficient resources
  assignContainersOnNode(clusterResource, node, application, priority, 
      rmContainer, new ResourceLimits(Resources.none()));
  
  // Doesn't matter... since it's already charged for at time of reservation
  // "re-reservation" is *free*
  return new CSAssignment(Resources.none(), NodeType.NODE_LOCAL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private synchronized CSAssignment 
assignReservedContainer(FiCaSchedulerApp application, 
    FiCaSchedulerNode node, RMContainer rmContainer, Resource clusterResource) {
  // Do we still need this reservation?
  Priority priority = rmContainer.getReservedPriority();
  if (application.getTotalRequiredResources(priority) == 0) {
    // Release
    return new CSAssignment(application, rmContainer);
  }

  // Try to assign if we have sufficient resources
  assignContainersOnNode(clusterResource, node, application, priority, 
      rmContainer, false);
  
  // Doesn't matter... since it's already charged for at time of reservation
  // "re-reservation" is *free*
  return new CSAssignment(Resources.none(), NodeType.NODE_LOCAL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private int assignNodeLocalContainers(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority) {
  int assignedContainers = 0;
  ResourceRequest request = 
    application.getResourceRequest(priority, node.getHostName());
  if (request != null) {
    // Don't allocate on this node if we don't need containers on this rack
    ResourceRequest rackRequest =
        application.getResourceRequest(priority, 
            node.getRMNode().getRackName());
    if (rackRequest == null || rackRequest.getNumContainers() <= 0) {
      return 0;
    }
    
    int assignableContainers = 
      Math.min(
          getMaxAllocatableContainers(application, priority, node, 
              NodeType.NODE_LOCAL), 
              request.getNumContainers());
    assignedContainers = 
      assignContainer(node, application, priority, 
          assignableContainers, request, NodeType.NODE_LOCAL);
  }
  return assignedContainers;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private synchronized CSAssignment 
assignReservedContainer(FiCaSchedulerApp application, 
    FiCaSchedulerNode node, RMContainer rmContainer, Resource clusterResource) {
  // Do we still need this reservation?
  Priority priority = rmContainer.getReservedPriority();
  if (application.getTotalRequiredResources(priority) == 0) {
    // Release
    return new CSAssignment(application, rmContainer);
  }

  // Try to assign if we have sufficient resources
  assignContainersOnNode(clusterResource, node, application, priority, 
      rmContainer);
  
  // Doesn't matter... since it's already charged for at time of reservation
  // "re-reservation" is *free*
  return new CSAssignment(Resources.none(), NodeType.NODE_LOCAL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
@Test
/**
 * Ensure that when negative paramaters are given (signaling delay scheduling
 * no tin use), the least restrictive locality level is returned.
 */
public void testLocalityLevelWithoutDelays() {
  Queue queue = Mockito.mock(Queue.class);
  Priority prio = Mockito.mock(Priority.class);
  Mockito.when(prio.getPriority()).thenReturn(1);

  ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1);
  FSSchedulerApp schedulerApp =
      new FSSchedulerApp(applicationAttemptId, "user1", queue , null, null);
  assertEquals(NodeType.OFF_SWITCH, schedulerApp.getAllowedLocalityLevel(
      prio, 10, -1.0, -1.0));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private CSAssignment createReservedIncreasedCSAssignment(
    SchedContainerChangeRequest request) {
  CSAssignment assignment =
      new CSAssignment(request.getDeltaCapacity(), NodeType.NODE_LOCAL, null,
          application, CSAssignment.SkippedType.NONE, false);
  Resources.addTo(assignment.getAssignmentInformation().getReserved(),
      request.getDeltaCapacity());
  assignment.getAssignmentInformation().incrReservations();
  assignment.getAssignmentInformation().addReservationDetails(
      request.getContainerId(), application.getCSLeafQueue().getQueuePath());
  assignment.setIncreasedAllocation(true);
  
  LOG.info("Reserved increase container request:" + request.toString());
  
  return assignment;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private CSAssignment createSuccessfullyIncreasedCSAssignment(
    SchedContainerChangeRequest request, boolean fromReservation) {
  CSAssignment assignment =
      new CSAssignment(request.getDeltaCapacity(), NodeType.NODE_LOCAL, null,
          application, CSAssignment.SkippedType.NONE, fromReservation);
  Resources.addTo(assignment.getAssignmentInformation().getAllocated(),
      request.getDeltaCapacity());
  assignment.getAssignmentInformation().incrAllocations();
  assignment.getAssignmentInformation().addAllocationDetails(
      request.getContainerId(), application.getCSLeafQueue().getQueuePath());
  assignment.setIncreasedAllocation(true);
  
  // notify application
  application
      .getCSLeafQueue()
      .getOrderingPolicy()
      .containerAllocated(application,
          application.getRMContainer(request.getContainerId()));

  LOG.info("Approved increase container request:" + request.toString()
      + " fromReservation=" + fromReservation);    
  
  return assignment;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private int getMaxAllocatableContainers(FiCaSchedulerApp application,
    Priority priority, FiCaSchedulerNode node, NodeType type) {
  int maxContainers = 0;
  
  ResourceRequest offSwitchRequest = 
    application.getResourceRequest(priority, ResourceRequest.ANY);
  if (offSwitchRequest != null) {
    maxContainers = offSwitchRequest.getNumContainers();
  }

  if (type == NodeType.OFF_SWITCH) {
    return maxContainers;
  }

  if (type == NodeType.RACK_LOCAL) {
    ResourceRequest rackLocalRequest = 
      application.getResourceRequest(priority, node.getRMNode().getRackName());
    if (rackLocalRequest == null) {
      return maxContainers;
    }

    maxContainers = Math.min(maxContainers, rackLocalRequest.getNumContainers());
  }

  if (type == NodeType.NODE_LOCAL) {
    ResourceRequest nodeLocalRequest = 
      application.getResourceRequest(priority, node.getRMNode().getNodeAddress());
    if (nodeLocalRequest != null) {
      maxContainers = Math.min(maxContainers, nodeLocalRequest.getNumContainers());
    }
  }

  return maxContainers;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private int assignOffSwitchContainers(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority) {
  int assignedContainers = 0;
  ResourceRequest request = 
    application.getResourceRequest(priority, ResourceRequest.ANY);
  if (request != null) {
    assignedContainers = 
      assignContainer(node, application, priority, 
          request.getNumContainers(), request, NodeType.OFF_SWITCH);
  }
  return assignedContainers;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
/**
 * Return the level at which we are allowed to schedule containers, given the
 * current size of the cluster and thresholds indicating how many nodes to
 * fail at (as a fraction of cluster size) before relaxing scheduling
 * constraints.
 */
public synchronized NodeType getAllowedLocalityLevel(Priority priority,
    int numNodes, double nodeLocalityThreshold, double rackLocalityThreshold) {
  // upper limit on threshold
  if (nodeLocalityThreshold > 1.0) { nodeLocalityThreshold = 1.0; }
  if (rackLocalityThreshold > 1.0) { rackLocalityThreshold = 1.0; }

  // If delay scheduling is not being used, can schedule anywhere
  if (nodeLocalityThreshold < 0.0 || rackLocalityThreshold < 0.0) {
    return NodeType.OFF_SWITCH;
  }

  // Default level is NODE_LOCAL
  if (!allowedLocalityLevel.containsKey(priority)) {
    allowedLocalityLevel.put(priority, NodeType.NODE_LOCAL);
    return NodeType.NODE_LOCAL;
  }

  NodeType allowed = allowedLocalityLevel.get(priority);

  // If level is already most liberal, we're done
  if (allowed.equals(NodeType.OFF_SWITCH)) return NodeType.OFF_SWITCH;

  double threshold = allowed.equals(NodeType.NODE_LOCAL) ? nodeLocalityThreshold :
    rackLocalityThreshold;

  // Relax locality constraints once we've surpassed threshold.
  if (getSchedulingOpportunities(priority) > (numNodes * threshold)) {
    if (allowed.equals(NodeType.NODE_LOCAL)) {
      allowedLocalityLevel.put(priority, NodeType.RACK_LOCAL);
      resetSchedulingOpportunities(priority);
    }
    else if (allowed.equals(NodeType.RACK_LOCAL)) {
      allowedLocalityLevel.put(priority, NodeType.OFF_SWITCH);
      resetSchedulingOpportunities(priority);
    }
  }
  return allowedLocalityLevel.get(priority);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
/**
 * Should be called when the scheduler assigns a container at a higher
 * degree of locality than the current threshold. Reset the allowed locality
 * level to a higher degree of locality.
 */
public synchronized void resetAllowedLocalityLevel(Priority priority,
    NodeType level) {
  NodeType old = allowedLocalityLevel.get(priority);
  LOG.info("Raising locality level from " + old + " to " + level + " at " +
      " priority " + priority);
  allowedLocalityLevel.put(priority, level);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
public CSAssignment(Resource resource, NodeType type) {
  this.resource = resource;
  this.type = type;
  this.application = null;
  this.excessReservation = null;
  this.skipped = false;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; //导入依赖的package包/类
private Resource assignNodeLocalContainers(Resource clusterResource,
    ResourceRequest nodeLocalResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.NODE_LOCAL, 
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        nodeLocalResourceRequest, NodeType.NODE_LOCAL, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}