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;
   }
private boolean atSpeculativeCap ( TaskType type ) { float numTasks = ( type == TaskType . MAP ) ? ( float ) ( runningMapTasks - speculativeMapTasks ) : ( float ) ( runningReduceTasks - speculativeReduceTasks ) ; if ( numTasks == 0 ) { return true ; } int speculativeTaskCount = type == TaskType . MAP ? speculativeMapTasks : speculativeReduceTasks ; int totalSpeculativeTaskCount = type == TaskType . MAP ? totalSpeculativeMapTasks . get ( ) : totalSpeculativeReduceTasks . get ( ) ; if ( speculativeTaskCount < MIN_SPEC_CAP ) { return false ; } ClusterStatus c = jobtracker . getClusterStatus ( false ) ; int numSlots = ( type == TaskType . MAP ? c . getMaxMapTasks ( ) : c . getMaxReduceTasks ( ) ) ; if ( speculativeTaskCount < numSlots * MIN_SLOTS_CAP ) { return false ; } 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.
@param tasks the set of tasks to test
@param type the type of task (MAP/REDUCE) that we are considering
@return has the cap been reached?