import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
/**
 * Utility method to validate a list resource requests, by insuring that the
 * requested memory/vcore is non-negative and not greater than max
 */
public static void normalizeAndValidateRequests(List<ResourceRequest> ask,
    Resource maximumResource, String queueName, YarnScheduler scheduler,
    RMContext rmContext)
    throws InvalidResourceRequestException {

  QueueInfo queueInfo = null;
  try {
    queueInfo = scheduler.getQueueInfo(queueName, false, false);
  } catch (IOException e) {
  }

  for (ResourceRequest resReq : ask) {
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maximumResource,
        queueName, scheduler, rmContext, queueInfo);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getRunningContainers()) {
    completedContainer(container, 
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  updateMaximumAllocation(node, false);
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getRMNode().getTotalCapability());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
@Test
public void testAMCrashAtScheduled() {
  // This is to test sending CONTAINER_FINISHED event at SCHEDULED state.
  // Verify the state transition is correct.
  scheduleApplicationAttempt();
  ContainerStatus cs =
      SchedulerUtils.createAbnormalContainerStatus(
          BuilderUtils.newContainerId(
              applicationAttempt.getAppAttemptId(), 1),
          SchedulerUtils.LOST_CONTAINER);
  // send CONTAINER_FINISHED event at SCHEDULED state,
  // The state should be FINAL_SAVING with previous state SCHEDULED
  NodeId anyNodeId = NodeId.newInstance("host", 1234);
  applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
      applicationAttempt.getAppAttemptId(), cs, anyNodeId));
  // createApplicationAttemptState will return previous state (SCHEDULED),
  // if the current state is FINAL_SAVING.
  assertEquals(YarnApplicationAttemptState.SCHEDULED,
      applicationAttempt.createApplicationAttemptState());
  // send ATTEMPT_UPDATE_SAVED event,
  // verify the state is changed to state FAILED.
  sendAttemptUpdateSavedEvent(applicationAttempt);
  assertEquals(RMAppAttemptState.FAILED,
      applicationAttempt.getAppAttemptState());
  verifyApplicationAttemptFinished(RMAppAttemptState.FAILED);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
/**
 * Utility method to validate a list resource requests, by insuring that the
 * requested memory/vcore is non-negative and not greater than max
 */
public static void normalizeAndValidateRequests(List<ResourceRequest> ask,
    Resource maximumResource, String queueName, YarnScheduler scheduler,
    RMContext rmContext)
    throws InvalidResourceRequestException {
  // Get queue from scheduler
  QueueInfo queueInfo = null;
  try {
    queueInfo = scheduler.getQueueInfo(queueName, false, false);
  } catch (IOException e) {
  }

  for (ResourceRequest resReq : ask) {
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maximumResource,
        queueName, scheduler, rmContext, queueInfo);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getRunningContainers()) {
    completedContainer(container, 
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  updateMaximumAllocation(node, false);
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getTotalResource());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private void handleExcessReservedContainer(Resource clusterResource,
    CSAssignment assignment, FiCaSchedulerNode node, FiCaSchedulerApp app) {
  if (assignment.getExcessReservation() != null) {
    RMContainer excessReservedContainer = assignment.getExcessReservation();
    
    if (excessReservedContainer.hasIncreaseReservation()) {
      unreserveIncreasedContainer(clusterResource,
          app, node, excessReservedContainer);
    } else {
      completedContainer(clusterResource, assignment.getApplication(),
          scheduler.getNode(excessReservedContainer.getAllocatedNode()),
          excessReservedContainer,
          SchedulerUtils.createAbnormalContainerStatus(
              excessReservedContainer.getContainerId(),
              SchedulerUtils.UNRESERVED_CONTAINER),
          RMContainerEventType.RELEASED, null, false);
    }

    assignment.setExcessReservation(null);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getRunningContainers()) {
    containerCompleted(container, 
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getRMNode().getTotalCapability());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private synchronized void removeNode(RMNode nodeInfo) {
  FiCaSchedulerNode node = getNode(nodeInfo.getNodeID());
  if (node == null) {
    return;
  }
  // Kill running containers
  for(RMContainer container : node.getCopiedListOfRunningContainers()) {
    super.completedContainer(container,
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(), 
            SchedulerUtils.LOST_CONTAINER),
            RMContainerEventType.KILL);
  }
  
  //Remove the node
  this.nodes.remove(nodeInfo.getNodeID());
  updateMaximumAllocation(node, false);
  
  // Update cluster metrics
  Resources.subtractFrom(clusterResource, node.getTotalResource());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
/**
 * Check application can be moved to queue with labels enabled. All labels in
 * application life time will be checked
 *
 * @param appId
 * @param dest
 * @throws YarnException
 */
private void checkQueuePartition(FiCaSchedulerApp app, LeafQueue dest)
    throws YarnException {
  if (!YarnConfiguration.areNodeLabelsEnabled(conf)) {
    return;
  }
  Set<String> targetqueuelabels = dest.getAccessibleNodeLabels();
  AppSchedulingInfo schedulingInfo = app.getAppSchedulingInfo();
  Set<String> appLabelexpressions = schedulingInfo.getRequestedPartitions();
  // default partition access always available remove empty label
  appLabelexpressions.remove(RMNodeLabelsManager.NO_LABEL);
  Set<String> nonAccessiblelabels = new HashSet<String>();
  for (String label : appLabelexpressions) {
    if (!SchedulerUtils.checkQueueLabelExpression(targetqueuelabels, label,
        null)) {
      nonAccessiblelabels.add(label);
    }
  }
  if (nonAccessiblelabels.size() > 0) {
    throw new YarnException(
        "Specified queue=" + dest.getQueueName() + " can't satisfy following "
            + "apps label expressions =" + nonAccessiblelabels
            + " accessible node labels =" + targetqueuelabels);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private void killToPreemptContainers(Resource clusterResource,
    FiCaSchedulerNode node,
    CSAssignment assignment) {
  if (assignment.getContainersToKill() != null) {
    StringBuilder sb = new StringBuilder("Killing containers: [");

    for (RMContainer c : assignment.getContainersToKill()) {
      FiCaSchedulerApp application = csContext.getApplicationAttempt(
          c.getApplicationAttemptId());
      LeafQueue q = application.getCSLeafQueue();
      q.completedContainer(clusterResource, application, node, c, SchedulerUtils
              .createPreemptedContainerStatus(c.getContainerId(),
                  SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL,
          null, false);
      sb.append("(container=" + c.getContainerId() + " resource=" + c
          .getAllocatedResource() + ")");
    }

    sb.append("] for container=" + assignment.getAssignmentInformation()
        .getFirstAllocatedOrReservedContainerId() + " resource=" + assignment
        .getResource());
    LOG.info(sb.toString());

  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private void validateResourceRequest(
    ApplicationSubmissionContext submissionContext)
    throws InvalidResourceRequestException {
  // Validation of the ApplicationSubmissionContext needs to be completed
  // here. Only those fields that are dependent on RM's configuration are
  // checked here as they have to be validated whether they are part of new
  // submission or just being recovered.

  // Check whether AM resource requirements are within required limits
  if (!submissionContext.getUnmanagedAM()) {
    ResourceRequest amReq = BuilderUtils.newResourceRequest(
        RMAppAttemptImpl.AM_CONTAINER_PRIORITY, ResourceRequest.ANY,
        submissionContext.getResource(), 1);
    try {
      SchedulerUtils.validateResourceRequest(amReq,
          scheduler.getMaximumResourceCapability());
    } catch (InvalidResourceRequestException e) {
      LOG.warn("RM app submission failed in validating AM resource request"
          + " for application " + submissionContext.getApplicationId(), e);
      throw e;
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private synchronized void doneApplicationAttempt(
    ApplicationAttemptId applicationAttemptId,
    RMAppAttemptState rmAppAttemptFinalState, boolean keepContainers)
    throws IOException {
  FiCaSchedulerApp attempt = getApplicationAttempt(applicationAttemptId);
  SchedulerApplication<FiCaSchedulerApp> application =
      applications.get(applicationAttemptId.getApplicationId());
  if (application == null || attempt == null) {
    throw new IOException("Unknown application " + applicationAttemptId + 
    " has completed!");
  }

  // Kill all 'live' containers
  for (RMContainer container : attempt.getLiveContainers()) {
    if (keepContainers
        && container.getState().equals(RMContainerState.RUNNING)) {
      // do not kill the running container in the case of work-preserving AM
      // restart.
      LOG.info("Skip killing " + container.getContainerId());
      continue;
    }
    completedContainer(container,
      SchedulerUtils.createAbnormalContainerStatus(
        container.getContainerId(), SchedulerUtils.COMPLETED_APPLICATION),
      RMContainerEventType.KILL);
  }

  // Clean up pending requests, metrics etc.
  attempt.stop(rmAppAttemptFinalState);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
protected void warnOrKillContainer(RMContainer container) {
  ApplicationAttemptId appAttemptId = container.getApplicationAttemptId();
  FSAppAttempt app = getSchedulerApp(appAttemptId);
  FSLeafQueue queue = app.getQueue();
  LOG.info("Preempting container (prio=" + container.getContainer().getPriority() +
      "res=" + container.getContainer().getResource() +
      ") from queue " + queue.getName());
  
  Long time = app.getContainerPreemptionTime(container);

  if (time != null) {
    // if we asked for preemption more than maxWaitTimeBeforeKill ms ago,
    // proceed with kill
    if (time + waitTimeBeforeKill < getClock().getTime()) {
      ContainerStatus status =
        SchedulerUtils.createPreemptedContainerStatus(
          container.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER);

      recoverResourceRequestForContainer(container);
      // TODO: Not sure if this ever actually adds this to the list of cleanup
      // containers on the RMNode (see SchedulerNode.releaseContainer()).
      completedContainer(container, status, RMContainerEventType.KILL);
      LOG.info("Killing container" + container +
          " (after waiting for premption for " +
          (getClock().getTime() - time) + "ms)");
    }
  } else {
    // track the request in the FSAppAttempt itself
    app.addPreemption(container, getClock().getTime());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
private synchronized void removeNode(RMNode rmNode) {
  FSSchedulerNode node = getFSSchedulerNode(rmNode.getNodeID());
  // This can occur when an UNHEALTHY node reconnects
  if (node == null) {
    return;
  }
  Resources.subtractFrom(clusterResource, rmNode.getTotalCapability());
  updateRootQueueMetrics();

  // Remove running containers
  List<RMContainer> runningContainers = node.getRunningContainers();
  for (RMContainer container : runningContainers) {
    completedContainer(container,
        SchedulerUtils.createAbnormalContainerStatus(
            container.getContainerId(),
            SchedulerUtils.LOST_CONTAINER),
        RMContainerEventType.KILL);
  }

  // Remove reservations, if any
  RMContainer reservedContainer = node.getReservedContainer();
  if (reservedContainer != null) {
    completedContainer(reservedContainer,
        SchedulerUtils.createAbnormalContainerStatus(
            reservedContainer.getContainerId(),
            SchedulerUtils.LOST_CONTAINER),
        RMContainerEventType.KILL);
  }

  nodes.remove(rmNode.getNodeID());
  queueMgr.getRootQueue().setSteadyFairShare(clusterResource);
  queueMgr.getRootQueue().recomputeSteadyShares();
  updateMaximumAllocation(node, false);
  LOG.info("Removed node " + rmNode.getNodeAddress() +
      " cluster capacity: " + clusterResource);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
public Map<AccessType, AccessControlList> getAcls(String queue) {
  Map<AccessType, AccessControlList> acls =
    new HashMap<AccessType, AccessControlList>();
  for (QueueACL acl : QueueACL.values()) {
    acls.put(SchedulerUtils.toAccessType(acl), getAcl(queue, acl));
  }
  return acls;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
@Override
public void dropContainerReservation(RMContainer container) {
  if(LOG.isDebugEnabled()){
    LOG.debug("DROP_RESERVATION:" + container.toString());
  }
  completedContainer(container,
      SchedulerUtils.createAbnormalContainerStatus(
          container.getContainerId(),
          SchedulerUtils.UNRESERVED_CONTAINER),
      RMContainerEventType.KILL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
@Override
public void killContainer(RMContainer cont) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("KILL_CONTAINER: container" + cont.toString());
  }
  recoverResourceRequestForContainer(cont);
  completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus(
    cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER),
    RMContainerEventType.KILL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
protected void warnOrKillContainer(RMContainer container) {
  ApplicationAttemptId appAttemptId = container.getApplicationAttemptId();
  FSAppAttempt app = getSchedulerApp(appAttemptId);
  FSLeafQueue queue = app.getQueue();
  LOG.info("Preempting container (prio=" + container.getContainer().getPriority() +
      "res=" + container.getContainer().getResource() +
      ") from queue " + queue.getName());
  
  Long time = app.getContainerPreemptionTime(container);

  if (time != null) {
    // if we asked for preemption more than maxWaitTimeBeforeKill ms ago,
    // proceed with kill
    if (time + waitTimeBeforeKill < getClock().getTime()) {
      ContainerStatus status =
        SchedulerUtils.createPreemptedContainerStatus(
          container.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER);

      // TODO: Not sure if this ever actually adds this to the list of cleanup
      // containers on the RMNode (see SchedulerNode.releaseContainer()).
      completedContainer(container, status, RMContainerEventType.KILL);
      LOG.info("Killing container" + container +
          " (after waiting for preemption for " +
          (getClock().getTime() - time) + "ms)");
    }
  } else {
    // track the request in the FSAppAttempt itself
    app.addPreemption(container, getClock().getTime());
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
@Override
public void killContainer(RMContainer cont) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("KILL_CONTAINER: container" + cont.toString());
  }
  completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus(
    cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER),
    RMContainerEventType.KILL);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; //导入依赖的package包/类
@Override
public void killContainer(RMContainer cont) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("KILL_CONTAINER: container" + cont.toString());
  }
  recoverResourceRequestForContainer(cont);
  completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus(
  	      cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER),
  	      RMContainerEventType.KILL);
}