@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;
        }
    }
@ Override public TransactionInfoBaseMessage getNextMessage ( ) throws IOException { if ( m_closed ) { throw new IOException ( "Closed" ) ; } if ( m_head == null ) { 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 ( ) ; } } } ; 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 ) ; m_head = null ; } else { m_taskCount -- ; } } else if ( ( m_taskCount - m_tasksPendingInCurrentTail == 0 ) && m_tail != null ) { m_tasksPendingInCurrentTail = 0 ; m_tail . compile ( ) ; if ( m_head != null ) { scheduleDiscard ( m_head ) ; } m_head = m_tail ; m_tail = null ; nextTask = getNextMessage ( ) ; } if ( nextTask != null && nextTask . getSpHandle ( ) > m_snapshotSpHandle ) { return nextTask ; } else { return null ; } }
Try to get the next task message from the queue.

@return the next task message. null will be returned if either
the next task message is not ready or the queue is empty now.
@throws IOException
If failed to pull the next message out of the queue.