TaskTracker启动过程源码级分析

  1 /**
  2    * Main service loop.  Will stay in this loop forever.
  3    */
  4   State offerService() throws Exception {
  5     long lastHeartbeat = System.currentTimeMillis();//上一次发心跳距现在时间
  6     //此循环主要根据控制完成task个数控制心跳间隔。//TaskTracker进行是一直存在的
  7     while (running && !shuttingDown) {
  8       try {
  9         long now = System.currentTimeMillis();//获得当前时间    
 10         
 11         // accelerate to account for multiple finished tasks up-front
 12       //通过完成的任务数动态控制心跳间隔时间  
 13         long remaining = 
 14           (lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;
 15         while (remaining > 0) {
 16           // sleeps for the wait time or 
 17           // until there are *enough* empty slots to schedule tasks
 18           synchronized (finishedCount) {
 19             finishedCount.wait(remaining);
 20             
 21             // Recompute
 22             now = System.currentTimeMillis();
 23             remaining = 
 24               (lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;
 25             
 26             if (remaining <= 0) {
 27               // Reset count 
 28               finishedCount.set(0);
 29               break;
 30             }
 31           }
 32         }
 33 
 34         // If the TaskTracker is just starting up:
 35         // 1. Verify the buildVersion
 36         // 2. Get the system directory & filesystem
 37         if(justInited) {
 38           String jobTrackerBV = jobClient.getBuildVersion();
 39           if(!VersionInfo.getBuildVersion().equals(jobTrackerBV)) {
 40             String msg = "Shutting down. Incompatible buildVersion." +
 41             "\nJobTracker‘s: " + jobTrackerBV + 
 42             "\nTaskTracker‘s: "+ VersionInfo.getBuildVersion();
 43             LOG.error(msg);
 44             try {
 45               jobClient.reportTaskTrackerError(taskTrackerName, null, msg);
 46             } catch(Exception e ) {
 47               LOG.info("Problem reporting to jobtracker: " + e);
 48             }
 49             return State.DENIED;
 50           }
 51           
 52           String dir = jobClient.getSystemDir();
 53           if (dir == null) {
 54             throw new IOException("Failed to get system directory");
 55           }
 56           systemDirectory = new Path(dir);
 57           systemFS = systemDirectory.getFileSystem(fConf);
 58         }
 59 
 60         now = System.currentTimeMillis();
 61         if (now > (lastCheckDirsTime + diskHealthCheckInterval)) {
 62           localStorage.checkDirs();
 63           lastCheckDirsTime = now;
 64           int numFailures = localStorage.numFailures();
 65           // Re-init the task tracker if there were any new failures
 66           if (numFailures > lastNumFailures) {
 67             lastNumFailures = numFailures;
 68             return State.STALE;
 69           }
 70         }
 71 
 72         // Send the heartbeat and process the jobtracker‘s directives
 73         //在transmitHeartBeat()函数处理中，
 74         //TaskTracker会创建一个新的TaskTrackerStatus对象记录目前任务的执行状况，
 75         //检查目前执行的Task数目以及本地磁盘的空间使用情况等，
 76         //如果可以接收新的Task则设置heartbeat()的askForNewTask参数为true。
 77         //然后通过IPC接口调用JobTracker的heartbeat()方法发送过去，
 78         //heartbeat()返回值TaskTrackerAction数组。
 79         HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);//发送Heartbeat到JobTracker，得到response 
 80 
 81         // Note the time when the heartbeat returned, use this to decide when to send the
 82         // next heartbeat   
 83         lastHeartbeat = System.currentTimeMillis();
 84         
 85         // Check if the map-event list needs purging
 86         Set<JobID> jobs = heartbeatResponse.getRecoveredJobs();
 87         if (jobs.size() > 0) {
 88           synchronized (this) {
 89             // purge the local map events list
 90             for (JobID job : jobs) {
 91               RunningJob rjob;
 92               synchronized (runningJobs) {
 93                 rjob = runningJobs.get(job);          
 94                 if (rjob != null) {
 95                   synchronized (rjob) {
 96                     FetchStatus f = rjob.getFetchStatus();
 97                     if (f != null) {
 98                       f.reset();
 99                     }
100                   }
101                 }
102               }
103             }
104 
105             // Mark the reducers in shuffle for rollback
106             synchronized (shouldReset) {
107               for (Map.Entry<TaskAttemptID, TaskInProgress> entry 
108                    : runningTasks.entrySet()) {
109                 if (entry.getValue().getStatus().getPhase() == Phase.SHUFFLE) {
110                   this.shouldReset.add(entry.getKey());
111                 }
112               }
113             }
114           }
115         }
116         //然后调用HeartbeatResponse的getActions()函数获得JobTracker传过来的所有指令即一个TaskTrackerAction数组。
117         TaskTrackerAction[] actions = heartbeatResponse.getActions();
118         if(LOG.isDebugEnabled()) {
119           LOG.debug("Got heartbeatResponse from JobTracker with responseId: " + 
120                     heartbeatResponse.getResponseId() + " and " + 
121                     ((actions != null) ? actions.length : 0) + " actions");
122         }
123         if (reinitTaskTracker(actions)) {
124           return State.STALE;
125         }
126             
127         // resetting heartbeat interval from the response.
128         heartbeatInterval = heartbeatResponse.getHeartbeatInterval();
129         justStarted = false;
130         justInited = false;
131         if (actions != null){ 
132             //遍历这个数组，如果是一个新任务指令即LaunchTaskAction则调用调用addToTaskQueue加入到待执行队列，
133             //否则加入到tasksToCleanup队列，交给一个taskCleanupThread线程来处理，如执行KillJobAction或者KillTaskAction等。
134           for(TaskTrackerAction action: actions) {
135             if (action instanceof LaunchTaskAction) {
136               addToTaskQueue((LaunchTaskAction)action);  //如果是运行一个新的Task，则将Action添加到任务队列中,加入TaskLauncher线程的执行队列
137             } else if (action instanceof CommitTaskAction) {
138               CommitTaskAction commitAction = (CommitTaskAction)action;
139               if (!commitResponses.contains(commitAction.getTaskID())) {
140                 LOG.info("Received commit task action for " + 
141                           commitAction.getTaskID());
142                 commitResponses.add(commitAction.getTaskID());
143               }
144             } else {//杀死任务或作业
145               tasksToCleanup.put(action);
146             }
147           }
148         }
149       //杀死一定时间没没有汇报进度的task 
150         markUnresponsiveTasks();
151       //当剩余磁盘空间小于mapred.local.dir.minspacekill(默认为0)时，寻找合适的任务将其杀掉以释放空间 
152         killOverflowingTasks();
153             
154         //we‘ve cleaned up, resume normal operation
155         if (!acceptNewTasks && isIdle()) {
156           acceptNewTasks=true;
157         }
158         //The check below may not be required every iteration but we are 
159         //erring on the side of caution here. We have seen many cases where
160         //the call to jetty‘s getLocalPort() returns different values at 
161         //different times. Being a real paranoid here.
162         checkJettyPort(server.getPort());
163       } catch (InterruptedException ie) {
164         LOG.info("Interrupted. Closing down.");
165         return State.INTERRUPTED;
166       } catch (DiskErrorException de) {
167         String msg = "Exiting task tracker for disk error:\n" +
168           StringUtils.stringifyException(de);
169         LOG.error(msg);
170         synchronized (this) {
171           jobClient.reportTaskTrackerError(taskTrackerName, 
172                                            "DiskErrorException", msg);
173         }
174         return State.STALE;
175       } catch (RemoteException re) {
176         String reClass = re.getClassName();
177         if (DisallowedTaskTrackerException.class.getName().equals(reClass)) {
178           LOG.info("Tasktracker disallowed by JobTracker.");
179           return State.DENIED;
180         }
181       } catch (Exception except) {
182         String msg = "Caught exception: " + 
183           StringUtils.stringifyException(except);
184         LOG.error(msg);
185       }
186     }
187 
188     return State.NORMAL;
189   }