import org.apache.hadoop.mapred.JvmManager.JvmEnv; //导入依赖的package包/类
/**
 * Launch a new JVM for the task.
 * 
 * This method launches the new JVM for the task by executing the
 * the JVM command using the {@link Shell.ShellCommandExecutor}
 */
void launchTaskJVM(TaskController.TaskControllerContext context) 
                                    throws IOException {
  initializeTask(context);

  JvmEnv env = context.env;
  List<String> wrappedCommand = 
    TaskLog.captureOutAndError(env.setup, env.vargs, env.stdout, env.stderr,
        env.logSize, true);
  ShellCommandExecutor shexec = 
      new ShellCommandExecutor(wrappedCommand.toArray(new String[0]), 
                                env.workDir, env.env);
  // set the ShellCommandExecutor for later use.
  context.shExec = shexec;
  shexec.execute();
}
 

import org.apache.hadoop.mapred.JvmManager.JvmEnv; //导入依赖的package包/类
/**
 * Launch a new JVM for the task.
 * 
 * This method launches the new JVM for the task by executing the
 * the JVM command using the {@link Shell.ShellCommandExecutor}
 */
void launchTaskJVM(TaskController.TaskControllerContext context) 
                                    throws IOException {
  JvmEnv env = context.env;
  List<String> wrappedCommand = 
    TaskLog.captureOutAndError(env.setup, env.vargs, env.stdout, env.stderr,
        env.logSize, true);
  ShellCommandExecutor shexec = 
      new ShellCommandExecutor(wrappedCommand.toArray(new String[0]), 
                                env.workDir, env.env);
  // set the ShellCommandExecutor for later use.
  context.shExec = shexec;
  shexec.execute();
}
 

import org.apache.hadoop.mapred.JvmManager.JvmEnv; //导入依赖的package包/类
/**
 * Launch a task JVM that will run as the owner of the job.
 * 
 * This method launches a task JVM by executing a setuid
 * executable that will switch to the user and run the
 * task.
 */
@Override
void launchTaskJVM(TaskController.TaskControllerContext context) 
                                      throws IOException {
  JvmEnv env = context.env;
  // get the JVM command line.
  String cmdLine = 
    TaskLog.buildCommandLine(env.setup, env.vargs, env.stdout, env.stderr,
        env.logSize, true);

  StringBuffer sb = new StringBuffer();
  //export out all the environment variable before child command as
  //the setuid/setgid binaries would not be getting, any environmental
  //variables which begin with LD_*.
  for(Entry<String, String> entry : env.env.entrySet()) {
    sb.append("export ");
    sb.append(entry.getKey());
    sb.append("=");
    sb.append(entry.getValue());
    sb.append("\n");
  }
  sb.append(cmdLine);
  // write the command to a file in the
  // task specific cache directory
  writeCommand(sb.toString(), getTaskCacheDirectory(context));
  
  // Call the taskcontroller with the right parameters.
  List<String> launchTaskJVMArgs = buildLaunchTaskArgs(context);
  ShellCommandExecutor shExec =  buildTaskControllerExecutor(
                                  TaskCommands.LAUNCH_TASK_JVM, 
                                  env.conf.getUser(),
                                  launchTaskJVMArgs, env.workDir, env.env);
  context.shExec = shExec;
  try {
    shExec.execute();
  } catch (Exception e) {
    LOG.warn("Exception thrown while launching task JVM : " + 
        StringUtils.stringifyException(e));
    LOG.warn("Exit code from task is : " + shExec.getExitCode());
    LOG.warn("Output from task-contoller is : " + shExec.getOutput());
    throw new IOException(e);
  }
  if(LOG.isDebugEnabled()) {
    LOG.debug("output after executing task jvm = " + shExec.getOutput()); 
  }
}
 

import org.apache.hadoop.mapred.JvmManager.JvmEnv; //导入依赖的package包/类
/**
 * Launch a task JVM that will run as the owner of the job.
 * 
 * This method launches a task JVM by executing a setuid executable that will
 * switch to the user and run the task. Also does initialization of the first
 * task in the same setuid process launch.
 */
@Override
void launchTaskJVM(TaskController.TaskControllerContext context) 
                                      throws IOException {
  JvmEnv env = context.env;
  // get the JVM command line.
  String cmdLine = 
    TaskLog.buildCommandLine(env.setup, env.vargs, env.stdout, env.stderr,
        env.logSize, true);

  StringBuffer sb = new StringBuffer();
  //export out all the environment variable before child command as
  //the setuid/setgid binaries would not be getting, any environmental
  //variables which begin with LD_*.
  for(Entry<String, String> entry : env.env.entrySet()) {
    sb.append("export ");
    sb.append(entry.getKey());
    sb.append("=");
    sb.append(entry.getValue());
    sb.append("\n");
  }
  sb.append(cmdLine);
  // write the command to a file in the
  // task specific cache directory
  writeCommand(sb.toString(), getTaskCacheDirectory(context, 
      context.env.workDir));
  
  // Call the taskcontroller with the right parameters.
  List<String> launchTaskJVMArgs = buildLaunchTaskArgs(context, 
      context.env.workDir);
  ShellCommandExecutor shExec =  buildTaskControllerExecutor(
                                  TaskControllerCommands.LAUNCH_TASK_JVM, 
                                  env.conf.getUser(),
                                  launchTaskJVMArgs, env.workDir, env.env);
  context.shExec = shExec;
  try {
    shExec.execute();
  } catch (Exception e) {
    int exitCode = shExec.getExitCode();
    LOG.warn("Exit code from task is : " + exitCode);
    // 143 (SIGTERM) and 137 (SIGKILL) exit codes means the task was
    // terminated/killed forcefully. In all other cases, log the
    // task-controller output
    if (exitCode != 143 && exitCode != 137) {
      LOG.warn("Exception thrown while launching task JVM : "
          + StringUtils.stringifyException(e));
      LOG.info("Output from LinuxTaskController's launchTaskJVM follows:");
      logOutput(shExec.getOutput());
    }
    throw new IOException(e);
  }
  if (LOG.isDebugEnabled()) {
    LOG.info("Output from LinuxTaskController's launchTaskJVM follows:");
    logOutput(shExec.getOutput());
  }
}
 

import org.apache.hadoop.mapred.JvmManager.JvmEnv; //导入依赖的package包/类
private void initializeTask() throws IOException {
  tip.setJobConf(localizedJobConf);

  // ////////// The central method being tested
  tip.localizeTask(task);
  // //////////

  // check the functionality of localizeTask
  for (String dir : trackerFConf.getStrings(MRConfig.LOCAL_DIR)) {
    File attemptDir =
        new File(dir, TaskTracker.getLocalTaskDir(task.getUser(), jobId
            .toString(), taskId.toString(), task.isTaskCleanupTask()));
    assertTrue("attempt-dir " + attemptDir + " in localDir " + dir
        + " is not created!!", attemptDir.exists());
  }

  attemptWorkDir =
      lDirAlloc.getLocalPathToRead(TaskTracker.getTaskWorkDir(
          task.getUser(), task.getJobID().toString(), task.getTaskID()
              .toString(), task.isTaskCleanupTask()), trackerFConf);
  assertTrue("atttempt work dir for " + taskId.toString()
      + " is not created in any of the configured dirs!!",
      attemptWorkDir != null);

  TaskRunner runner = task.createRunner(tracker, tip);
  tip.setTaskRunner(runner);

  // /////// Few more methods being tested
  runner.setupChildTaskConfiguration(lDirAlloc);
  TaskRunner.createChildTmpDir(new File(attemptWorkDir.toUri().getPath()),
      localizedJobConf);
  attemptLogFiles = runner.prepareLogFiles(task.getTaskID(),
      task.isTaskCleanupTask());

  // Make sure the task-conf file is created
  Path localTaskFile =
      lDirAlloc.getLocalPathToRead(TaskTracker.getTaskConfFile(task
          .getUser(), task.getJobID().toString(), task.getTaskID()
          .toString(), task.isTaskCleanupTask()), trackerFConf);
  assertTrue("Task conf file " + localTaskFile.toString()
      + " is not created!!", new File(localTaskFile.toUri().getPath())
      .exists());

  // /////// One more method being tested. This happens in child space.
  localizedTaskConf = new JobConf(localTaskFile);
  TaskRunner.setupChildMapredLocalDirs(task, localizedTaskConf);
  // ///////

  // Initialize task via TaskController
  TaskControllerContext taskContext =
      new TaskController.TaskControllerContext();
  taskContext.env =
      new JvmEnv(null, null, null, null, -1, new File(localizedJobConf
          .get(TaskTracker.JOB_LOCAL_DIR)), null, localizedJobConf);
  taskContext.task = task;
  // /////////// The method being tested
  taskController.initializeTask(taskContext);
  // ///////////
}