Java 类org.apache.hadoop.util.ProcfsBasedProcessTree 实例源码

void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (reservedPhysicalMemoryOnTT == JobConf.DISABLED_MEMORY_LIMIT
      && totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1 and " +
             "reserved physical memory is not configured. " +
             "TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

private void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (fConf.get(TaskMemoryManagerThread.TT_RESERVED_PHYSICAL_MEMORY_MB) == null
      && totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1 and " +
                 "reserved physical memory is not configured. " +
             "TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1."
        + " TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

private void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (fConf.get(TaskMemoryManagerThread.TT_RESERVED_PHYSICAL_MEMORY_MB) == null
      && totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1 and " +
                 "reserved physical memory is not configured. " +
             "TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

private void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1."
        + " TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (reservedPhysicalMemoryOnTT == JobConf.DISABLED_MEMORY_LIMIT
      && totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1 and " +
             "reserved physical memory is not configured. " +
             "TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1."
        + " TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1."
        + " TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1."
        + " TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

private void setTaskMemoryManagerEnabledFlag() {
  if (!ProcfsBasedProcessTree.isAvailable()) {
    LOG.info("ProcessTree implementation is missing on this system. "
        + "TaskMemoryManager is disabled.");
    taskMemoryManagerEnabled = false;
    return;
  }

  if (totalMemoryAllottedForTasks == JobConf.DISABLED_MEMORY_LIMIT) {
    taskMemoryManagerEnabled = false;
    LOG.warn("TaskTracker's totalMemoryAllottedForTasks is -1."
        + " TaskMemoryManager is disabled.");
    return;
  }

  taskMemoryManagerEnabled = true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit, long memLimitPhysical) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
  this.memLimitPhysical = memLimitPhysical;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

/**
 * Kill the task and clean up ProcessTreeInfo
 * @param tid task attempt ID of the task to be killed.
 * @param msg diagnostics message
 */
private void killTask(TaskAttemptID tid, String msg) {
  // Kill the task and mark it as killed.
  taskTracker.cleanUpOverMemoryTask(tid, false, msg);
  // Now destroy the ProcessTree, remove it from monitoring map.
  ProcessTreeInfo ptInfo = processTreeInfoMap.get(tid);
  ProcfsBasedProcessTree pTree = ptInfo.getProcessTree();
  pTree.destroy(true/*in the background*/);
  processTreeInfoMap.remove(tid);
  LOG.info("Removed ProcessTree with root " + ptInfo.getPID());
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

/**
 * Kill the task and clean up ProcessTreeInfo
 * @param tid task attempt ID of the task to be killed.
 * @param msg diagonostic message
 * @param wasFailure if true, fail the task
 */
private void killTask(TaskAttemptID tid, String msg, boolean wasFailure) {
  // Kill the task and mark it as killed.
  taskTracker.cleanUpOverMemoryTask(tid, wasFailure, msg);
  // Now destroy the ProcessTree, remove it from monitoring map.
  ProcessTreeInfo ptInfo = processTreeInfoMap.get(tid);
  ProcfsBasedProcessTree pTree = ptInfo.getProcessTree();
  try {
    LinuxSystemCall.killProcessGroup(Integer.parseInt(ptInfo.getPID()));
  } catch (java.io.IOException e) {
    LOG.error("Could not kill process group " + ptInfo.getPID(), e);
  }
  processTreeInfoMap.remove(tid);
  LOG.info("Removed ProcessTree with root " + ptInfo.getPID());
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

/**
 * Kill the task and clean up ProcessTreeInfo
 * @param tid task attempt ID of the task to be killed.
 * @param msg diagonostic message
 * @param wasFailure if true, fail the task
 */
private void killTask(TaskAttemptID tid, String msg, boolean wasFailure) {
  // Kill the task and mark it as killed.
  taskTracker.cleanUpOverMemoryTask(tid, wasFailure, msg);
  // Now destroy the ProcessTree, remove it from monitoring map.
  ProcessTreeInfo ptInfo = processTreeInfoMap.get(tid);
  ProcfsBasedProcessTree pTree = ptInfo.getProcessTree();
  try {
    LinuxSystemCall.killProcessGroup(Integer.parseInt(ptInfo.getPID()));
  } catch (java.io.IOException e) {
    LOG.error("Could not kill process group " + ptInfo.getPID(), e);
  }
  processTreeInfoMap.remove(tid);
  LOG.info("Removed ProcessTree with root " + ptInfo.getPID());
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public TaskMemoryManagerThread(TaskTracker taskTracker) {

  this(taskTracker.getTotalMemoryAllottedForTasksOnTT() * 1024 * 1024L,
    taskTracker.getJobConf().getLong(
      "mapred.tasktracker.taskmemorymanager.monitoring-interval", 
      5000L),
    taskTracker.getJobConf().getLong(
      "mapred.tasktracker.procfsbasedprocesstree.sleeptime-before-sigkill",
      ProcfsBasedProcessTree.DEFAULT_SLEEPTIME_BEFORE_SIGKILL));

  this.taskTracker = taskTracker;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit, 
    long sleepTimeBeforeSigKill, String pidFile) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  if (this.pTree != null) {
    this.pTree.setSigKillInterval(sleepTimeBeforeSigKill);
  }
  this.memLimit = memLimit;
  this.pidFile = pidFile;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

/**
 * Load pid of the task from the pidFile.
 * 
 * @param pidFileName
 * @return the pid of the task process.
 */
private String getPid(String pidFileName) {
  if ((new File(pidFileName)).exists()) {
    return ProcfsBasedProcessTree.getPidFromPidFile(pidFileName);
   }
   return null;
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit, long memLimitPhysical) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
  this.memLimitPhysical = memLimitPhysical;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

/**
 * Kill the task and clean up ProcessTreeInfo
 * @param tid task attempt ID of the task to be killed.
 * @param msg diagnostics message
 */
private void killTask(TaskAttemptID tid, String msg) {
  // Kill the task and mark it as killed.
  taskTracker.cleanUpOverMemoryTask(tid, false, msg);
  // Now destroy the ProcessTree, remove it from monitoring map.
  ProcessTreeInfo ptInfo = processTreeInfoMap.get(tid);
  ProcfsBasedProcessTree pTree = ptInfo.getProcessTree();
  pTree.destroy(true/*in the background*/);
  processTreeInfoMap.remove(tid);
  LOG.info("Removed ProcessTree with root " + ptInfo.getPID());
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
}

boolean isProcessTreeOverLimit(ProcfsBasedProcessTree pTree, 
                                  String tId, long limit) {
  long currentMemUsage = pTree.getCumulativeVmem();
  // as processes begin with an age 1, we want to see if there are processes
  // more than 1 iteration old.
  long curMemUsageOfAgedProcesses = pTree.getCumulativeVmem(1);
  return isProcessTreeOverLimit(tId, currentMemUsage, 
                                curMemUsageOfAgedProcesses, limit);
}

private boolean isProcfsBasedTreeAvailable() {
  try {
    if (!ProcfsBasedProcessTree.isAvailable()) {
      LOG.info("Currently ProcessTree has only one implementation "
          + "ProcfsBasedProcessTree, which is not available on this "
          + "system. Not testing");
      return false;
    }
  } catch (Exception e) {
    LOG.info(StringUtils.stringifyException(e));
    return false;
  }
  return true;
}

public ProcessTreeInfo(TaskAttemptID tid, String pid,
    ProcfsBasedProcessTree pTree, long memLimit) {
  this.tid = tid;
  this.pid = pid;
  this.pTree = pTree;
  this.memLimit = memLimit;
}