import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
private synchronized void initScheduler(Configuration conf) {
  validateConf(conf);
  //Use ConcurrentSkipListMap because applications need to be ordered
  this.applications =
      new ConcurrentSkipListMap<ApplicationId, SchedulerApplication<FiCaSchedulerApp>>();
  this.minimumAllocation =
      Resources.createResource(conf.getInt(
          YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB));
  initMaximumResourceCapability(
      Resources.createResource(conf.getInt(
          YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB),
        conf.getInt(
          YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES,
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES),
        conf.getInt(
          YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_GCORES,
          YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_GCORES)));
  this.usePortForNodeName = conf.getBoolean(
      YarnConfiguration.RM_SCHEDULER_INCLUDE_PORT_IN_NODE_NAME,
      YarnConfiguration.DEFAULT_RM_SCHEDULER_USE_PORT_FOR_NODE_NAME);
  this.metrics = QueueMetrics.forQueue(DEFAULT_QUEUE_NAME, null, false,
      conf);
  this.activeUsersManager = new ActiveUsersManager(metrics);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@VisibleForTesting
public synchronized void addApplication(ApplicationId applicationId,
    String queue, String user, boolean isAppRecovering) {
  SchedulerApplication<FiCaSchedulerApp> application =
      new SchedulerApplication<FiCaSchedulerApp>(DEFAULT_QUEUE, user);
  applications.put(applicationId, application);
  metrics.submitApp(user);
  LOG.info("Accepted application " + applicationId + " from user: " + user
      + ", currently num of applications: " + applications.size());
  if (isAppRecovering) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(applicationId + " is recovering. Skip notifying APP_ACCEPTED");
    }
  } else {
    rmContext.getDispatcher().getEventHandler()
      .handle(new RMAppEvent(applicationId, RMAppEventType.APP_ACCEPTED));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
private int assignContainersOnNode(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority 
) {
  // Data-local
  int nodeLocalContainers = 
    assignNodeLocalContainers(node, application, priority); 

  // Rack-local
  int rackLocalContainers = 
    assignRackLocalContainers(node, application, priority);

  // Off-switch
  int offSwitchContainers =
    assignOffSwitchContainers(node, application, priority);


  LOG.debug("assignContainersOnNode:" +
      " node=" + node.getRMNode().getNodeAddress() + 
      " application=" + application.getApplicationId().getId() +
      " priority=" + priority.getPriority() + 
      " #assigned=" + 
      (nodeLocalContainers + rackLocalContainers + offSwitchContainers));


  return (nodeLocalContainers + rackLocalContainers + offSwitchContainers);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的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.common.fica.FiCaSchedulerApp; //导入依赖的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.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void attachContainer(Resource clusterResource,
    FiCaSchedulerApp application, RMContainer rmContainer) {
  if (application != null) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.allocateResource(clusterResource, rmContainer.getContainer()
        .getResource(), node.getLabels());
    LOG.info("movedContainer" + " queueMoveIn=" + getQueueName()
        + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity="
        + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster="
        + clusterResource);
    // Inform the parent
    if (parent != null) {
      parent.attachContainer(clusterResource, application, rmContainer);
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void detachContainer(Resource clusterResource,
    FiCaSchedulerApp application, RMContainer rmContainer) {
  if (application != null) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.releaseResource(clusterResource,
        rmContainer.getContainer().getResource(),
        node.getLabels());
    LOG.info("movedContainer" + " queueMoveOut=" + getQueueName()
        + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity="
        + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster="
        + clusterResource);
    // Inform the parent
    if (parent != null) {
      parent.detachContainer(clusterResource, application, rmContainer);
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
private synchronized void doneApplication(ApplicationId applicationId,
    RMAppState finalState) {
  SchedulerApplication<FiCaSchedulerApp> application =
      applications.get(applicationId);
  if (application == null){
    // The AppRemovedSchedulerEvent maybe sent on recovery for completed apps,
    // ignore it.
    LOG.warn("Couldn't find application " + applicationId);
    return;
  }
  CSQueue queue = (CSQueue) application.getQueue();
  if (!(queue instanceof LeafQueue)) {
    LOG.error("Cannot finish application " + "from non-leaf queue: "
        + queue.getQueueName());
  } else {
    queue.finishApplication(applicationId, application.getUser());
  }
  application.stop(finalState);
  applications.remove(applicationId);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
public LeafQueue(CapacitySchedulerContext cs, 
    String queueName, CSQueue parent, CSQueue old) throws IOException {
  super(cs, queueName, parent, old);
  this.scheduler = cs;

  this.activeUsersManager = new ActiveUsersManager(metrics); 

  if(LOG.isDebugEnabled()) {
    LOG.debug("LeafQueue:" + " name=" + queueName
      + ", fullname=" + getQueuePath());
  }

  Comparator<FiCaSchedulerApp> applicationComparator =
      cs.getApplicationComparator();
  this.pendingApplications = 
      new TreeSet<FiCaSchedulerApp>(applicationComparator);
  this.activeApplications = new TreeSet<FiCaSchedulerApp>(applicationComparator);
  
  setupQueueConfigs(cs.getClusterResource());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void submitApplicationAttempt(FiCaSchedulerApp application,
    String userName) {
  // Careful! Locking order is important!
  synchronized (this) {
    User user = getUser(userName);
    // Add the attempt to our data-structures
    addApplicationAttempt(application, user);
  }

  // We don't want to update metrics for move app
  if (application.isPending()) {
    metrics.submitAppAttempt(userName);
  }
  getParent().submitApplicationAttempt(application, userName);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
private synchronized void addApplicationAttempt(FiCaSchedulerApp application,
    User user) {
  // Accept 
  user.submitApplication();
  pendingApplications.add(application);
  applicationAttemptMap.put(application.getApplicationAttemptId(), application);

  // Activate applications
  activateApplications();
  
  LOG.info("Application added -" +
      " appId: " + application.getApplicationId() +
      " user: " + user + "," + " leaf-queue: " + getQueueName() +
      " #user-pending-applications: " + user.getPendingApplications() +
      " #user-active-applications: " + user.getActiveApplications() +
      " #queue-pending-applications: " + getNumPendingApplications() +
      " #queue-active-applications: " + getNumActiveApplications()
      );
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的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.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void attachContainer(Resource clusterResource,
    FiCaSchedulerApp application, RMContainer rmContainer) {
  if (application != null) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    allocateResource(clusterResource, application, rmContainer.getContainer()
        .getResource(), node.getLabels());
    LOG.info("movedContainer" + " container=" + rmContainer.getContainer()
        + " resource=" + rmContainer.getContainer().getResource()
        + " queueMoveIn=" + this + " usedCapacity=" + getUsedCapacity()
        + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used="
        + queueUsage.getUsed() + " cluster=" + clusterResource);
    // Inform the parent queue
    getParent().attachContainer(clusterResource, application, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void detachContainer(Resource clusterResource,
    FiCaSchedulerApp application, RMContainer rmContainer) {
  if (application != null) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    releaseResource(clusterResource, application, rmContainer.getContainer()
        .getResource(), node.getLabels());
    LOG.info("movedContainer" + " container=" + rmContainer.getContainer()
        + " resource=" + rmContainer.getContainer().getResource()
        + " queueMoveOut=" + this + " usedCapacity=" + getUsedCapacity()
        + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used="
        + queueUsage.getUsed() + " cluster=" + clusterResource);
    // Inform the parent queue
    getParent().detachContainer(clusterResource, application, rmContainer);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Test (timeout = 5000)
public void testApplicationComparator()
{
  CapacityScheduler cs = new CapacityScheduler();
  Comparator<FiCaSchedulerApp> appComparator= cs.getApplicationComparator();
  ApplicationId id1 = ApplicationId.newInstance(1, 1);
  ApplicationId id2 = ApplicationId.newInstance(1, 2);
  ApplicationId id3 = ApplicationId.newInstance(2, 1);
  //same clusterId
  FiCaSchedulerApp app1 = Mockito.mock(FiCaSchedulerApp.class);
  when(app1.getApplicationId()).thenReturn(id1);
  FiCaSchedulerApp app2 = Mockito.mock(FiCaSchedulerApp.class);
  when(app2.getApplicationId()).thenReturn(id2);
  FiCaSchedulerApp app3 = Mockito.mock(FiCaSchedulerApp.class);
  when(app3.getApplicationId()).thenReturn(id3);
  assertTrue(appComparator.compare(app1, app2) < 0);
  //different clusterId
  assertTrue(appComparator.compare(app1, app3) < 0);
  assertTrue(appComparator.compare(app2, app3) < 0);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Test
public void testUserQueueAcl() throws Exception {

  // Manipulate queue 'a'
  LeafQueue d = stubLeafQueue((LeafQueue) queues.get(D));

  // Users
  final String user_d = "user_d";

  // Submit applications
  final ApplicationAttemptId appAttemptId_0 = TestUtils
      .getMockApplicationAttemptId(0, 1);
  FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_d, d, null,
      spyRMContext);
  d.submitApplicationAttempt(app_0, user_d);

  // Attempt the same application again
  final ApplicationAttemptId appAttemptId_1 = TestUtils
      .getMockApplicationAttemptId(0, 2);
  FiCaSchedulerApp app_1 = new FiCaSchedulerApp(appAttemptId_1, user_d, d, null,
      spyRMContext);
  d.submitApplicationAttempt(app_1, user_d); // same user
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
void printString(CSQueue nq, String indent) {
  if (nq instanceof ParentQueue) {
    System.out.println(indent + nq.getQueueName()
        + " cur:" + nq.getAbsoluteUsedCapacity()
        + " guar:" + nq.getAbsoluteCapacity()
        );
    for (CSQueue q : ((ParentQueue)nq).getChildQueues()) {
      printString(q, indent + "  ");
    }
  } else {
    System.out.println(indent + nq.getQueueName()
        + " pen:" + ((LeafQueue) nq).getTotalResourcePending()
        + " cur:" + nq.getAbsoluteUsedCapacity()
        + " guar:" + nq.getAbsoluteCapacity()
        );
    for (FiCaSchedulerApp a : ((LeafQueue)nq).getApplications()) {
      System.out.println(indent + "  " + a.getApplicationId());
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void detachContainer(Resource clusterResource,
    FiCaSchedulerApp application, RMContainer rmContainer) {
  if (application != null) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.releaseResource(clusterResource,
        rmContainer.getContainer().getResource(),
        node.getLabels(),false);
    LOG.info("movedContainer" + " queueMoveOut=" + getQueueName()
        + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity="
        + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster="
        + clusterResource);
    // Inform the parent
    if (parent != null) {
      parent.detachContainer(clusterResource, application, rmContainer);
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void completedContainer(Resource clusterResource,
    FiCaSchedulerApp application, FiCaSchedulerNode node, 
    RMContainer rmContainer, ContainerStatus containerStatus, 
    RMContainerEventType event, CSQueue completedChildQueue,
    boolean sortQueues) {
  if (application != null) {
    internalReleaseResource(clusterResource, node,
        rmContainer.getContainer().getResource(), false, completedChildQueue,
        sortQueues);

    // Inform the parent
    if (parent != null) {
      // complete my parent
      parent.completedContainer(clusterResource, application, 
          node, rmContainer, null, event, this, sortQueues);
    }    
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; //导入依赖的package包/类
@Override
public void detachContainer(Resource clusterResource,
    FiCaSchedulerApp application, RMContainer rmContainer) {
  if (application != null) {
    FiCaSchedulerNode node =
        scheduler.getNode(rmContainer.getContainer().getNodeId());
    super.releaseResource(clusterResource,
        rmContainer.getContainer().getResource(),
        node.getPartition(), false);
    LOG.info("movedContainer" + " queueMoveOut=" + getQueueName()
        + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity="
        + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster="
        + clusterResource);
    // Inform the parent
    if (parent != null) {
      parent.detachContainer(clusterResource, application, rmContainer);
    }
  }
}