protected void update ( ) { //Making more granual locking so that clusterStatus can be fetched from Jobtracker. ClusterStatus clusterStatus = taskTrackerManager . getClusterStatus ( ) ; // Recompute locality delay from JobTracker heartbeat interval if enabled. // This will also lock the JT, so do it outside of a fair scheduler lock. if ( autoComputeLocalityDelay ) { JobTracker jobTracker = ( JobTracker ) taskTrackerManager ; localityDelayNodeLocal = Math . min ( MAX_AUTOCOMPUTED_LOCALITY_DELAY , ( long ) ( 1.5 * jobTracker . getNextHeartbeatInterval ( ) ) ) ; localityDelayRackLocal = localityDelayNodeLocal ; } // Got clusterStatus hence acquiring scheduler lock now // Remove non-running jobs synchronized ( this ) { // Reload allocations file if it hasn't been loaded in a while if ( poolMgr . reloadAllocsIfNecessary ( ) ) { // Check if the cluster have enough slots for reserving poolMgr . checkMinimumSlotsAvailable ( clusterStatus , TaskType . MAP ) ; poolMgr . checkMinimumSlotsAvailable ( clusterStatus , TaskType . REDUCE ) ; } List < JobInProgress > toRemove = new ArrayList < JobInProgress > ( ) ; for ( JobInProgress job : infos . keySet ( ) ) { int runState = job . getStatus ( ) . getRunState ( ) ; if ( runState == JobStatus . SUCCEEDED || runState == JobStatus . FAILED || runState == JobStatus . KILLED ) { toRemove . add ( job ) ; } } for ( JobInProgress job : toRemove ) { infos . remove ( job ) ; poolMgr . removeJob ( job ) ; } // Update running jobs with deficits since last update, and compute new // slot allocations, weight, shares and task counts long now = clock . getTime ( ) ; long timeDelta = now - lastUpdateTime ; updateDeficits ( timeDelta ) ; updateRunnability ( ) ; updateTaskCounts ( ) ; updateWeights ( ) ; updateMinAndMaxSlots ( ) ; updateFairShares ( clusterStatus ) ; if ( preemptionEnabled ) { updatePreemptionVariables ( ) ; } sortJobs ( ) ; updatePoolMetrics ( ) ; dumpStatus ( now ) ; lastUpdateTime = now ; } }
Recompute the internal variables used by the scheduler - per - job weights fair shares deficits minimum slot allocations and numbers of running and needed tasks of each type .