boolean matchesMemoryRequirements ( JobInProgress job , TaskType taskType , TaskTrackerStatus taskTracker ) { LOG . debug ( "Matching memory requirements of " + job . getJobID ( ) . toString ( ) + " for scheduling on " + taskTracker . trackerName ) ; if ( ! isSchedulingBasedOnMemEnabled ( ) ) { LOG . debug ( "Scheduling based on job's memory requirements is disabled." + " Ignoring any value set by job." ) ; return true ; } Long memUsedOnTT = getMemReservedForTasks ( taskTracker , taskType ) ; long totalMemUsableOnTT = 0 ; long memForThisTask = 0 ; if ( taskType == TaskType . MAP ) { memForThisTask = job . getJobConf ( ) . getMemoryForMapTask ( ) ; totalMemUsableOnTT = scheduler . getMemSizeForMapSlot ( ) * taskTracker . getMaxMapSlots ( ) ; } else if ( taskType == TaskType . REDUCE ) { memForThisTask = job . getJobConf ( ) . getMemoryForReduceTask ( ) ; totalMemUsableOnTT = scheduler . getMemSizeForReduceSlot ( ) * taskTracker . getMaxReduceSlots ( ) ; } long freeMemOnTT = totalMemUsableOnTT - memUsedOnTT . longValue ( ) ; if ( memForThisTask > freeMemOnTT ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "memForThisTask (" + memForThisTask + ") > freeMemOnTT (" + freeMemOnTT + "). A " + taskType + " task from " + job . getJobID ( ) . toString ( ) + " cannot be scheduled on TT " + taskTracker . trackerName ) ; } return false ; } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "memForThisTask = " + memForThisTask + ". freeMemOnTT = " + freeMemOnTT + ". A " + taskType . toString ( ) + " task from " + job . getJobID ( ) . toString ( ) + " matches memory requirements " + "on TT " + taskTracker . trackerName ) ; } return true ; }
Check if a TT has enough memory to run of task specified from this job .