Java 类org.apache.hadoop.mapred.TaskController.TaskControllerContext 实例源码

synchronized public TaskInProgress getTaskForJvm(JVMId jvmId) {
  if (jvmToRunningTask.containsKey(jvmId)) {
    //Incase of JVM reuse, tasks are returned to previously launched
    //JVM via this method. However when a new task is launched
    //the task being returned has to be initialized.
    TaskRunner taskRunner = jvmToRunningTask.get(jvmId);
    JvmRunner jvmRunner = jvmIdToRunner.get(jvmId);
    Task task = taskRunner.getTaskInProgress().getTask();
    TaskControllerContext context = 
      new TaskController.TaskControllerContext();
    context.env = jvmRunner.env;
    context.task = task;
    //If we are returning the same task as which the JVM was launched
    //we don't initialize task once again.
    if(!jvmRunner.env.conf.get("mapred.task.id").
        equals(task.getTaskID().toString())) {
      tracker.getTaskController().initializeTask(context);
    }

    jvmRunner.taskGiven(task);
    return taskRunner.getTaskInProgress();

  }
  return null;
}

synchronized public TaskInProgress getTaskForJvm(JVMId jvmId) {
  if (jvmToRunningTask.containsKey(jvmId)) {
    //Incase of JVM reuse, tasks are returned to previously launched
    //JVM via this method. However when a new task is launched
    //the task being returned has to be initialized.
    TaskRunner taskRunner = jvmToRunningTask.get(jvmId);
    JvmRunner jvmRunner = jvmIdToRunner.get(jvmId);
    Task task = taskRunner.getTaskInProgress().getTask();
    TaskControllerContext context = 
      new TaskController.TaskControllerContext();
    context.env = jvmRunner.env;
    context.task = task;
    //If we are returning the same task as which the JVM was launched
    //we don't initialize task once again.
    if(!jvmRunner.env.conf.get("mapred.task.id").
        equals(task.getTaskID().toString())) {
      tracker.getTaskController().initializeTask(context);
    }

    jvmRunner.taskGiven(task);
    return taskRunner.getTaskInProgress();

  }
  return null;
}

public void runChild(JvmEnv env) {
  initalContext = new TaskControllerContext();
  try {
    env.vargs.add(Integer.toString(jvmId.getId()));
    //Launch the task controller to run task JVM
    initalContext.task = jvmToRunningTask.get(jvmId).getTask();
    initalContext.env = env;
    tracker.getTaskController().initializeTask(initalContext);
    tracker.getTaskController().launchTaskJVM(initalContext);
  } catch (IOException ioe) {
    // do nothing
    // error and output are appropriately redirected
  } finally { // handle the exit code
    shexec = initalContext.shExec;
    if (shexec == null) {
      return;
    }

    kill();

    int exitCode = shexec.getExitCode();
    updateOnJvmExit(jvmId, exitCode);
    LOG.info("JVM : " + jvmId +" exited. Number of tasks it ran: " + 
        numTasksRan);
    try {
      // In case of jvm-reuse,
      //the task jvm cleans up the common workdir for every 
      //task at the beginning of each task in the task JVM.
      //For the last task, we do it here.
      if (env.conf.getNumTasksToExecutePerJvm() != 1) {
        tracker.directoryCleanupThread.addToQueue(
            TaskTracker.buildTaskControllerPathDeletionContexts(
                tracker.getLocalFileSystem(), tracker.getLocalDirs(),
                initalContext.task,
                true /* workDir */,
                tracker.getTaskController()));
      }
    } catch (IOException ie){}
  }
}

public void runChild(JvmEnv env) {
  initalContext = new TaskControllerContext();
  try {
    env.vargs.add(Integer.toString(jvmId.getId()));
    //Launch the task controller to run task JVM
    initalContext.task = jvmToRunningTask.get(jvmId).getTask();
    initalContext.env = env;
    tracker.getTaskController().initializeTask(initalContext);
    tracker.getTaskController().launchTaskJVM(initalContext);
  } catch (IOException ioe) {
    // do nothing
    // error and output are appropriately redirected
  } finally { // handle the exit code
    shexec = initalContext.shExec;
    if (shexec == null) {
      return;
    }

    kill();

    int exitCode = shexec.getExitCode();
    updateOnJvmExit(jvmId, exitCode);
    LOG.info("JVM : " + jvmId +" exited. Number of tasks it ran: " + 
        numTasksRan);
    try {
      // In case of jvm-reuse,
      //the task jvm cleans up the common workdir for every 
      //task at the beginning of each task in the task JVM.
      //For the last task, we do it here.
      if (env.conf.getNumTasksToExecutePerJvm() != 1) {
        tracker.directoryCleanupThread.addToQueue(
            TaskTracker.buildTaskControllerPathDeletionContexts(
                tracker.getLocalFileSystem(), tracker.getLocalDirs(),
                initalContext.task,
                true /* workDir */,
                tracker.getTaskController()));
      }
    } catch (IOException ie){}
  }
}

synchronized public TaskInProgress getTaskForJvm(JVMId jvmId)
    throws IOException {
  if (jvmToRunningTask.containsKey(jvmId)) {
    //Incase of JVM reuse, tasks are returned to previously launched
    //JVM via this method. However when a new task is launched
    //the task being returned has to be initialized.
    TaskRunner taskRunner = jvmToRunningTask.get(jvmId);
    JvmRunner jvmRunner = jvmIdToRunner.get(jvmId);
    Task task = taskRunner.getTaskInProgress().getTask();

    // Initialize task dirs
    TaskControllerContext context =
        new TaskController.TaskControllerContext();
    context.env = jvmRunner.env;
    context.task = task;
    // If we are returning the same task as which the JVM was launched
    // we don't initialize task once again.
    if (!jvmRunner.env.conf.get(JobContext.TASK_ATTEMPT_ID).equals(
        task.getTaskID().toString())) {
      try {
        tracker.getTaskController().initializeTask(context);
      } catch (IOException e) {
        LOG.warn("Failed to initialize the new task "
            + task.getTaskID().toString() + " to be given to JVM with id "
            + jvmId);
        throw e;
      }
    }

    return taskRunner.getTaskInProgress();
  }
  return null;
}

public JvmRunner(JvmEnv env, JobID jobId) {
  this.env = env;
  this.jvmId = new JVMId(jobId, isMap, rand.nextInt());
  this.numTasksToRun = env.conf.getNumTasksToExecutePerJvm();

  this.initalContext = new TaskControllerContext();
  initalContext.sleeptimeBeforeSigkill = tracker.getJobConf()
    .getLong(TTConfig.TT_SLEEP_TIME_BEFORE_SIG_KILL,
             ProcessTree.DEFAULT_SLEEPTIME_BEFORE_SIGKILL);
  LOG.info("In JvmRunner constructed JVM ID: " + jvmId);
}

public void doStackTrace(String pid) {
  TaskControllerContext context = new TaskControllerContext ();
  context.pid = pid;
  taskController.doStackTrace(context);
}

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);
  // ///////////
}