/**

    * Construct the splits, etc.  This is invoked from an async

    * thread so that split-computation doesn't block anyone.

    */

   //任务Task分两种: MapTask 和reduceTask，它们的管理对象都是TaskInProgress 。

   public synchronized void initTasks()

   throws IOException, KillInterruptedException, UnknownHostException {

     if (tasksInited || isComplete()) {

       return;

     }

     synchronized(jobInitKillStatus){

       if(jobInitKillStatus.killed || jobInitKillStatus.initStarted) {

         return;

       }

       jobInitKillStatus.initStarted = true;

     }

     LOG.info("Initializing " + jobId);

     final long startTimeFinal = this.startTime;

     // log job info as the user running the job

     try {

     userUGI.doAs(new PrivilegedExceptionAction<Object>() {

       @Override

       public Object run() throws Exception {

         JobHistory.JobInfo.logSubmitted(getJobID(), conf, jobFile,

             startTimeFinal, hasRestarted());

         return null;

       }

     });

     } catch(InterruptedException ie) {

       throw new IOException(ie);

     }

     // log the job priority

     setPriority(this.priority);

     //

     // generate security keys needed by Tasks

     //

     generateAndStoreTokens();

     //

     // read input splits and create a map per a split

     //

     TaskSplitMetaInfo[] splits = createSplits(jobId);

     if (numMapTasks != splits.length) {

       throw new IOException("Number of maps in JobConf doesn't match number of " +

               "recieved splits for job " + jobId + "! " +

               "numMapTasks=" + numMapTasks + ", #splits=" + splits.length);

     }

     numMapTasks = splits.length;//map task的个数就是input split的个数

     // Sanity check the locations so we don't create/initialize unnecessary tasks

     for (TaskSplitMetaInfo split : splits) {

       NetUtils.verifyHostnames(split.getLocations());

     }

     jobtracker.getInstrumentation().addWaitingMaps(getJobID(), numMapTasks);

     jobtracker.getInstrumentation().addWaitingReduces(getJobID(), numReduceTasks);

     this.queueMetrics.addWaitingMaps(getJobID(), numMapTasks);

     this.queueMetrics.addWaitingReduces(getJobID(), numReduceTasks);

     maps = new TaskInProgress[numMapTasks]; //为每个map tasks生成一个TaskInProgress来处理一个input split

     for(int i=0; i < numMapTasks; ++i) {

       inputLength += splits[i].getInputDataLength();

       maps[i] = new TaskInProgress(jobId, jobFile,         //类型是map task

                                    splits[i],

                                    jobtracker, conf, this, i, numSlotsPerMap);

     }

     LOG.info("Input size for job " + jobId + " = " + inputLength

         + ". Number of splits = " + splits.length);

     // Set localityWaitFactor before creating cache

     localityWaitFactor =

       conf.getFloat(LOCALITY_WAIT_FACTOR, DEFAULT_LOCALITY_WAIT_FACTOR);

     /* 对于map task，将其放入nonRunningMapCache，是一个Map<Node,List<TaskInProgress>>，也即对于map task来讲，其将会被分配到其input

     split所在的Node上。在此，Node代表一个datanode或者机架或者数据中  心。nonRunningMapCache将在JobTracker向TaskTracker分配map task的 时候使用。

     */

     if (numMapTasks > 0) {

         //通过createCache()方法为这些TaskInProgress对象产生一个未执行任务的Map缓存nonRunningMapCache。

         //slave端的TaskTracker向master发送心跳时，就可以直接从这个cache中取任务去执行。

       nonRunningMapCache = createCache(splits, maxLevel);

     }

     // set the launch time

     this.launchTime = jobtracker.getClock().getTime();

     //

     // Create reduce tasks

     //

     //其次JobInProgress会创建Reduce的监控对象，这个比较简单，根据JobConf里指定的Reduce数目创建，

     //缺省只创建1个Reduce任务。监控和调度Reduce任务的是TaskInProgress类，不过构造方法有所不同，

     //TaskInProgress会根据不同参数分别创建具体的MapTask或者ReduceTask。同样地，

     //initTasks()也会通过createCache()方法产生nonRunningReduceCache成员。

     this.reduces = new TaskInProgress[numReduceTasks];

     for (int i = 0; i < numReduceTasks; i++) {

       reduces[i] = new TaskInProgress(jobId, jobFile,     //这是reduce task

                                       numMapTasks, i,

                                       jobtracker, conf, this, numSlotsPerReduce);

       /*reducetask放入nonRunningReduces，其将在JobTracker向TaskTracker分配reduce task的时候使用。*/

       nonRunningReduces.add(reduces[i]);

     }

     // Calculate the minimum number of maps to be complete before

     // we should start scheduling reduces

     completedMapsForReduceSlowstart =

       (int)Math.ceil(

           (conf.getFloat("mapred.reduce.slowstart.completed.maps",

                          DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART) *

            numMapTasks));

     // ... use the same for estimating the total output of all maps

     resourceEstimator.setThreshhold(completedMapsForReduceSlowstart);

     // create cleanup two cleanup tips, one map and one reduce.

   //创建两个cleanup task，一个用来清理map，一个用来清理reduce.

     cleanup = new TaskInProgress[2];

     // cleanup map tip. This map doesn't use any splits. Just assign an empty

     // split.

     TaskSplitMetaInfo emptySplit = JobSplit.EMPTY_TASK_SPLIT;

     cleanup[0] = new TaskInProgress(jobId, jobFile, emptySplit,

             jobtracker, conf, this, numMapTasks, 1);

     cleanup[0].setJobCleanupTask();

     // cleanup reduce tip.

     cleanup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,

                        numReduceTasks, jobtracker, conf, this, 1);

     cleanup[1].setJobCleanupTask();

     // create two setup tips, one map and one reduce.

     //创建两个初始化 task，一个初始化map，一个初始化reduce.

     setup = new TaskInProgress[2];

     // setup map tip. This map doesn't use any split. Just assign an empty

     // split.

     setup[0] = new TaskInProgress(jobId, jobFile, emptySplit,

             jobtracker, conf, this, numMapTasks + 1, 1);

     setup[0].setJobSetupTask();

     // setup reduce tip.

     setup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,

                        numReduceTasks + 1, jobtracker, conf, this, 1);

     setup[1].setJobSetupTask();

     synchronized(jobInitKillStatus){

       jobInitKillStatus.initDone = true;

       if(jobInitKillStatus.killed) {

         throw new KillInterruptedException("Job " + jobId + " killed in init");

       }

     }

     //JobInProgress创建完TaskInProgress后，最后构造JobStatus并记录job正在执行中，

     //然后再调用JobHistory.JobInfo.logInited()记录job的执行日志。

     tasksInited = true;

     JobHistory.JobInfo.logInited(profile.getJobID(), this.launchTime,

                                  numMapTasks, numReduceTasks);

    // Log the number of map and reduce tasks

    LOG.info("Job " + jobId + " initialized successfully with " + numMapTasks

             + " map tasks and " + numReduceTasks + " reduce tasks.");

   }

 //JobQueueTaskScheduler最重要的方法是assignTasks，他实现了工作调度。

   @Override

   public synchronized List<Task> assignTasks(TaskTracker taskTracker)

       throws IOException {

     TaskTrackerStatus taskTrackerStatus = taskTracker.getStatus();

     ClusterStatus clusterStatus = taskTrackerManager.getClusterStatus();

     final int numTaskTrackers = clusterStatus.getTaskTrackers();

     final int clusterMapCapacity = clusterStatus.getMaxMapTasks();

     final int clusterReduceCapacity = clusterStatus.getMaxReduceTasks();

     Collection<JobInProgress> jobQueue =

       jobQueueJobInProgressListener.getJobQueue();

     //首先它会检查 TaskTracker 端还可以做多少个 map 和 reduce 任务，将要派发的任务数是否超出这个数，

     //是否超出集群的任务平均剩余可负载数。如果都没超出，则为此TaskTracker 分配一个 MapTask 或 ReduceTask 。

     //

     // Get map + reduce counts for the current tracker.

     //

     final int trackerMapCapacity = taskTrackerStatus.getMaxMapSlots();

     final int trackerReduceCapacity = taskTrackerStatus.getMaxReduceSlots();

     final int trackerRunningMaps = taskTrackerStatus.countMapTasks();

     final int trackerRunningReduces = taskTrackerStatus.countReduceTasks();

     // Assigned tasks

     List<Task> assignedTasks = new ArrayList<Task>();

     //

     // Compute (running + pending) map and reduce task numbers across pool

     //

   //计算剩余的map和reduce的工作量：remaining

     int remainingReduceLoad = 0;

     int remainingMapLoad = 0;

     synchronized (jobQueue) {

       for (JobInProgress job : jobQueue) {

         if (job.getStatus().getRunState() == JobStatus.RUNNING) {

           remainingMapLoad += (job.desiredMaps() - job.finishedMaps());

           if (job.scheduleReduces()) {

             remainingReduceLoad +=

               (job.desiredReduces() - job.finishedReduces());

           }

         }

       }

     }

     // Compute the 'load factor' for maps and reduces

     double mapLoadFactor = 0.0;

     if (clusterMapCapacity > 0) {

       mapLoadFactor = (double)remainingMapLoad / clusterMapCapacity;

     }

     double reduceLoadFactor = 0.0;

     if (clusterReduceCapacity > 0) {

       reduceLoadFactor = (double)remainingReduceLoad / clusterReduceCapacity;

     }

     //

     // In the below steps, we allocate first map tasks (if appropriate),

     // and then reduce tasks if appropriate.  We go through all jobs

     // in order of job arrival; jobs only get serviced if their

     // predecessors are serviced, too.

     //

     //

     // We assign tasks to the current taskTracker if the given machine

     // has a workload that's less than the maximum load of that kind of

     // task.

     // However, if the cluster is close to getting loaded i.e. we don't

     // have enough _padding_ for speculative executions etc., we only

     // schedule the "highest priority" task i.e. the task from the job

     // with the highest priority.

     //

     final int trackerCurrentMapCapacity =

       Math.min((int)Math.ceil(mapLoadFactor * trackerMapCapacity),

                               trackerMapCapacity);

     int availableMapSlots = trackerCurrentMapCapacity - trackerRunningMaps;

     boolean exceededMapPadding = false;

     if (availableMapSlots > 0) {

       exceededMapPadding =

         exceededPadding(true, clusterStatus, trackerMapCapacity);

     }

     int numLocalMaps = 0;

     int numNonLocalMaps = 0;

     scheduleMaps:

     for (int i=0; i < availableMapSlots; ++i) {

       synchronized (jobQueue) {

         for (JobInProgress job : jobQueue) {

           if (job.getStatus().getRunState() != JobStatus.RUNNING) {

             continue;

           }

           Task t = null;

           // Try to schedule a node-local or rack-local Map task

           t =

             job.obtainNewNodeOrRackLocalMapTask(taskTrackerStatus,

                 numTaskTrackers, taskTrackerManager.getNumberOfUniqueHosts());

           if (t != null) {

             assignedTasks.add(t);

             ++numLocalMaps;

             // Don't assign map tasks to the hilt!

             // Leave some free slots in the cluster for future task-failures,

             // speculative tasks etc. beyond the highest priority job

             if (exceededMapPadding) {

               break scheduleMaps;

             }

             // Try all jobs again for the next Map task

             break;

           }

           // Try to schedule a node-local or rack-local Map task

           //产生 Map 任务使用 JobInProgress 的obtainNewMapTask() 方法，

           //实质上最后调用了 JobInProgress 的 findNewMapTask() 访问nonRunningMapCache 。

           t =

             job.obtainNewNonLocalMapTask(taskTrackerStatus, numTaskTrackers,

                                    taskTrackerManager.getNumberOfUniqueHosts());

           if (t != null) {

             assignedTasks.add(t);

             ++numNonLocalMaps;

             // We assign at most 1 off-switch or speculative task

             // This is to prevent TaskTrackers from stealing local-tasks

             // from other TaskTrackers.

             break scheduleMaps;

           }

         }

       }

     }

     int assignedMaps = assignedTasks.size();

     //

     // Same thing, but for reduce tasks

     // However we _never_ assign more than 1 reduce task per heartbeat

     ////分配完map task，再分配reduce task

     final int trackerCurrentReduceCapacity =

       Math.min((int)Math.ceil(reduceLoadFactor * trackerReduceCapacity),

                trackerReduceCapacity);

     final int availableReduceSlots =

       Math.min((trackerCurrentReduceCapacity - trackerRunningReduces), 1);

     boolean exceededReducePadding = false;

     if (availableReduceSlots > 0) {

       exceededReducePadding = exceededPadding(false, clusterStatus,

                                               trackerReduceCapacity);

       synchronized (jobQueue) {

         for (JobInProgress job : jobQueue) {

           if (job.getStatus().getRunState() != JobStatus.RUNNING ||

               job.numReduceTasks == 0) {

             continue;

           }

           //使用JobInProgress.obtainNewReduceTask() 方法，

           //实质上最后调用了JobInProgress的 findNewReduceTask() 访问 nonRuningReduceCache

           Task t =

             job.obtainNewReduceTask(taskTrackerStatus, numTaskTrackers,

                                     taskTrackerManager.getNumberOfUniqueHosts()

                                     );

           if (t != null) {

             assignedTasks.add(t);

             break;

           }

           // Don't assign reduce tasks to the hilt!

           // Leave some free slots in the cluster for future task-failures,

           // speculative tasks etc. beyond the highest priority job

           if (exceededReducePadding) {

             break;

           }

         }

       }

     }

     if (LOG.isDebugEnabled()) {

       LOG.debug("Task assignments for " + taskTrackerStatus.getTrackerName() + " --> " +

                 "[" + mapLoadFactor + ", " + trackerMapCapacity + ", " +

                 trackerCurrentMapCapacity + ", " + trackerRunningMaps + "] -> [" +

                 (trackerCurrentMapCapacity - trackerRunningMaps) + ", " +

                 assignedMaps + " (" + numLocalMaps + ", " + numNonLocalMaps +

                 ")] [" + reduceLoadFactor + ", " + trackerReduceCapacity + ", " +

                 trackerCurrentReduceCapacity + "," + trackerRunningReduces +

                 "] -> [" + (trackerCurrentReduceCapacity - trackerRunningReduces) +

                 ", " + (assignedTasks.size()-assignedMaps) + "]");

     }

     return assignedTasks;

   }