JobTracker节点后台线程之ExpireLaunchingTasks

待我称王封你为后i 2022-06-14 10:13 202阅读 0赞

上一篇博文主要讲到了JobTracker节点的后台线程ExpireTrackers，其中谈到了TaskTracker节点要向JobTracker节点定时地发送心跳包，来报告它当前的状态信息(TaskTracker的第一个心跳包同时起到了向JobTracker注册的作用)，而JobTracker对TaskTracker发送过来的心跳包的响应主要内容是分配给该TaskTracker的任务，也就是TaskTracker节点要执行这些任务。关于JobTracker是如何给一个TaskTracker节点分配合适的任务，不是本文将要讨论的内容，这涉及到Map-Reduce的作业调度器TaskSchedule。目前对于TaskSchedule的研究一直都是比较热门的话题，因为作业调度器TaskSchedule的好坏将关系到整个Hadoop集群的作业吞吐量。这里有一个异常情况就是，JobTracker节点在分给一个TaskTracker节点一批任务之后，它还需要确认TaskTracker节点是否收到了这些任务，所以为了保证用户响应的时间和充分利用这个Hadoop集群的资源，JobTracker节点必须能及时的检测到这样的任务，并能把它们交给其它比较空闲的TaskTracker来执行。至于JobTracker是如何这样做的，那就是本文将要重点阐述的了。

当JobTracker节点启动之后，它会开启一个后台线程ExpireLaunchingTasks来定时的检测分配给TaskTracker的任务是否已经被其成功收到。与ExpireLaunchingTasks线程相关的具体做法是：

1).当JobTracker给某一个TaskTracker节点分配了合适的Task之后，就会把这给Task通知给ExpireLaunchingTasks线程，好让它来对这些Task进行expire监测；

2).当ExpireLaunchingTasks线程收到一个需要监测的Task之后，会记录此时的时间，并把这个时间看作是监测的开始时间，即建立一个Task和开始时间的映射存储到它的属性launchingTasks中；

3).TaskTracker节点会定时的向JobTracker节点报告它当前的状态信息，这个状态信息主要包括正在TaskTracker节点上执行的Tasks状态信息(Task的状态信息又主要包括的是Task当前的执行进度)；

4).JobTracker收到TaskTracker节点发送过来的关于它正在执行的Tasks的状态信息之后，会针对每一个Task，先从ExpireLaunchingTasks线程中销毁它以前关于这个Task的expire监测，即从中删除Task与开始时间的映射；当然，如果这个Task的expire监测早就被销毁了，那此时就不用了；

5).ExpireLaunchingTasks作为一个后台线程，它会定时的对所有已注册的Task任务进行expire检测，如果有某一个Task在 **TASKTRACER\_EXPIRE\_INTERVAL** ms时间内没有被分配给它的TaskTracker节点执行的话(这里指的是开始执行，至于完没完成并不重要)，那么他就会通知这个Task所属的Job，说这个Task已经失败了，让Job重新找其它合适的TaskTracker节点来执行(当然这个工作本质上是由TaskSchedule来完成的)，同时把这个已经expire的Task从launchingTasks中清除。每一次检查的时间间隔是 **TASKTRACER\_EXPIRE\_INTERVAL**/3 ms。

对于参数**TASKTRACER\_EXPIRE\_INTERVA**，它和ExpireTrackers线程中所使用的参数是相同的，这其中的原因是不言而喻的。关于ExpireLaunchingTasks线程的实现是很简单的，它的源码如下：

private class ExpireLaunchingTasks implements Runnable {
        /**
         * This is a map of the tasks that have been assigned to task trackers,
         * but that have not yet been seen in a status report.
         * map: task-id -> time-assigned 
         */
        private Map<TaskAttemptID, Long> launchingTasks = new LinkedHashMap<TaskAttemptID, Long>();
          
        public void run() {
          while (true) {
            try {
              // Every 3 minutes check for any tasks that are overdue
              Thread.sleep(TASKTRACKER_EXPIRY_INTERVAL/3);
              long now = System.currentTimeMillis();
              LOG.debug("Starting launching task sweep");
              synchronized (JobTracker.this) {
                synchronized (launchingTasks) {
                  Iterator<Map.Entry<TaskAttemptID, Long>> itr = launchingTasks.entrySet().iterator();
                  while (itr.hasNext()) {
                    Map.Entry<TaskAttemptID, Long> pair = itr.next();
                    TaskAttemptID taskId = pair.getKey();
                    long age = now - (pair.getValue()).longValue();
                    LOG.info(taskId + " is " + age + " ms debug.");
                    if (age > TASKTRACKER_EXPIRY_INTERVAL) {
                      LOG.info("Launching task " + taskId + " timed out.");
                      TaskInProgress tip = null;
                      tip = taskidToTIPMap.get(taskId);
                      if (tip != null) {
                        JobInProgress job = tip.getJob();
                        String trackerName = getAssignedTracker(taskId);
                        TaskTrackerStatus trackerStatus = getTaskTracker(trackerName);
                        // This might happen when the tasktracker has already expired and this thread tries to call failedtask
                        // again. expire tasktracker should have called failed task!
                        if (trackerStatus != null)
                          job.failedTask(tip, taskId, "Error launching task", tip.isMapTask()? TaskStatus.Phase.MAP: TaskStatus.Phase.STARTING, TaskStatus.State.FAILED, trackerName);
                      }
                      itr.remove();
                    } else {
                      // the tasks are sorted by start time, so once we find one that we want to keep, we are done for this cycle.
                      break;
                    }
                  }
                  
                }
              }
            } catch (InterruptedException ie) {
              // all done
              break;
            } catch (Exception e) {
              LOG.error("Expire Launching Task Thread got exception: " +  StringUtils.stringifyException(e));
            }
          }
        }
    
        public void addNewTask1(TaskAttemptID taskName) {
          synchronized (launchingTasks) {
            launchingTasks.put(taskName, System.currentTimeMillis());
          }
        }
          
        public void removeTask(TaskAttemptID taskName) {
          synchronized (launchingTasks) {
            launchingTasks.remove(taskName);
          }
        }
        
    }

对于 ExpireLaunchingTasks 的设计，笔者不得不吐糟一下了。为了提高系统的效率，ExpireLaunchingTasks的检测间隔时间设置为**TASKTRACER\_EXPIRE\_INTERVAL**/3 ms，也即是说，对于一个异常的Task，这个后台检测线程在最坏情况下会延迟**TASKTRACER\_EXPIRE\_INTERVAL**/3 ms才会发现，那么就那些短作业而言，如作业在正常情况的执行时间小于**TASKTRACER\_EXPIRE\_INTERVAL**/3 ms，这种延迟相对于它们的处理时间而言是不可容忍的，所以对于短作业居多的应用场景中，应该将**TASKTRACER\_EXPIRE\_INTERVAL**的值设置的尽量小，但这里又有一个问题了，在大规模的集群下，如果**TASKTRACER\_EXPIRE\_INTERVAL**的值设置的比较小，那么ExpireTrackers线程就有可能频繁的判定集群中的TaskTracker节点发生了宕机。所以说，Hadoop的这种设计在某些应用场景下面局很难调和了。