protected synchronized StealerBasedRebalanceTask scheduleNextTask ( boolean executeService ) { // Make sure there is work left to do. if ( doneSignal . getCount ( ) == 0 ) { logger . info ( "All tasks completion signaled... returning" ) ; return null ; } // Limit number of tasks outstanding. if ( this . numTasksExecuting >= maxParallelRebalancing ) { logger . info ( "Executing more tasks than [" + this . numTasksExecuting + "] the parallel allowed " + maxParallelRebalancing ) ; return null ; } // Shuffle list of stealer IDs each time a new task to schedule needs to // be found. Randomizing the order should avoid prioritizing one // specific stealer's work ahead of all others. List < Integer > stealerIds = new ArrayList < Integer > ( tasksByStealer . keySet ( ) ) ; Collections . shuffle ( stealerIds ) ; for ( int stealerId : stealerIds ) { if ( nodeIdsWithWork . contains ( stealerId ) ) { logger . info ( "Stealer " + stealerId + " is already working... continuing" ) ; continue ; } for ( StealerBasedRebalanceTask sbTask : tasksByStealer . get ( stealerId ) ) { int donorId = sbTask . getStealInfos ( ) . get ( 0 ) . getDonorId ( ) ; if ( nodeIdsWithWork . contains ( donorId ) ) { logger . info ( "Stealer " + stealerId + " Donor " + donorId + " is already working... continuing" ) ; continue ; } // Book keeping addNodesToWorkerList ( Arrays . asList ( stealerId , donorId ) ) ; numTasksExecuting ++ ; // Remove this task from list thus destroying list being // iterated over. This is safe because returning directly out of // this branch. tasksByStealer . get ( stealerId ) . remove ( sbTask ) ; try { if ( executeService ) { logger . info ( "Stealer " + stealerId + " Donor " + donorId + " going to schedule work" ) ; service . execute ( sbTask ) ; } } catch ( RejectedExecutionException ree ) { logger . error ( "Stealer " + stealerId + "Rebalancing task rejected by executor service." , ree ) ; throw new VoldemortRebalancingException ( "Stealer " + stealerId + "Rebalancing task rejected by executor service." ) ; } return sbTask ; } } printRemainingTasks ( stealerIds ) ; return null ; }
Schedule at most one task .