List < Task > getSetupAndCleanupTasks ( TaskTrackerStatus taskTracker ) throws IOException { int maxMapTasks = taskScheduler . getMaxSlots ( taskTracker , TaskType . MAP ) ; int maxReduceTasks = taskScheduler . getMaxSlots ( taskTracker , TaskType . REDUCE ) ; int numMaps = taskTracker . countOccupiedMapSlots ( ) ; int numReduces = taskTracker . countOccupiedReduceSlots ( ) ; int numTaskTrackers = getClusterStatus ( ) . getTaskTrackers ( ) ; int numUniqueHosts = getNumberOfUniqueHosts ( ) ; List < JobInProgress > cachedJobs = new ArrayList < JobInProgress > ( ) ; // get a snapshot of all the jobs in the system synchronized ( jobs ) { cachedJobs . addAll ( jobs . values ( ) ) ; } Task t = null ; if ( numMaps < maxMapTasks ) { for ( JobInProgress job : cachedJobs ) { t = job . obtainJobCleanupTask ( taskTracker , numTaskTrackers , numUniqueHosts , true ) ; if ( t != null ) { return Collections . singletonList ( t ) ; } } for ( JobInProgress job : cachedJobs ) { t = job . obtainTaskCleanupTask ( taskTracker , true ) ; if ( t != null ) { return Collections . singletonList ( t ) ; } } for ( JobInProgress job : cachedJobs ) { t = job . obtainJobSetupTask ( taskTracker , numTaskTrackers , numUniqueHosts , true ) ; if ( t != null ) { return Collections . singletonList ( t ) ; } } } if ( numReduces < maxReduceTasks ) { for ( JobInProgress job : cachedJobs ) { t = job . obtainJobCleanupTask ( taskTracker , numTaskTrackers , numUniqueHosts , false ) ; if ( t != null ) { return Collections . singletonList ( t ) ; } } for ( JobInProgress job : cachedJobs ) { t = job . obtainTaskCleanupTask ( taskTracker , false ) ; if ( t != null ) { return Collections . singletonList ( t ) ; } } for ( JobInProgress job : cachedJobs ) { t = job . obtainJobSetupTask ( taskTracker , numTaskTrackers , numUniqueHosts , false ) ; if ( t != null ) { return Collections . singletonList ( t ) ; } } } return null ; }
returns cleanup tasks first then setup tasks .