private void failTasksWithMaxRssMemory ( long rssMemoryInUsage , long availableRssMemory ) { List < TaskAttemptID > tasksToKill = new ArrayList < TaskAttemptID > ( ) ; 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 getTaskCumulativeRssmem ( tid2 ) > getTaskCumulativeRssmem ( tid1 ) ? 1 : - 1 ; } } ) ; long rssMemoryStillInUsage = rssMemoryInUsage ; long availableRssMemoryAfterKilling = availableRssMemory ; // Fail the tasks one by one until the memory requirement is met while ( ( rssMemoryStillInUsage > maxRssMemoryAllowedForAllTasks || availableRssMemoryAfterKilling < reservedRssMemory ) && ! allTasks . isEmpty ( ) ) { TaskAttemptID tid = allTasks . remove ( 0 ) ; if ( ! isKillable ( tid ) ) { continue ; } long rssmem = getTaskCumulativeRssmem ( tid ) ; if ( rssmem == 0 ) { break ; // Skip tasks without process tree information currently } tasksToKill . add ( tid ) ; rssMemoryStillInUsage -= rssmem ; availableRssMemoryAfterKilling += rssmem ; } // Now kill the tasks. if ( ! tasksToKill . isEmpty ( ) ) { for ( TaskAttemptID tid : tasksToKill ) { long taskMemoryLimit = getTaskMemoryLimit ( tid ) ; long taskMemory = getTaskCumulativeRssmem ( tid ) ; String pid = processTreeInfoMap . get ( tid ) . getPID ( ) ; String msg = HIGH_MEMORY_KEYWORD + " task:" + tid + " pid:" + pid + " taskMemory:" + taskMemory + " taskMemoryLimit:" + taskMemoryLimit + " availableMemory:" + availableRssMemory + " totalMemory:" + rssMemoryInUsage + " totalMemoryLimit:" + maxRssMemoryAllowedForAllTasks ; if ( taskMemory > taskMemoryLimit ) { msg = "Failing " + msg ; LOG . warn ( msg ) ; killTask ( tid , msg , true ) ; } else { msg = "Killing " + msg ; LOG . warn ( msg ) ; killTask ( tid , msg , false ) ; } } } else { LOG . error ( "The total physical memory usage is overflowing TTs limits. " + "But found no alive task to kill for freeing memory." ) ; } }
Starting from the tasks use the highest amount of RSS memory fail the tasks until the RSS memory meets the requirement