@ Override public TransactionInfoBaseMessage getNextMessage ( ) throws IOException { if ( m_closed ) { throw new IOException ( "Closed" ) ; } if ( m_head == null ) { // Get another buffer asynchronously final Runnable r = new Runnable ( ) { @ Override public void run ( ) { try { BBContainer cont = m_reader . poll ( PersistentBinaryDeque . UNSAFE_CONTAINER_FACTORY ) ; if ( cont != null ) { m_headBuffers . offer ( new RejoinTaskBuffer ( cont ) ) ; } } catch ( Throwable t ) { VoltDB . crashLocalVoltDB ( "Error retrieving buffer data in task log" , true , t ) ; } finally { m_pendingPolls . decrementAndGet ( ) ; } } } ; //Always keep three buffers ready to go for ( int ii = m_pendingPolls . get ( ) + m_headBuffers . size ( ) ; ii < 3 ; ii ++ ) { m_pendingPolls . incrementAndGet ( ) ; m_es . execute ( r ) ; } m_head = m_headBuffers . poll ( ) ; } TransactionInfoBaseMessage nextTask = null ; if ( m_head != null ) { nextTask = m_head . nextTask ( ) ; if ( nextTask == null ) { scheduleDiscard ( m_head ) ; // current buffer is completely consumed, move to the next m_head = null ; } else { m_taskCount -- ; } } else if ( ( m_taskCount - m_tasksPendingInCurrentTail == 0 ) && m_tail != null ) { m_tasksPendingInCurrentTail = 0 ; /* * there is only one buffer left which hasn't been pushed into the * queue yet. set it to head directly, short-circuiting the queue. */ m_tail . compile ( ) ; if ( m_head != null ) { scheduleDiscard ( m_head ) ; } m_head = m_tail ; m_tail = null ; nextTask = getNextMessage ( ) ; } // SPs or fragments that's before the actual snapshot fragment may end up in the task log, // because there can be multiple snapshot fragments enabling the task log due to snapshot // collision. Need to filter tasks here based on their spHandles. if ( nextTask != null && nextTask . getSpHandle ( ) > m_snapshotSpHandle ) { return nextTask ; } else { return null ; } }
Try to get the next task message from the queue .