import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public synchronized void containerLaunchedOnNode(ContainerId containerId,
    NodeId nodeId) {
  // Inform the container
  RMContainer rmContainer = 
      getRMContainer(containerId);
  if (rmContainer == null) {
    // Some unknown container sneaked into the system. Kill it.
    rmContext.getDispatcher().getEventHandler()
      .handle(new RMNodeCleanContainerEvent(nodeId, containerId));
    return;
  }

  rmContainer.handle(new RMContainerEvent(containerId,
    RMContainerEventType.LAUNCHED));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public synchronized void containerLaunchedOnNode(ContainerId containerId,
    NodeId nodeId) {
  // Inform the container
  RMContainer rmContainer = getRMContainer(containerId);
  if (rmContainer == null) {
    // Some unknown container sneaked into the system. Kill it.
    rmContext.getDispatcher().getEventHandler()
      .handle(new RMNodeCleanContainerEvent(nodeId, containerId));
    return;
  }

  rmContainer.handle(new RMContainerEvent(containerId,
      RMContainerEventType.LAUNCHED));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
public synchronized ContainersAndNMTokensAllocation
    pullNewlyAllocatedContainersAndNMTokens() {
  List<Container> returnContainerList =
      new ArrayList<Container>(newlyAllocatedContainers.size());
  List<NMToken> nmTokens = new ArrayList<NMToken>();
  for (Iterator<RMContainer> i = newlyAllocatedContainers.iterator(); i
    .hasNext();) {
    RMContainer rmContainer = i.next();
    Container container = rmContainer.getContainer();
    try {
      // create container token and NMToken altogether.
      container.setContainerToken(rmContext.getContainerTokenSecretManager()
        .createContainerToken(container.getId(), container.getNodeId(),
          getUser(), container.getResource(), container.getPriority(),
          rmContainer.getCreationTime(), this.logAggregationContext));
      NMToken nmToken =
          rmContext.getNMTokenSecretManager().createAndGetNMToken(getUser(),
            getApplicationAttemptId(), container);
      if (nmToken != null) {
        nmTokens.add(nmToken);
      }
    } catch (IllegalArgumentException e) {
      // DNS might be down, skip returning this container.
      LOG.error("Error trying to assign container token and NM token to" +
          " an allocated container " + container.getId(), e);
      continue;
    }
    returnContainerList.add(container);
    i.remove();
    rmContainer.handle(new RMContainerEvent(rmContainer.getContainerId(),
      RMContainerEventType.ACQUIRED));
  }
  return new ContainersAndNMTokensAllocation(returnContainerList, nmTokens);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
private Container updateContainerAndNMToken(RMContainer rmContainer,
    boolean newContainer, boolean increasedContainer) {
  Container container = rmContainer.getContainer();
  ContainerType containerType = ContainerType.TASK;
  // The working knowledge is that masterContainer for AM is null as it
  // itself is the master container.
  if (isWaitingForAMContainer()) {
    containerType = ContainerType.APPLICATION_MASTER;
  }
  try {
    // create container token and NMToken altogether.
    container.setContainerToken(rmContext.getContainerTokenSecretManager()
        .createContainerToken(container.getId(), container.getNodeId(),
            getUser(), container.getResource(), container.getPriority(),
            rmContainer.getCreationTime(), this.logAggregationContext,
            rmContainer.getNodeLabelExpression(), containerType));
    NMToken nmToken =
        rmContext.getNMTokenSecretManager().createAndGetNMToken(getUser(),
            getApplicationAttemptId(), container);
    if (nmToken != null) {
      updatedNMTokens.add(nmToken);
    }
  } catch (IllegalArgumentException e) {
    // DNS might be down, skip returning this container.
    LOG.error("Error trying to assign container token and NM token to"
        + " an updated container " + container.getId(), e);
    return null;
  }
  
  if (newContainer) {
    rmContainer.handle(new RMContainerEvent(
        rmContainer.getContainerId(), RMContainerEventType.ACQUIRED));
  } else {
    rmContainer.handle(new RMContainerUpdatesAcquiredEvent(
        rmContainer.getContainerId(), increasedContainer));
  }
  return container;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
private void sentRMContainerLaunched(MockRM rm, ContainerId containerId) {
  CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
  RMContainer rmContainer = cs.getRMContainer(containerId);
  if (rmContainer != null) {
    rmContainer.handle(
        new RMContainerEvent(containerId, RMContainerEventType.LAUNCHED));
  } else {
    Assert.fail("Cannot find RMContainer");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public List<Container> pullNewlyAllocatedContainers() {
  List<Container> returnContainerList = new ArrayList<Container>(
      newlyAllocatedContainers.size());
  for (RMContainer rmContainer : newlyAllocatedContainers) {
    rmContainer.handle(new RMContainerEvent(rmContainer.getContainerId(),
        RMContainerEventType.ACQUIRED));
    returnContainerList.add(rmContainer.getContainer());
  }
  newlyAllocatedContainers.clear();
  return returnContainerList;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
public synchronized void containerLaunchedOnNode(ContainerId containerId,
    NodeId nodeId) {
  // Inform the container
  RMContainer rmContainer = 
      getRMContainer(containerId);
  if (rmContainer == null) {
    // Some unknown container sneaked into the system. Kill it.
    this.rmContext.getDispatcher().getEventHandler()
      .handle(new RMNodeCleanContainerEvent(nodeId, containerId));
    return;
  }

  rmContainer.handle(new RMContainerEvent(containerId,
    RMContainerEventType.LAUNCHED));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
public synchronized ContainersAndNMTokensAllocation
    pullNewlyAllocatedContainersAndNMTokens() {
  List<Container> returnContainerList =
      new ArrayList<Container>(newlyAllocatedContainers.size());
  List<NMToken> nmTokens = new ArrayList<NMToken>();
  for (Iterator<RMContainer> i = newlyAllocatedContainers.iterator(); i
    .hasNext();) {
    RMContainer rmContainer = i.next();
    Container container = rmContainer.getContainer();
    try {
      // create container token and NMToken altogether.
      container.setContainerToken(rmContext.getContainerTokenSecretManager()
        .createContainerToken(container.getId(), container.getNodeId(),
          getUser(), container.getResource()));
      NMToken nmToken =
          rmContext.getNMTokenSecretManager().createAndGetNMToken(getUser(),
            getApplicationAttemptId(), container);
      if (nmToken != null) {
        nmTokens.add(nmToken);
      }
    } catch (IllegalArgumentException e) {
      // DNS might be down, skip returning this container.
      LOG.error("Error trying to assign container token and NM token to" +
          " an allocated container " + container.getId(), e);
      continue;
    }
    returnContainerList.add(container);
    i.remove();
    rmContainer.handle(new RMContainerEvent(rmContainer.getContainerId(),
      RMContainerEventType.ACQUIRED));
  }
  return new ContainersAndNMTokensAllocation(returnContainerList, nmTokens);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public RMContainer allocate(NodeType type, FSSchedulerNode node,
    Priority priority, ResourceRequest request,
    Container container) {
  // Update allowed locality level
  NodeType allowed = allowedLocalityLevel.get(priority);
  if (allowed != null) {
    if (allowed.equals(NodeType.OFF_SWITCH) &&
        (type.equals(NodeType.NODE_LOCAL) ||
            type.equals(NodeType.RACK_LOCAL))) {
      this.resetAllowedLocalityLevel(priority, type);
    }
    else if (allowed.equals(NodeType.RACK_LOCAL) &&
        type.equals(NodeType.NODE_LOCAL)) {
      this.resetAllowedLocalityLevel(priority, type);
    }
  }

  // Required sanity check - AM can call 'allocate' to update resource 
  // request without locking the scheduler, hence we need to check
  if (getTotalRequiredResources(priority) <= 0) {
    return null;
  }
  
  // Create RMContainer
  RMContainer rmContainer = new RMContainerImpl(container, 
      getApplicationAttemptId(), node.getNodeID(),
      appSchedulingInfo.getUser(), rmContext);

  // Add it to allContainers list.
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations
  List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
      type, node, priority, request, container);
  this.attemptResourceUsage.incUsed(container.getResource());

  // Update resource requests related to "request" and store in RMContainer
  ((RMContainerImpl) rmContainer).setResourceRequests(resourceRequestList);

  // Inform the container
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));

  if (LOG.isDebugEnabled()) {
    LOG.debug("allocate: applicationAttemptId=" 
        + container.getId().getApplicationAttemptId() 
        + " container=" + container.getId() + " host="
        + container.getNodeId().getHost() + " type=" + type);
  }
  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.ALLOC_CONTAINER, "SchedulerApp", 
      getApplicationId(), container.getId());
  
  return rmContainer;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public RMContainer allocate(NodeType type, FiCaSchedulerNode node,
    Priority priority, ResourceRequest request, 
    Container container) {

  if (isStopped) {
    return null;
  }
  
  // Required sanity check - AM can call 'allocate' to update resource 
  // request without locking the scheduler, hence we need to check
  if (getTotalRequiredResources(priority) <= 0) {
    return null;
  }
  
  // Create RMContainer
  RMContainer rmContainer = new RMContainerImpl(container, this
      .getApplicationAttemptId(), node.getNodeID(),
      appSchedulingInfo.getUser(), this.rmContext);

  // Add it to allContainers list.
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations
  List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
      type, node, priority, request, container);

  attemptResourceUsage.incUsed(node.getPartition(), container.getResource());

  // Update resource requests related to "request" and store in RMContainer
  ((RMContainerImpl)rmContainer).setResourceRequests(resourceRequestList);

  // Inform the container
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));

  if (LOG.isDebugEnabled()) {
    LOG.debug("allocate: applicationAttemptId=" 
        + container.getId().getApplicationAttemptId() 
        + " container=" + container.getId() + " host="
        + container.getNodeId().getHost() + " type=" + type);
  }
  RMAuditLogger.logSuccess(getUser(),
      AuditConstants.ALLOC_CONTAINER, "SchedulerApp",
      getApplicationId(), container.getId());
  
  return rmContainer;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
protected synchronized void containerIncreasedOnNode(ContainerId containerId,
    SchedulerNode node, Container increasedContainerReportedByNM) {
  // Get the application for the finished container
  SchedulerApplicationAttempt application =
      getCurrentAttemptForContainer(containerId);
  if (application == null) {
    LOG.info("Unknown application "
        + containerId.getApplicationAttemptId().getApplicationId()
        + " increased container " + containerId + " on node: " + node);
    this.rmContext.getDispatcher().getEventHandler()
        .handle(new RMNodeCleanContainerEvent(node.getNodeID(), containerId));
    return;
  }

  RMContainer rmContainer = getRMContainer(containerId);
  Resource rmContainerResource = rmContainer.getAllocatedResource();
  Resource nmContainerResource = increasedContainerReportedByNM.getResource();
  
  
  if (Resources.equals(nmContainerResource, rmContainerResource)){
    // NM reported expected container size, tell RMContainer. Which will stop
    // container expire monitor
    rmContainer.handle(new RMContainerEvent(containerId,
        RMContainerEventType.NM_DONE_CHANGE_RESOURCE));
  } else if (Resources.fitsIn(getResourceCalculator(), clusterResource,
      nmContainerResource, rmContainerResource)) {
    // when rmContainerResource >= nmContainerResource, we won't do anything,
    // it is possible a container increased is issued by RM, but AM hasn't
    // told NM.
  } else if (Resources.fitsIn(getResourceCalculator(), clusterResource,
      rmContainerResource, nmContainerResource)) {
    // When rmContainerResource <= nmContainerResource, it could happen when a
    // container decreased by RM before it is increased in NM.
    
    // Tell NM to decrease the container
    this.rmContext.getDispatcher().getEventHandler()
        .handle(new RMNodeDecreaseContainerEvent(node.getNodeID(),
            Arrays.asList(rmContainer.getContainer())));
  } else {
    // Something wrong happened, kill the container
    LOG.warn("Something wrong happened, container size reported by NM"
        + " is not expected, ContainerID=" + containerId
        + " rm-size-resource:" + rmContainerResource + " nm-size-reosurce:"
        + nmContainerResource);
    this.rmContext.getDispatcher().getEventHandler()
        .handle(new RMNodeCleanContainerEvent(node.getNodeID(), containerId));
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public RMContainer allocate(NodeType type, FiCaSchedulerNode node,
    Priority priority, ResourceRequest request, 
    Container container) {

  if (isStopped) {
    return null;
  }
  
  // Required sanity check - AM can call 'allocate' to update resource 
  // request without locking the scheduler, hence we need to check
  if (getTotalRequiredResources(priority) <= 0) {
    return null;
  }
  
  // Create RMContainer
  RMContainer rmContainer =
      new RMContainerImpl(container, this.getApplicationAttemptId(),
          node.getNodeID(), appSchedulingInfo.getUser(), this.rmContext,
          request.getNodeLabelExpression());

  updateAMContainerDiagnostics(AMState.ASSIGNED, null);

  // Add it to allContainers list.
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations
  List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
      type, node, priority, request, container);

  attemptResourceUsage.incUsed(node.getPartition(), container.getResource());

  // Update resource requests related to "request" and store in RMContainer
  ((RMContainerImpl)rmContainer).setResourceRequests(resourceRequestList);

  // Inform the container
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));

  if (LOG.isDebugEnabled()) {
    LOG.debug("allocate: applicationAttemptId=" 
        + container.getId().getApplicationAttemptId() 
        + " container=" + container.getId() + " host="
        + container.getNodeId().getHost() + " type=" + type);
  }
  RMAuditLogger.logSuccess(getUser(),
      AuditConstants.ALLOC_CONTAINER, "SchedulerApp",
      getApplicationId(), container.getId());
  
  return rmContainer;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public RMContainer allocate(NodeType type, FSSchedulerNode node,
    Priority priority, ResourceRequest request,
    Container container) {
  // Update allowed locality level
  NodeType allowed = allowedLocalityLevel.get(priority);
  if (allowed != null) {
    if (allowed.equals(NodeType.OFF_SWITCH) &&
        (type.equals(NodeType.NODE_LOCAL) ||
            type.equals(NodeType.RACK_LOCAL))) {
      this.resetAllowedLocalityLevel(priority, type);
    }
    else if (allowed.equals(NodeType.RACK_LOCAL) &&
        type.equals(NodeType.NODE_LOCAL)) {
      this.resetAllowedLocalityLevel(priority, type);
    }
  }

  // Required sanity check - AM can call 'allocate' to update resource 
  // request without locking the scheduler, hence we need to check
  if (getTotalRequiredResources(priority) <= 0) {
    return null;
  }
  
  // Create RMContainer
  RMContainer rmContainer = new RMContainerImpl(container, 
      getApplicationAttemptId(), node.getNodeID(),
      appSchedulingInfo.getUser(), rmContext);

  // Add it to allContainers list.
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations
  List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
      type, node, priority, request, container);
  Resources.addTo(currentConsumption, container.getResource());

  // Update resource requests related to "request" and store in RMContainer
  ((RMContainerImpl) rmContainer).setResourceRequests(resourceRequestList);

  // Inform the container
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));

  if (LOG.isDebugEnabled()) {
    LOG.debug("allocate: applicationAttemptId=" 
        + container.getId().getApplicationAttemptId() 
        + " container=" + container.getId() + " host="
        + container.getNodeId().getHost() + " type=" + type);
  }
  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.ALLOC_CONTAINER, "SchedulerApp", 
      getApplicationId(), container.getId());
  
  return rmContainer;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public boolean containerResume(RMContainer rmContainer,Resource toResume){
 
 
 
 ContainerId containerId = rmContainer.getContainerId();
 
 if(isStopped){
  return false;
 }
 
 if(!isSuspending){
  return false;
 }
 
 if(!containersSuspended.contains(containerId)){
  return false;
 }
 //add resumed resource
 rmContainer.addResumedResource(toResume);
 
 //we try to update its resource consumption
 rmContainer.handle(
            new RMContainerEvent(containerId,RMContainerEventType.RESUME)  
     );

 //if all of its resource has been resumed
 if(!rmContainer.isSuspending()){
 //delete contaienr from containersSuspended
 this.containersSuspended.remove(containerId);
 }  
 
 //update resource usage
 queue.getMetrics().allocateResources(getUser(), 1, toResume, true);
 //update app resource usage
 Resources.addTo(currentConsumption,toResume);
 //inform RMContainer
 if(this.containersSuspended.size() == 0){  
  isSuspending = false;
  LOG.info("application "+this.getApplicationId()+"has been out of the suspended list");
 }
 
 LOG.info("app "+this.getApplicationAttemptId()+" consume resource"+currentConsumption);

   if (LOG.isDebugEnabled()) {
     LOG.debug("allocate: applicationAttemptId=" 
         + this.getApplicationAttemptId() 
         + " container=" + containerId + " host="
         + rmContainer.getContainer().getNodeId().getHost() 
       );
   }
 
 return true;
 
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public RMContainer allocate(NodeType type, FiCaSchedulerNode node,
    Priority priority, ResourceRequest request, 
    Container container) {

  if (isStopped) {
    return null;
  }
  
  // Required sanity check - AM can call 'allocate' to update resource 
  // request without locking the scheduler, hence we need to check
  if (getTotalRequiredResources(priority) <= 0) {
    return null;
  }
  
  // Create RMContainer
  RMContainer rmContainer = new RMContainerImpl(container, this
      .getApplicationAttemptId(), node.getNodeID(),
      appSchedulingInfo.getUser(), this.rmContext);

  // Add it to allContainers list.
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations
  List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
      type, node, priority, request, container);
  Resources.addTo(currentConsumption, container.getResource());
  
  // Update resource requests related to "request" and store in RMContainer 
  ((RMContainerImpl)rmContainer).setResourceRequests(resourceRequestList);

  // Inform the container
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));

  if (LOG.isDebugEnabled()) {
    LOG.debug("allocate: applicationAttemptId=" 
        + container.getId().getApplicationAttemptId() 
        + " container=" + container.getId() + " host="
        + container.getNodeId().getHost() + " type=" + type);
  }
  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.ALLOC_CONTAINER, "SchedulerApp", 
      getApplicationId(), container.getId());
  
  return rmContainer;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent; //导入依赖的package包/类
synchronized public RMContainer allocate(NodeType type, FSSchedulerNode node,
    Priority priority, ResourceRequest request,
    Container container) {
  // Update allowed locality level
  NodeType allowed = allowedLocalityLevel.get(priority);
  if (allowed != null) {
    if (allowed.equals(NodeType.OFF_SWITCH) &&
        (type.equals(NodeType.NODE_LOCAL) ||
            type.equals(NodeType.RACK_LOCAL))) {
      this.resetAllowedLocalityLevel(priority, type);
    }
    else if (allowed.equals(NodeType.RACK_LOCAL) &&
        type.equals(NodeType.NODE_LOCAL)) {
      this.resetAllowedLocalityLevel(priority, type);
    }
  }

  // Required sanity check - AM can call 'allocate' to update resource 
  // request without locking the scheduler, hence we need to check
  if (getTotalRequiredResources(priority) <= 0) {
    return null;
  }
  
  // Create RMContainer
  RMContainer rmContainer = new RMContainerImpl(container, 
      getApplicationAttemptId(), node.getNodeID(), rmContext
      .getDispatcher().getEventHandler(), rmContext
      .getContainerAllocationExpirer());

  // Add it to allContainers list.
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations

  LOG.info("TO allocate in FSSchedulerApp: type: " + type + ", node " + node + ", resource: "
          + request.getCapability());
  appSchedulingInfo.allocate(type, node, priority, request, container);
  Resources.addTo(currentConsumption, container.getResource());

  // Inform the container
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));

  if (LOG.isDebugEnabled()) {
    LOG.debug("allocate: applicationAttemptId=" 
        + container.getId().getApplicationAttemptId() 
        + " container=" + container.getId() + " host="
        + container.getNodeId().getHost() + " type=" + type);
  }
  RMAuditLogger.logSuccess(getUser(), 
      AuditConstants.ALLOC_CONTAINER, "SchedulerApp", 
      getApplicationId(), container.getId());
  
  return rmContainer;
}