protected void preemptTasksIfNecessary ( ) { if ( ! preemptionEnabled || jobComparator == JobComparator . FIFO ) return ; long curTime = clock . getTime ( ) ; if ( curTime - lastPreemptCheckTime < preemptionInterval ) return ; lastPreemptCheckTime = curTime ; int currentMaxPreemptibleTasks = maxPreemptibleTasks ; boolean currentCountNonPreemptibleTasks = countNonPreemptibleTasks ; // Acquire locks on both the JobTracker (task tracker manager) and this // because we might need to call some JobTracker methods (killTask). synchronized ( taskTrackerManager ) { synchronized ( this ) { List < JobInProgress > jobs = new ArrayList < JobInProgress > ( infos . keySet ( ) ) ; for ( TaskType type : MAP_AND_REDUCE ) { int tasksToPreempt = 0 ; for ( JobInProgress job : jobs ) { if ( ! currentCountNonPreemptibleTasks && ! canBePreempted ( job ) ) { continue ; } tasksToPreempt += tasksToPreempt ( job , type , curTime ) ; } if ( tasksToPreempt > 0 ) { // for debugging purposes log the jobs by scheduling priority // to check whether preemption and scheduling are in sync. logJobStats ( sortedJobsByMapNeed , TaskType . MAP ) ; logJobStats ( sortedJobsByReduceNeed , TaskType . REDUCE ) ; } // Possibly adjust the maximum number of tasks to preempt. int actualTasksToPreempt = tasksToPreempt ; if ( ( currentMaxPreemptibleTasks >= 0 ) && ( tasksToPreempt > currentMaxPreemptibleTasks ) ) { actualTasksToPreempt = currentMaxPreemptibleTasks ; } LOG . info ( "preemptTasksIfNecessary: Should preempt " + tasksToPreempt + " " + type + " tasks, actually preempting " + actualTasksToPreempt + " tasks, countNonPreemptibleTasks = " + countNonPreemptibleTasks ) ; // Actually preempt the tasks. The policy for this is to pick // tasks from jobs that are above their min share and have very // negative deficits (meaning they've been over-scheduled). // However, we also want to minimize the amount of computation // wasted by preemption, so prefer tasks that started recently. preemptTasks ( jobs , type , tasksToPreempt ) ; } } } }
Check for jobs that need tasks preempted either because they have been below their guaranteed share for their pool s preemptionTimeout or they have been below half their fair share for the fairSharePreemptionTimeout . If such jobs exist compute how many tasks of each type need to be preempted and then select the right ones using selectTasksToPreempt .