int moveTasksFromBuffer ( Task [ ] tasks , int bufferIndex ) { // While a buffer is being drained it may be concurrently appended to. // The // number of tasks removed are tracked so that the length can be // decremented // by the delta rather than set to zero. Queue < Task > buffer = buffers [ bufferIndex ] ; int removedFromBuffer = 0 ; Task task ; int maxIndex = - 1 ; while ( ( task = buffer . poll ( ) ) != null ) { removedFromBuffer ++ ; // The index into the output array is determined by calculating the // offset // since the last drain int index = task . getOrder ( ) - drainedOrder ; if ( index < 0 ) { // The task was missed by the last drain and can be run // immediately task . run ( ) ; } else if ( index >= tasks . length ) { // Due to concurrent additions, the order exceeds the capacity // of the // output array. It is added to the end as overflow and the // remaining // tasks in the buffer will be handled by the next drain. maxIndex = tasks . length - 1 ; addTaskToChain ( tasks , task , maxIndex ) ; break ; } else { // Add the task to the array so that it is run in sequence maxIndex = Math . max ( index , maxIndex ) ; addTaskToChain ( tasks , task , index ) ; } } bufferLengths . addAndGet ( bufferIndex , - removedFromBuffer ) ; return maxIndex ; }
Moves the tasks from the specified buffer into the output array .