Task运行过程分析1

调用org.apache.hadoop.mapred.TaskTracker.java中的startNewTask方法

 /**
   * Start a new task.
   * All exceptions are handled locally, so that we don't mess up the
   * task tracker.
   * @throws InterruptedException 
   */
  void startNewTask(final TaskInProgress tip) throws InterruptedException {
    Thread launchThread = new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          RunningJob rjob = localizeJob(tip);
          tip.getTask().setJobFile(rjob.getLocalizedJobConf().toString());
          // Localization is done. Neither rjob.jobConf nor rjob.ugi can be null
          launchTaskForJob(tip, new JobConf(rjob.getJobConf()), rjob); 
        } catch (Throwable e) {
          String msg = ("Error initializing " + tip.getTask().getTaskID() + 
                        ":\n" + StringUtils.stringifyException(e));
          LOG.warn(msg);
          tip.reportDiagnosticInfo(msg);
          try {
            tip.kill(true);
            tip.cleanup(false, true);
          } catch (IOException ie2) {
            LOG.info("Error cleaning up " + tip.getTask().getTaskID(), ie2);
          } catch (InterruptedException ie2) {
            LOG.info("Error cleaning up " + tip.getTask().getTaskID(), ie2);
          }
          if (e instanceof Error) {
            LOG.error("TaskLauncher error " + 
                StringUtils.stringifyException(e));
          }
        }
      }
    });
    launchThread.start();
  }

由于用户应用程序相关文件（比如jar包，字典文件等）可能很大，这使得任务花费较长时间从HDFS上下载这些数据，如果TaskTracker串行启动各个任务，势必会延长任务的启动时间。为了解决该问题，TaskTracker采用多线程启动任务，即为每个任务单独启动一个线程，简化代码如下：

void startNewTask(final TaskInProgress tip){
    Thread launchThread = new Thread(new Runnable()){
        public void run(){
            ...
            RunningJob rjob = localizeJob(tip) ;//作业本地化
            tip.getTask().setJobFile(rjob.getLocalizedJobConf().toString());
           // Localization is done. Neither rjob.jobConf nor rjob.ugi can be null
           launchTaskForJob(tip, new JobConf(rjob.getJobConf()), rjob);
           ... 
        }
    }
}

调用org.apache.hadoop.mapred.TaskTracker.java中的localizeJob方法，初始化job目录

 // intialize the job directory
  RunningJob localizeJob(TaskInProgress tip) 
  throws IOException, InterruptedException {
    Task t = tip.getTask();
    JobID jobId = t.getJobID();
    RunningJob rjob = addTaskToJob(jobId, tip);
    InetSocketAddress ttAddr = getTaskTrackerReportAddress();
    try {
      synchronized (rjob) {
        if (!rjob.localized) {
          while (rjob.localizing) {
            rjob.wait();
          }
          if (!rjob.localized) {
            //this thread is localizing the job
            rjob.localizing = true;
          }
        }
      }
      if (!rjob.localized) {
        Path localJobConfPath = initializeJob(t, rjob, ttAddr);
        JobConf localJobConf = new JobConf(localJobConfPath);
        //to be doubly sure, overwrite the user in the config with the one the TT 
        //thinks it is
        localJobConf.setUser(t.getUser());
        //also reset the #tasks per jvm
        resetNumTasksPerJvm(localJobConf);
        //set the base jobconf path in rjob; all tasks will use
        //this as the base path when they run
        synchronized (rjob) {
          rjob.localizedJobConf = localJobConfPath;
          rjob.jobConf = localJobConf;  
          rjob.keepJobFiles = ((localJobConf.getKeepTaskFilesPattern() != null) ||
              localJobConf.getKeepFailedTaskFiles());

          rjob.localized = true;
        }
      } 
    } finally {
      synchronized (rjob) {
        if (rjob.localizing) {
          rjob.localizing = false;
          rjob.notifyAll();
        }
      }
    }
    synchronized (runningJobs) {
      runningJobs.notify(); //notify the fetcher thread
    }
    return rjob;
  }

为了防止同一个作业的多个任务同时进行作业本地化，TaskTracker需要对相关数据结构加锁。一种常见的加锁方法是：

RunningJob rjob ；
...//对rjob赋值
synchronized(job){//获取rjob锁
    initializeJob(rjob) ;//作业本地化。如果作业文件很大，则该函数运行时间很长，这导致rjob锁长时间不释放
}

在以上实现方案中，作业本地化过程会一直占用rjob锁，这导致很多其他需要该锁的线程或者函数不得不等待。
为了避免作业本地化过程中长时间占用rjob锁，TaskTracker为每个正在运行的作业维护了两个
变量：localizing和localized，分别表示正在进行作业本地化和已经完成作业本地化。通过对这两个变量的控制可避免作业本地化时对RunningJob对象加锁，且能够保证只有作业的第一个任务进行作业本地化：

  synchronized (rjob) {
        if (!rjob.localized) { //该作业尚未完成本地化工作
          while (rjob.localizing) {//另为一个任务正在进行作业本地化
            rjob.wait();
          }
          if (!rjob.localized) {//没有任务进行作业本地化
            //this thread is localizing the job
            rjob.localizing = true;//让当前任务对该作业进行本地化
          }
        }
  }
  if (!rjob.localized) {//运行到此，说明当前没有任务进行作业本地化
       initializeJob(rjob); //进行作业本地化工作
  }

org.apache.hadoop.mapred.TaskTracker.java中的startNewTask方法调用了org.apache.hadoop.mapred.TaskTracker.java中的launchTaskForJob方法

 protected void launchTaskForJob(TaskInProgress tip, JobConf jobConf,
                                RunningJob rjob) throws IOException {
    synchronized (tip) {
      jobConf.set(JobConf.MAPRED_LOCAL_DIR_PROPERTY,
                  localStorage.getDirsString());
      tip.setJobConf(jobConf);
      tip.setUGI(rjob.ugi);
      tip.launchTask(rjob);
    }
  }

调用org.apache.hadoop.mapred.TaskTracker.java的launchTask方法

 /**
     * Kick off the task execution
     */
    public synchronized void launchTask(RunningJob rjob) throws IOException {
      if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED ||
          this.taskStatus.getRunState() == TaskStatus.State.FAILED_UNCLEAN ||
          this.taskStatus.getRunState() == TaskStatus.State.KILLED_UNCLEAN) {
        localizeTask(task);
        if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED) {
          this.taskStatus.setRunState(TaskStatus.State.RUNNING);
        }
        setTaskRunner(task.createRunner(TaskTracker.this, this, rjob));
        this.runner.start();
        long now = System.currentTimeMillis();
        this.taskStatus.setStartTime(now);
        this.lastProgressReport = now;
      } else {
        LOG.info("Not launching task: " + task.getTaskID() + 
            " since it's state is " + this.taskStatus.getRunState());
      }
    }

调用了org.apache.hadoop.mapred.Task.java中的CreateRunner方法。Task是个抽象类，两个子类分别是MapTask和ReduceTask。先关注Map的CreateRunner方法。
在org.apache.hadoop.mapred.MapTask.java中的CreateRunner方法

  @Override
  public TaskRunner createRunner(TaskTracker tracker, 
                                 TaskTracker.TaskInProgress tip,
                                 TaskTracker.RunningJob rjob
                                 ) throws IOException {
    return new MapTaskRunner(tip, tracker, this.conf, rjob);
  }

然后会启动MapTaskRunner所继承的run方法

 @Override
  public final void run() {
    String errorInfo = "Child Error";
    try {

      //before preparing the job localize 
      //all the archives
      TaskAttemptID taskid = t.getTaskID();
      final LocalDirAllocator lDirAlloc = new LocalDirAllocator("mapred.local.dir");
      //simply get the location of the workDir and pass it to the child. The
      //child will do the actual dir creation
      final File workDir =
      new File(new Path(localdirs[rand.nextInt(localdirs.length)], 
          TaskTracker.getTaskWorkDir(t.getUser(), taskid.getJobID().toString(), 
          taskid.toString(),
          t.isTaskCleanupTask())).toString());

      String user = tip.getUGI().getUserName();

      // Set up the child task's configuration. After this call, no localization
      // of files should happen in the TaskTracker's process space. Any changes to
      // the conf object after this will NOT be reflected to the child.
      // setupChildTaskConfiguration(lDirAlloc);

      if (!prepare()) {
        return;
      }

      // Accumulates class paths for child.
      List classPaths = getClassPaths(conf, workDir,
                                              taskDistributedCacheManager);

      long logSize = TaskLog.getTaskLogLength(conf);

      //  Build exec child JVM args.
      Vector vargs = getVMArgs(taskid, workDir, classPaths, logSize);

      tracker.addToMemoryManager(t.getTaskID(), t.isMapTask(), conf);

      // set memory limit using ulimit if feasible and necessary ...
      String setup = getVMSetupCmd();
      // Set up the redirection of the task's stdout and stderr streams
      File[] logFiles = prepareLogFiles(taskid, t.isTaskCleanupTask());
      File stdout = logFiles[0];
      File stderr = logFiles[1];
      tracker.getTaskTrackerInstrumentation().reportTaskLaunch(taskid, stdout,
                 stderr);

      Map env = new HashMap();
      errorInfo = getVMEnvironment(errorInfo, user, workDir, conf, env, taskid,
                                   logSize);

      // flatten the env as a set of export commands
      List  setupCmds = new ArrayList();
      for(Entry entry : env.entrySet()) {
        StringBuffer sb = new StringBuffer();
        sb.append("export ");
        sb.append(entry.getKey());
        sb.append("=\"");
        sb.append(entry.getValue());
        sb.append("\"");
        setupCmds.add(sb.toString());
      }
      setupCmds.add(setup);

      launchJvmAndWait(setupCmds, vargs, stdout, stderr, logSize, workDir);
      tracker.getTaskTrackerInstrumentation().reportTaskEnd(t.getTaskID());
      if (exitCodeSet) {
        if (!killed && exitCode != 0) {
          if (exitCode == 65) {
            tracker.getTaskTrackerInstrumentation().taskFailedPing(t.getTaskID());
          }
          throw new IOException("Task process exit with nonzero status of " +
              exitCode + ".");
        }
      }
    } catch (FSError e) {
      LOG.fatal("FSError", e);
      try {
        tracker.fsErrorInternal(t.getTaskID(), e.getMessage());
      } catch (IOException ie) {
        LOG.fatal(t.getTaskID()+" reporting FSError", ie);
      }
    } catch (Throwable throwable) {
      LOG.warn(t.getTaskID() + " : " + errorInfo, throwable);
      Throwable causeThrowable = new Throwable(errorInfo, throwable);
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      causeThrowable.printStackTrace(new PrintStream(baos));
      try {
        tracker.reportDiagnosticInfoInternal(t.getTaskID(), baos.toString());
      } catch (IOException e) {
        LOG.warn(t.getTaskID()+" Reporting Diagnostics", e);
      }
    } finally {

      // It is safe to call TaskTracker.TaskInProgress.reportTaskFinished with
      // *false* since the task has either
      // a) SUCCEEDED - which means commit has been done
      // b) FAILED - which means we do not need to commit
      tip.reportTaskFinished(false);
    }
  }

  void launchJvmAndWait(List  setup, Vector vargs, File stdout,
      File stderr, long logSize, File workDir)
      throws InterruptedException, IOException {
    jvmManager.launchJvm(this, jvmManager.constructJvmEnv(setup, vargs, stdout,
        stderr, logSize, workDir, conf));
    synchronized (lock) {
      while (!done) {
        lock.wait();
      }
    }
  }

  public boolean validateTipToJvm(TaskInProgress tip, JVMId jvmId) {
    if (jvmId.isMapJVM()) {
      return mapJvmManager.validateTipToJvm(tip, jvmId);
    } else {
      return reduceJvmManager.validateTipToJvm(tip, jvmId);
    }
  }

 private synchronized void reapJvm( 
        TaskRunner t, JvmEnv env) throws IOException, InterruptedException {
      if (t.getTaskInProgress().wasKilled()) {
        //the task was killed in-flight
        //no need to do the rest of the operations
        //如果task被杀死则直接返回
        return;
      }
      boolean spawnNewJvm = false;
      JobID jobId = t.getTask().getJobID();
       //检查是否有空闲的槽，如果小于最大jvm数，则重新开启一个jvm，不让你从现有job的空闲jvm中选择一个，或者杀死另外job的空闲jvm
      //Check whether there is a free slot to start a new JVM.
      //,or, Kill a (idle) JVM and launch a new one
      //When this method is called, we *must* 
      // (1) spawn a new JVM (if we are below the max) 
      // (2) find an idle JVM (that belongs to the same job), or,
      // (3) kill an idle JVM (from a different job) 
      // (the order of return is in the order above)
      int numJvmsSpawned = jvmIdToRunner.size();
      JvmRunner runnerToKill = null;
      if (numJvmsSpawned >= maxJvms) {
        //go through the list of JVMs for all jobs.
        Iterator<>> jvmIter = 
          jvmIdToRunner.entrySet().iterator();

        while (jvmIter.hasNext()) {
          JvmRunner jvmRunner = jvmIter.next().getValue();
          JobID jId = jvmRunner.jvmId.getJobId();
          //look for a free JVM for this job; if one exists then just break
          if (jId.equals(jobId) && !jvmRunner.isBusy() && !jvmRunner.ranAll()){
            setRunningTaskForJvm(jvmRunner.jvmId, t); //reserve the JVM
            LOG.info("No new JVM spawned for jobId/taskid: " + 
                     jobId+"/"+t.getTask().getTaskID() +
                     ". Attempting to reuse: " + jvmRunner.jvmId);
            return;
          }
          //Cases when a JVM is killed: 
          // (1) the JVM under consideration belongs to the same job 
          //     (passed in the argument). In this case, kill only when
          //     the JVM ran all the tasks it was scheduled to run (in terms
          //     of count).
          // (2) the JVM under consideration belongs to a different job and is
          //     currently not busy
          //But in both the above cases, we see if we can assign the current
          //task to an idle JVM (hence we continue the loop even on a match)
          if ((jId.equals(jobId) && jvmRunner.ranAll()) ||
              (!jId.equals(jobId) && !jvmRunner.isBusy())) {
            runnerToKill = jvmRunner;
            spawnNewJvm = true;
          }
        }
      } else {
        spawnNewJvm = true;
      }

      if (spawnNewJvm) {
        if (runnerToKill != null) {
          LOG.info("Killing JVM: " + runnerToKill.jvmId);
          killJvmRunner(runnerToKill);
        }
        spawnNewJvm(jobId, env, t);
        return;
      }
      //*MUST* never reach this
      try {
        LOG.fatal("Inconsistent state!!! " +
                "JVM Manager reached an unstable state " +
              "while reaping a JVM for task: " + t.getTask().getTaskID()+
              " " + getDetails() + ". Aborting. ");
      } catch (Exception e) {
        LOG.fatal(e);
      } finally {
        System.exit(-1);
      }
    }

 private void spawnNewJvm(JobID jobId, JvmEnv env,  
        TaskRunner t) {
      JvmRunner jvmRunner = new JvmRunner(env, jobId, t.getTask());
      jvmIdToRunner.put(jvmRunner.jvmId, jvmRunner);
      //spawn the JVM in a new thread. Note that there will be very little
      //extra overhead of launching the new thread for a new JVM since
      //most of the cost is involved in launching the process. Moreover,
      //since we are going to be using the JVM for running many tasks,
      //the thread launch cost becomes trivial when amortized over all
      //tasks. Doing it this way also keeps code simple.
      jvmRunner.setDaemon(true);
      jvmRunner.setName("JVM Runner " + jvmRunner.jvmId + " spawned.");
      setRunningTaskForJvm(jvmRunner.jvmId, t);
      LOG.info(jvmRunner.getName());
      jvmRunner.start();
    }

  @Override
      public void run() {
        try {
          runChild(env);
        } catch (InterruptedException ie) {
          return;
        } catch (IOException e) {
          LOG.warn("Caught IOException in JVMRunner", e);
        } catch (Throwable e) {
          LOG.error("Caught Throwable in JVMRunner. Aborting TaskTracker.", e);
          System.exit(1);
        } finally {
          jvmFinished();
        }
      }

  public void runChild(JvmEnv env) throws IOException, InterruptedException{
        int exitCode = 0;
        try {
          env.vargs.add(Integer.toString(jvmId.getId()));
          TaskRunner runner = jvmToRunningTask.get(jvmId);
          if (runner != null) {
            Task task = runner.getTask();
            //Launch the task controller to run task JVM
            String user = task.getUser();
            TaskAttemptID taskAttemptId = task.getTaskID();
            String taskAttemptIdStr = task.isTaskCleanupTask() ? 
                (taskAttemptId.toString() + TaskTracker.TASK_CLEANUP_SUFFIX) :
                  taskAttemptId.toString(); 
                exitCode = tracker.getTaskController().launchTask(user,
                    jvmId.jobId.toString(), taskAttemptIdStr, env.setup,
                    env.vargs, env.workDir, env.stdout.toString(),
                    env.stderr.toString());
          }
        } catch (IOException ioe) {
          // do nothing
          // error and output are appropriately redirected
        } finally { // handle the exit code
          // although the process has exited before we get here,
          // make sure the entire process group has also been killed.
          kill();
          updateOnJvmExit(jvmId, exitCode);
          LOG.info("JVM : " + jvmId + " exited with exit code " + exitCode
              + ". Number of tasks it ran: " + numTasksRan);
          deleteWorkDir(tracker, firstTask);
        }
      }

 /**
   * Create all of the directories for the task and launches the child jvm.
   * @param user the user name
   * @param attemptId the attempt id
   * @throws IOException
   */
  @Override
  public int launchTask(String user, 
                                  String jobId,
                                  String attemptId,
                                  List setup,
                                  List jvmArguments,
                                  File currentWorkDirectory,
                                  String stdout,
                                  String stderr) throws IOException {
    ShellCommandExecutor shExec = null;
    try {                   
      FileSystem localFs = FileSystem.getLocal(getConf());

      //create the attempt dirs
      new Localizer(localFs, 
          getConf().getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY)).
          initializeAttemptDirs(user, jobId, attemptId);

      // create the working-directory of the task 
      if (!currentWorkDirectory.mkdir()) {
        throw new IOException("Mkdirs failed to create " 
                    + currentWorkDirectory.toString());
      }

      //mkdir the loglocation
      String logLocation = TaskLog.getAttemptDir(jobId, attemptId).toString();
      if (!localFs.mkdirs(new Path(logLocation))) {
        throw new IOException("Mkdirs failed to create " 
                   + logLocation);
      }
      //read the configuration for the job
      FileSystem rawFs = FileSystem.getLocal(getConf()).getRaw();
      long logSize = 0; //TODO MAPREDUCE-1100
      // get the JVM command line.
      String cmdLine = 
        TaskLog.buildCommandLine(setup, jvmArguments,
            new File(stdout), new File(stderr), logSize, true);

      // write the command to a file in the
      // task specific cache directory
      // TODO copy to user dir
      Path p = new Path(allocator.getLocalPathForWrite(
          TaskTracker.getPrivateDirTaskScriptLocation(user, jobId, attemptId),
          getConf()), COMMAND_FILE);

      String commandFile = writeCommand(cmdLine, rawFs, p);
      rawFs.setPermission(p, TaskController.TASK_LAUNCH_SCRIPT_PERMISSION);
      /*
       * MAPREDUCE-2374: if another thread fork(2)ed a child process during the
       * window when writeCommand (above) had taskjvm.sh open for write, that
       * child process might still have a writeable fd open to the script.
       *
       * If we run the script with "bash -c /path/to/taskjvm.sh", then bash
       * would try to execve(2) the script and get ETXTBSY.  Instead, just have
       * bash interpret the script with "bash /path/to/taskjvm.sh".
       */
      shExec = new ShellCommandExecutor(new String[]{
          "bash", commandFile},
          currentWorkDirectory);
      shExec.execute();
    } catch (Exception e) {
      if (shExec == null) {
        return -1;
      }
      int exitCode = shExec.getExitCode();
      LOG.warn("Exit code from task is : " + exitCode);
      LOG.info("Output from DefaultTaskController's launchTask follows:");
      logOutput(shExec.getOutput());
      return exitCode;
    }
    return 0;
  }