private boolean atSpeculativeCap ( TaskType type ) { float numTasks = ( type == TaskType . MAP ) ? ( float ) ( runningMapTasks - speculativeMapTasks ) : ( float ) ( runningReduceTasks - speculativeReduceTasks ) ; if ( numTasks == 0 ) { return true ; // avoid divide by zero } int speculativeTaskCount = type == TaskType . MAP ? speculativeMapTasks : speculativeReduceTasks ; int totalSpeculativeTaskCount = type == TaskType . MAP ? totalSpeculativeMapTasks . get ( ) : totalSpeculativeReduceTasks . get ( ) ; //return true if totalSpecTask < max(10, 0.01 * total-slots, // 0.1 * total-running-tasks) if ( speculativeTaskCount < MIN_SPEC_CAP ) { return false ; // at least one slow tracker's worth of slots(default=10) } ClusterStatus c = jobtracker . getClusterStatus ( false ) ; int numSlots = ( type == TaskType . MAP ? c . getMaxMapTasks ( ) : c . getMaxReduceTasks ( ) ) ; if ( speculativeTaskCount < numSlots * MIN_SLOTS_CAP ) { return false ; } // Check if the total CAP has been reached if ( totalSpeculativeTaskCount >= numSlots * TOTAL_SPECULATIVECAP ) { return true ; } boolean atCap = ( ( ( speculativeTaskCount ) / numTasks ) >= speculativeCap ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "SpeculativeCap is " + speculativeCap + ", specTasks/numTasks is " + ( ( speculativeTaskCount ) / numTasks ) + ", so atSpecCap() is returning " + atCap ) ; } return atCap ; }
Check to see if the maximum number of speculative tasks are already being executed currently .