import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private RunningContainers createRunningContainers(ContainerId appMasterContainerId,
                                                  String appMasterHost) throws Exception {
  int containerMemoryMB = Integer.parseInt(System.getenv(EnvKeys.YARN_CONTAINER_MEMORY_MB));

  // We can't get the -Xmx easily, so just recompute the -Xmx in the same way that the client does
  int maxHeapMemoryMB = Resources.computeMaxHeapSize(containerMemoryMB,
                                                     twillRuntimeSpec.getAMReservedMemory(),
                                                     twillRuntimeSpec.getAMMinHeapRatio());
  TwillRunResources appMasterResources = new DefaultTwillRunResources(
    0,
    appMasterContainerId.toString(),
    Integer.parseInt(System.getenv(EnvKeys.YARN_CONTAINER_VIRTUAL_CORES)),
    containerMemoryMB,
    maxHeapMemoryMB,
    appMasterHost, null);
  String appId = appMasterContainerId.getApplicationAttemptId().getApplicationId().toString();
  return new RunningContainers(twillRuntimeSpec, appId, appMasterResources, zkClient, applicationLocation,
                               twillSpec.getRunnables(), eventHandler);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
RunningContainers(TwillRuntimeSpecification twillRuntimeSpec, String appId, TwillRunResources appMasterResources,
                  ZKClient zookeeperClient, Location applicationLocation,
                  Map<String, RuntimeSpecification> runnables,
                  EventHandler eventHandler) {
  containers = HashBasedTable.create();
  runnableInstances = Maps.newHashMap();
  completedContainerCount = Maps.newHashMap();
  startSequence = Lists.newLinkedList();
  containerLock = new ReentrantLock();
  containerChange = containerLock.newCondition();
  resourceReport = new DefaultResourceReport(appId, appMasterResources);
  zkClient = zookeeperClient;
  containerStats = HashMultimap.create();
  this.applicationLocation = applicationLocation;
  this.runnableNames = runnables.keySet();
  this.logLevels = new TreeMap<>();
  this.maxRetries = Maps.newHashMap(twillRuntimeSpec.getMaxRetries());
  this.numRetries = Maps.newHashMap();
  this.eventHandler = eventHandler;
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private boolean waitForDebugPort(TwillController controller, String runnable, int timeLimit)
  throws InterruptedException {
  long millis = 0;
  while (millis < 1000 * timeLimit) {
    ResourceReport report = controller.getResourceReport();
    if (report == null || report.getRunnableResources(runnable) == null) {
      continue;
    }
    for (TwillRunResources resources : report.getRunnableResources(runnable)) {
      if (resources.getDebugPort() != null) {
        return true;
      }
    }
    TimeUnit.MILLISECONDS.sleep(100);
    millis += 100;
  }
  return false;
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private boolean waitForLogLevel(TwillController controller, String runnable, long timeout,
                                TimeUnit timeoutUnit, @Nullable LogEntry.Level expected) throws InterruptedException {

  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  do {
    ResourceReport report = controller.getResourceReport();
    if (report == null || report.getRunnableResources(runnable) == null) {
      continue;
    }
    for (TwillRunResources resources : report.getRunnableResources(runnable)) {
      LogEntry.Level level = resources.getLogLevels().get(Logger.ROOT_LOGGER_NAME);
      if (expected == level) {
         return true;
      }
    }
    TimeUnit.MILLISECONDS.sleep(100);
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  return false;
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private void waitForInstance(TwillController controller, String runnable, String yarnInstanceId,
                             long timeout, TimeUnit timeoutUnit) throws InterruptedException, TimeoutException {
  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  do {
    ResourceReport report = controller.getResourceReport();
    if (report != null && report.getRunnableResources(runnable) != null) {
      for (TwillRunResources resources : report.getRunnableResources(runnable)) {
        if (resources.getContainerId().endsWith(yarnInstanceId)) {
          return;
        }
      }
    }
    TimeUnit.SECONDS.sleep(1);
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  throw new TimeoutException("Timeout reached while waiting for runnable " +
                               runnable + " instance " + yarnInstanceId);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private void validateInstanceIds(String runnable, Set<Integer> instanceIds) {
  ResourceReport resourceReport = getResourceReport();
  if (resourceReport == null) {
    throw new IllegalStateException("Unable to get resource report since application has not started.");
  }
  Collection<TwillRunResources> runnableResources = resourceReport.getRunnableResources(runnable);
  if (runnableResources == null) {
    throw new RuntimeException("Unable to verify run resources for runnable " + runnable);
  }
  Set<Integer> existingInstanceIds = Sets.newHashSet();
  for (TwillRunResources twillRunResources : runnableResources) {
    existingInstanceIds.add(twillRunResources.getInstanceId());
  }
  LOG.info("Existing instance ids: {}", existingInstanceIds);
  for (int instanceId : instanceIds) {
    if (!existingInstanceIds.contains(instanceId)) {
      throw new IllegalArgumentException("Unable to find instance id " + instanceId + " for " + runnable);
    }
  }
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
@Override
public JsonElement serialize(TwillRunResources src, Type typeOfSrc, JsonSerializationContext context) {
  JsonObject json = new JsonObject();

  json.addProperty(CONTAINER_ID, src.getContainerId());
  json.addProperty(INSTANCE_ID, src.getInstanceId());
  json.addProperty(HOST, src.getHost());
  json.addProperty(MEMORY_MB, src.getMemoryMB());
  json.addProperty(MAX_HEAP_MEMORY_MB, src.getMaxHeapMemoryMB());
  json.addProperty(VIRTUAL_CORES, src.getVirtualCores());
  if (src.getDebugPort() != null) {
    json.addProperty(DEBUG_PORT, src.getDebugPort());
  }
  json.add(LOG_LEVELS, context.serialize(src.getLogLevels(),
                                         new TypeToken<Map<String, LogEntry.Level>>() { }.getType()));

  return json;
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
@Override
public TwillRunResources deserialize(JsonElement json, Type typeOfT,
                                         JsonDeserializationContext context) throws JsonParseException {
  JsonObject jsonObj = json.getAsJsonObject();
  Map<String, LogEntry.Level> logLevels =
    context.deserialize(jsonObj.get(LOG_LEVELS), new TypeToken<Map<String, LogEntry.Level>>() { }.getType());
  int memoryMB = jsonObj.get(MEMORY_MB).getAsInt();
  return new DefaultTwillRunResources(
    jsonObj.get(INSTANCE_ID).getAsInt(),
    jsonObj.get(CONTAINER_ID).getAsString(),
    jsonObj.get(VIRTUAL_CORES).getAsInt(),
    memoryMB,
    // For backward compatibility when a newer Twill client re-attached to running app started with older version.
    jsonObj.has(MAX_HEAP_MEMORY_MB) ? jsonObj.get(MAX_HEAP_MEMORY_MB).getAsInt() : memoryMB,
    jsonObj.get(HOST).getAsString(),
    jsonObj.has(DEBUG_PORT) ? jsonObj.get(DEBUG_PORT).getAsInt() : null,
    logLevels);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
/**
 * Checks to see if the report has loaded.
 * @param report - The {@link ResourceReport} for this Twill Application.
 * @return Return true if the report is null or incomplete.  Return false if the report is completely loaded.
 */
private boolean reportIsLoading(@Nullable final ResourceReport report) {
    if(report == null) {
        return true;
    }

    final String yarnApplicationID = report.getApplicationId();
    final Collection<TwillRunResources> runnableResources = report.getResources().get(PeriodicNotificationTwillRunnable.TWILL_RUNNABLE_NAME);

    if(runnableResources == null || runnableResources.isEmpty()) {
        LOG.info("Received Resource Report for YARN ApplicationID: {}, runnable resources are still loading...", yarnApplicationID);
        return true;
    } else {
        LOG.info("Received Resource Report for YARN ApplicationID: {}, runnable resources are loaded.", yarnApplicationID);
        return false;
    }
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
void start(String runnableName, ContainerInfo containerInfo, TwillContainerLauncher launcher) {
  containerLock.lock();
  try {
    int instanceId = getStartInstanceId(runnableName);
    RunId runId = getRunId(runnableName, instanceId);
    TwillContainerController controller = launcher.start(runId, instanceId,
                                                         TwillContainerMain.class, "$HADOOP_CONF_DIR");
    containers.put(runnableName, containerInfo.getId(), controller);
    TwillRunResources resources = new DynamicTwillRunResources(instanceId,
                                                               containerInfo.getId(),
                                                               containerInfo.getVirtualCores(),
                                                               containerInfo.getMemoryMB(),
                                                               containerInfo.getHost().getHostName(),
                                                               controller);
    resourceReport.addRunResources(runnableName, resources);
    containerStats.put(runnableName, containerInfo);

    if (startSequence.isEmpty() || !runnableName.equals(startSequence.peekLast())) {
      startSequence.addLast(runnableName);
    }
    containerChange.signalAll();

  } finally {
    containerLock.unlock();
  }
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
@Override
public boolean equals(Object o) {
  if (!(o instanceof TwillRunResources)) {
    return false;
  }
  TwillRunResources other = (TwillRunResources) o;
  return (instanceId == other.getInstanceId()) &&
    containerId.equals(other.getContainerId()) &&
    host.equals(other.getHost()) &&
    (virtualCores == other.getVirtualCores()) &&
    (memoryMB == other.getMemoryMB());
  // debugPort is ignored here
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
/**
 * Start a container for a runnable.
 */
void start(String runnableName, ContainerInfo containerInfo, TwillContainerLauncher launcher) {
  containerLock.lock();
  try {
    int instanceId = getStartInstanceId(runnableName);
    RunId runId = getRunId(runnableName, instanceId);
    TwillContainerController controller = launcher.start(runId, instanceId,
                                                         TwillContainerMain.class, "$HADOOP_CONF_DIR",
                                                         saveLogLevels());
    containers.put(runnableName, containerInfo.getId(), controller);
    TwillRunResources resources = new DynamicTwillRunResources(instanceId,
                                                               containerInfo.getId(),
                                                               containerInfo.getVirtualCores(),
                                                               containerInfo.getMemoryMB(),
                                                               launcher.getMaxHeapMemoryMB(),
                                                               containerInfo.getHost().getHostName(),
                                                               controller);
    resourceReport.addRunResources(runnableName, resources);
    containerStats.put(runnableName, containerInfo);

    if (startSequence.isEmpty() || !runnableName.equals(startSequence.peekLast())) {
      startSequence.addLast(runnableName);
    }
    containerChange.signalAll();       
    // call event handler containerLaunched. 
    eventHandler.containerLaunched(runnableName, instanceId, containerInfo.getId());
  } finally {
    containerLock.unlock();
  }
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private int getTwillContainersUsed(TwillController controller) {
  if (controller.getResourceReport() == null) {
    return 0;
  }

  int count = 1; // 1 for app master container
  ResourceReport resourceReport = controller.getResourceReport();
  for (Collection<TwillRunResources> resources : resourceReport.getResources().values()) {
    count += resources.size();
  }
  return count;
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private void waitForLogLevel(TwillController controller, String runnable, long timeout,
                             TimeUnit timeoutUnit, LogEntry.Level expected,
                             Map<String, LogEntry.Level> expectedArgs) throws InterruptedException {

  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  LogEntry.Level actual = null;
  Map<String, LogEntry.Level> actualArgs = null;
  boolean stopped = false;
  do {
    ResourceReport report = controller.getResourceReport();
    if (report == null || report.getRunnableResources(runnable) == null) {
      continue;
    }
    for (TwillRunResources resources : report.getRunnableResources(runnable)) {
      actual = resources.getLogLevels().get(Logger.ROOT_LOGGER_NAME);
      actualArgs = resources.getLogLevels();
      if (actual != null && actual.equals(expected)) {
        stopped = true;
        break;
      }
    }
    TimeUnit.MILLISECONDS.sleep(100);
  } while (!stopped && stopwatch.elapsedTime(timeoutUnit) < timeout);
  Assert.assertEquals(expected, actual);
  Assert.assertEquals(expectedArgs, actualArgs);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
public DefaultResourceReport(String applicationId, TwillRunResources masterResources,
                             Map<String, Collection<TwillRunResources>> resources, List<String> services) {
  this.applicationId = applicationId;
  this.appMasterResources = masterResources;
  this.usedResources = HashMultimap.create();
  for (Map.Entry<String, Collection<TwillRunResources>> entry : resources.entrySet()) {
    this.usedResources.putAll(entry.getKey(), entry.getValue());
  }
  this.services = new AtomicReference<>(services);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
/**
 * Remove the resource corresponding to the given runnable and container.
 *
 * @param runnableName name of runnable.
 * @param containerId container id of the runnable.
 */
public void removeRunnableResources(String runnableName, String containerId) {
  TwillRunResources toRemove = null;
  // could be faster if usedResources was a Table, but that makes returning the
  // report a little more complex, and this does not need to be terribly fast.
  for (TwillRunResources resources : usedResources.get(runnableName)) {
    if (resources.getContainerId().equals(containerId)) {
      toRemove = resources;
      break;
    }
  }
  usedResources.remove(runnableName, toRemove);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
@Override
public JsonElement serialize(ResourceReport src, Type typeOfSrc, JsonSerializationContext context) {
  JsonObject json = new JsonObject();

  json.addProperty("appMasterId", src.getApplicationId());
  json.add("appMasterResources", context.serialize(src.getAppMasterResources(), TwillRunResources.class));
  json.add("runnableResources", context.serialize(
    src.getResources(), new TypeToken<Map<String, Collection<TwillRunResources>>>() { }.getType()));
  json.add("services", context.serialize(
    src.getServices(), new TypeToken<List<String>>() { }.getType()));
  return json;
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
@Override
public ResourceReport deserialize(JsonElement json, Type typeOfT,
                                         JsonDeserializationContext context) throws JsonParseException {
  JsonObject jsonObj = json.getAsJsonObject();
  String appMasterId = jsonObj.get("appMasterId").getAsString();
  TwillRunResources masterResources = context.deserialize(
    jsonObj.get("appMasterResources"), TwillRunResources.class);
  Map<String, Collection<TwillRunResources>> resources = context.deserialize(
    jsonObj.get("runnableResources"), new TypeToken<Map<String, Collection<TwillRunResources>>>() { }.getType());
  List<String> services = context.deserialize(
    jsonObj.get("services"), new TypeToken<List<String>>() { }.getType());

  return new DefaultResourceReport(appMasterId, masterResources, resources, services);
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
private ResourceReportAdapter() {
  gson = new GsonBuilder()
            .serializeNulls()
            .setPrettyPrinting()
            .registerTypeAdapter(TwillRunResources.class, new TwillRunResourcesCodec())
            .registerTypeAdapter(ResourceReport.class, new ResourceReportCodec())
            .create();
}
 

import org.apache.twill.api.TwillRunResources; //导入依赖的package包/类
public DefaultResourceReport(String applicationId, TwillRunResources masterResources,
                             Map<String, Collection<TwillRunResources>> resources, List<String> services) {
  this.applicationId = applicationId;
  this.appMasterResources = masterResources;
  this.usedResources = HashMultimap.create();
  for (Map.Entry<String, Collection<TwillRunResources>> entry : resources.entrySet()) {
    this.usedResources.putAll(entry.getKey(), entry.getValue());
  }
  this.services = new AtomicReference<List<String>>(services);
}