import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
public static AllocateResponse newAllocateResponse(int responseId,
    List<ContainerStatus> completedContainers,
    List<Container> allocatedContainers, List<NodeReport> updatedNodes,
    Resource availResources, AMCommand command, int numClusterNodes,
    PreemptionMessage preempt) {
  AllocateResponse response = recordFactory
      .newRecordInstance(AllocateResponse.class);
  response.setNumClusterNodes(numClusterNodes);
  response.setResponseId(responseId);
  response.setCompletedContainersStatuses(completedContainers);
  response.setAllocatedContainers(allocatedContainers);
  response.setUpdatedNodes(updatedNodes);
  response.setAvailableResources(availResources);
  response.setAMCommand(command);
  response.setPreemptionMessage(preempt);

  return response;
}
 

import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
@Public
@Stable
public static AllocateResponse newInstance(int responseId,
    List<ContainerStatus> completedContainers,
    List<Container> allocatedContainers, List<NodeReport> updatedNodes,
    Resource availResources, AMCommand command, int numClusterNodes,
    PreemptionMessage preempt, List<NMToken> nmTokens) {
  AllocateResponse response = Records.newRecord(AllocateResponse.class);
  response.setNumClusterNodes(numClusterNodes);
  response.setResponseId(responseId);
  response.setCompletedContainersStatuses(completedContainers);
  response.setAllocatedContainers(allocatedContainers);
  response.setUpdatedNodes(updatedNodes);
  response.setAvailableResources(availResources);
  response.setAMCommand(command);
  response.setPreemptionMessage(preempt);
  response.setNMTokens(nmTokens);
  return response;
}
 

import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
private PreemptionMessage getPreemptionMessage(boolean strictContract,
    boolean contract, final ContainerId container) {
  PreemptionMessage preemptionMessage = recordFactory
      .newRecordInstance(PreemptionMessage.class);
  Set<PreemptionContainer> cntrs = new HashSet<PreemptionContainer>();
  PreemptionContainer preemptContainer = recordFactory
      .newRecordInstance(PreemptionContainer.class);
  preemptContainer.setId(container);
  cntrs.add(preemptContainer);
  if (strictContract) {
    StrictPreemptionContract set = recordFactory
        .newRecordInstance(StrictPreemptionContract.class);
    set.setContainers(cntrs);
    preemptionMessage.setStrictContract(set);
  }
  if (contract) {
    PreemptionContract preemptContract = recordFactory
        .newRecordInstance(PreemptionContract.class);
    preemptContract.setContainers(cntrs);
    preemptionMessage.setContract(preemptContract);
  }
  return preemptionMessage;
}
 

import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
private List<TaskAttemptId> validatePreemption(PreemptionMessage pM,
  CheckpointAMPreemptionPolicy policy, int supposedMemPreemption) {
  Resource effectivelyPreempted = Resource.newInstance(0, 0);

  List<TaskAttemptId> preempting = new ArrayList<TaskAttemptId>();

  for (Map.Entry<ContainerId, TaskAttemptId> ent :
      assignedContainers.entrySet()) {
    if (policy.isPreempted(ent.getValue())) {
      Resources.addTo(effectivelyPreempted,contToResourceMap.get(ent.getKey()));
      // preempt only reducers
      if (policy.isPreempted(ent.getValue())){
        assertEquals(TaskType.REDUCE, ent.getValue().getTaskId().getTaskType());
        preempting.add(ent.getValue());
      }
    }
  }

  // preempt enough
  assert (effectivelyPreempted.getMemory() >= supposedMemPreemption)
    : " preempted: " + effectivelyPreempted.getMemory();

  // preempt not too much enough
  assert effectivelyPreempted.getMemory() <= supposedMemPreemption + minAlloc;
  return preempting;
}
 

import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
private PreemptionMessage generatePreemptionMessage(
    Set<ContainerId> containerToPreempt,
    HashMap<ContainerId, Resource> resPerCont,
    Resource minimumAllocation, boolean strict) {

  Set<ContainerId> currentContPreemption = Collections.unmodifiableSet(
      new HashSet<ContainerId>(containerToPreempt));
  containerToPreempt.clear();
  Resource tot = Resource.newInstance(0, 0);
  for(ContainerId c : currentContPreemption){
    Resources.addTo(tot,
        resPerCont.get(c));
  }
  int numCont = (int) Math.ceil(tot.getMemory() /
            (double) minimumAllocation.getMemory());
  ResourceRequest rr = ResourceRequest.newInstance(
      Priority.newInstance(0), ResourceRequest.ANY,
      minimumAllocation, numCont);
  if (strict) {
    return generatePreemptionMessage(new Allocation(null, null,
                currentContPreemption, null, null));
  }
  return generatePreemptionMessage(new Allocation(null, null,
                        null, currentContPreemption,
                        Collections.singletonList(rr)));
}
 

import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
@Override
public synchronized PreemptionMessage getPreemptionMessage() {
  AllocateResponseProtoOrBuilder p = viaProto ? proto : builder;
  if (this.preempt != null) {
    return this.preempt;
  }
  if (!p.hasPreempt()) {
    return null;
  }
  this.preempt = convertFromProtoFormat(p.getPreempt());
  return this.preempt;
}
 

import org.apache.hadoop.yarn.api.records.PreemptionMessage; //导入依赖的package包/类
@Override
public synchronized void setPreemptionMessage(PreemptionMessage preempt) {
  maybeInitBuilder();
  if (null == preempt) {
    builder.clearPreempt();
  }
  this.preempt = preempt;
}