TaskTracker上进行作业本地化的过程

下面是在TaskTracker上进行作业本地化的过程：

（注意Task与TaskTracker是运行在同一个节点上的，因为Task所在的JVM是由TaskTracker启动的）

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

    //注意下面这种“多个任务同时启动，但只需要一次作业本地化过程”的处理方式，

    //用localized、localing两个标记来“已经本地化”和“正在本地化”两个状态

    //还需要对任务共同的job同步

    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;

  }