private void releaseBlocksAndResetBarriers ( ) throws IOException { LOG . debug ( "{}: End of stream alignment, feeding buffered data back." , inputGate . getOwningTaskName ( ) ) ; for ( int i = 0 ; i < blockedChannels . length ; i ++ ) { blockedChannels [ i ] = false ; } if ( currentBuffered == null ) { // common case: no more buffered data currentBuffered = bufferBlocker . rollOverReusingResources ( ) ; if ( currentBuffered != null ) { currentBuffered . open ( ) ; } } else { // uncommon case: buffered data pending // push back the pending data, if we have any LOG . debug ( "{}: Checkpoint skipped via buffered data:" + "Pushing back current alignment buffers and feeding back new alignment data first." , inputGate . getOwningTaskName ( ) ) ; // since we did not fully drain the previous sequence, we need to allocate a new buffer for this one BufferOrEventSequence bufferedNow = bufferBlocker . rollOverWithoutReusingResources ( ) ; if ( bufferedNow != null ) { bufferedNow . open ( ) ; queuedBuffered . addFirst ( currentBuffered ) ; numQueuedBytes += currentBuffered . size ( ) ; currentBuffered = bufferedNow ; } } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "{}: Size of buffered data: {} bytes" , inputGate . getOwningTaskName ( ) , currentBuffered == null ? 0L : currentBuffered . size ( ) ) ; } // the next barrier that comes must assume it is the first numBarriersReceived = 0 ; if ( startOfAlignmentTimestamp > 0 ) { latestAlignmentDurationNanos = System . nanoTime ( ) - startOfAlignmentTimestamp ; startOfAlignmentTimestamp = 0 ; } }
Releases the blocks on all channels and resets the barrier count . Makes sure the just written data is the next to be consumed .