private void failTasksWithMaxMemory ( long memoryToRelease ) { List < TaskAttemptID > allTasks = new ArrayList < TaskAttemptID > ( ) ; allTasks . addAll ( processTreeInfoMap . keySet ( ) ) ; // Sort the tasks descendingly according to RSS memory usage Collections . sort ( allTasks , new Comparator < TaskAttemptID > ( ) { @ Override public int compare ( TaskAttemptID tid1 , TaskAttemptID tid2 ) { return processTreeInfoMap . get ( tid2 ) . getMemoryUsed ( ) > processTreeInfoMap . get ( tid1 ) . getMemoryUsed ( ) ? 1 : - 1 ; } } ) ; long memoryReleased = 0 ; // Fail the tasks one by one until the memory requirement is met while ( memoryReleased < memoryToRelease && ! allTasks . isEmpty ( ) ) { TaskAttemptID tid = allTasks . remove ( 0 ) ; if ( ! isKillable ( tid ) ) { continue ; } long memoryUsed = processTreeInfoMap . get ( tid ) . getMemoryUsed ( ) ; if ( memoryUsed == 0 ) { break ; // Skip tasks without process tree information currently } tasksToKill . add ( tid ) ; memoryReleased += memoryUsed ; } if ( tasksToKill . isEmpty ( ) ) { LOG . error ( "The total memory usage is over CGroup limits. " + "But found no alive task to kill for freeing memory." ) ; } else if ( memoryReleased < memoryToRelease ) { LOG . error ( "The total memory usage is over CGroup limits. " + "But uanble to find enough tasks to kill for freeing memory." ) ; } killTasks ( ) ; }
Starting from the tasks use the highest amount of memory fail the tasks until the memory released meets the requirement