code: boolean canBeSpeculatedUsingProcessingRate(long currentTime) {
  
      TaskStatus.Phase p = getProcessingPhase();
      // check if the task is on one of following four phases
      if ((p != TaskStatus.Phase.MAP) && 
          (p != TaskStatus.Phase.SHUFFLE) &&
          (p != TaskStatus.Phase.SORT) &&
          (p != TaskStatus.Phase.REDUCE)) {
        return false;
      }
      
      DataStatistics taskStats = job.getRunningTaskStatistics(p);
      if (LOG.isDebugEnabled()) {
        LOG.debug("TaskID: " + this.id + "processing phase is " + p +
            " and processing rate for this phase is " + 
            getProcessingRate(p));
      }
      // Find if task should be speculated based on standard deviation
      // the max difference allowed between the tasks's progress rate
      // and the mean progress rate of sibling tasks.
      
      double maxDiff = (taskStats.std() == 0 ? 
          taskStats.mean()/3 : 
            job.getSlowTaskThreshold() * taskStats.std());
      
      // if stddev > mean - we are stuck. cap the max difference at a 
      // more meaningful number.
      maxDiff = Math.min(maxDiff, taskStats.mean() * job.getStddevMeanRatioMax());
  
      return (taskStats.mean() - processingRates.getRate(p) > maxDiff);
    }
docs: For the map task, using the bytes processed/sec as the processing rate
  For the reduce task, using different rate for different phase:
  copy: using the bytes copied/sec as the processing rate
  sort: using the accumulated progress rate as the processing rate
  reduce: using the the bytes processed/sec as the processing rate
  @param currentTime
  @return