private void progressTaskStatus ( SimulatorTaskInProgress tip , long now ) { TaskStatus status = tip . getTaskStatus ( ) ; if ( status . getRunState ( ) != State . RUNNING ) { return ; // nothing to be done } boolean isMap = tip . isMapTask ( ) ; // Time when the user space code started long startTime = - 1 ; // Time spent in map or just in the REDUCE phase of a reduce task long runTime = tip . getUserSpaceRunTime ( ) ; float progress = 0.0f ; if ( isMap ) { // We linearly estimate the progress of maps since their start startTime = status . getStartTime ( ) ; progress = ( ( float ) ( now - startTime ) ) / runTime ; } else { // We don't model reduce progress in the SHUFFLE or SORT phases // We use linear estimate for the 3rd, REDUCE phase Phase reducePhase = status . getPhase ( ) ; switch ( reducePhase ) { case SHUFFLE : progress = 0.0f ; // 0 phase is done out of 3 break ; case SORT : progress = 1.0f / 3 ; // 1 phase is done out of 3 break ; case REDUCE : { // REDUCE phase with the user code started when sort finished startTime = status . getSortFinishTime ( ) ; // 0.66f : 2 phases are done out of of 3 progress = 2.0f / 3 + ( ( ( float ) ( now - startTime ) ) / runTime ) / 3.0f ; } break ; default : // should never get here throw new IllegalArgumentException ( "Invalid reducePhase=" + reducePhase ) ; } } final float EPSILON = 0.0001f ; if ( progress < - EPSILON || progress > 1 + EPSILON ) { throw new IllegalStateException ( "Task progress out of range: " + progress ) ; } progress = Math . max ( Math . min ( 1.0f , progress ) , 0.0f ) ; status . setProgress ( progress ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Updated task progress, taskId=" + status . getTaskID ( ) + ", progress=" + status . getProgress ( ) ) ; } }
Updates the progress indicator of a task if it is running .