com.allanbank.mongodb.connection.message

Class PendingMessageQueue

java.lang.Object

com.allanbank.mongodb.connection.message.PendingMessageQueue

public final class PendingMessageQueue
extends Object

PendingMessageQueue provides an optimized queue for pending messages inspired by the Disruptor project.

To reduce thread contention the queue uses a set of integer values to track the position of the ready messages (the last message that is ready to be read), reserve (the first message that can be reserved to be written to), and the take (the next (first) message to be read). For an infinite queue the following invariant holds:

 
 take < readyBefore <= reserve
 
 

To make handling a limited size queue easier the size of the queue is forced to power of 2 less than 1048576. The roll over can then be handled with a simple mask operation.

Rather than allocate a pending message per request we use an array of pre-allocated PendingMessages and copy the data into and out of the objects. this has a net positive effect on object allocation and garbage collection time at the cost of a longer initialization.

Lastly, This queue assumes there is a single consumer of messages. This is true for the driver's use case but don't copy the code and expect it to work with multiple consumers. The consumer should use the following basic structure:

 
 PendingMessage pm = new PendingMessage();
 
 queue.take(pm); // Blocks.
 // Handle the message.
    
 // or
    
 if( queue.poll(pm) ) { // Non-blocking.
    // Handle The Message.
 }
 
 

Warning: This class has been carefully tuned for the driver's use case. Changes should be carefully bench marked and tested. Comments have been embedded in the source indicating attempted changes and reverts. Due to its position in the driver subtle changes in this class can cause large changes in the performance of the driver.

See Also:

Disruptor Project

API Warning:

This class is NOT part of the drivers API. This class may be mutated in incompatible ways between any two releases of the driver.

Copyright ©:

2012-2013, Allanbank Consulting, Inc., All Rights Reserved

Field Summary

Fields

Modifier and Type	Field and Description
static long	MAX_MESSAGE_ID_MASK The mask for constraining the size the message id.
static int	MAX_SIZE The maximum size of the queue.
static long	SPIN_TIME_NS Amount of time to spin before yielding.

Constructor Summary

Constructors

Constructor and Description
PendingMessageQueue(int size, LockType lockType) Creates a new PendingMessageQueue.

Method Summary

Methods

Modifier and Type	Method and Description
int	capacity() Returns the size of the queue.
void	drainTo(List<PendingMessage> pending) Drains the list of pending messages into the provided list.
protected int	increment(int index) Increments the index handling roll-over.
boolean	isEmpty() Returns true if the queue is empty.
protected void	markReady(int index) Marks the position as ready by incrementing the ready position to the provided position.
protected void	markReady2(int index) Marks the position and the next position as ready by incrementing the ready position to the provided position + 1.
protected void	notifyWaiters(boolean all) Notifies the waiting threads that the state of the queue has changed.
protected int	offer() Checks if there is remove for another message.
boolean	offer(PendingMessage pendingMessage) Puts a message onto the queue.
protected int	offer2() Checks if there is remove for another two message.
boolean	poll(PendingMessage copyOut) Returns the next message from the queue without blocking.
void	put(Message message, Callback<Reply> replyCallback) Puts a message onto the queue.
void	put(Message message, Callback<Reply> replyCallback, Message message2, Callback<Reply> replyCallback2) Puts two messages onto the queue.
void	put(PendingMessage pendingMessage) Puts a message onto the queue.
int	size() Returns the number of messages in the queue.
void	take(PendingMessage copyOut) Returns the next message from the queue and will block waiting for a message.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

MAX_MESSAGE_ID_MASK

public static final long MAX_MESSAGE_ID_MASK

The mask for constraining the size the message id.

See Also:

Constant Field Values

MAX_SIZE

public static final int MAX_SIZE

The maximum size of the queue. This it currently 2^20 but must be at most 2^30 to ensure masking works.

See Also:

Constant Field Values

SPIN_TIME_NS

public static final long SPIN_TIME_NS

Amount of time to spin before yielding. Set to 1/100 of a millisecond.

Constructor Detail

PendingMessageQueue

public PendingMessageQueue(int size,
                   LockType lockType)

Creates a new PendingMessageQueue.

Parameters:

size - The size of the queue to create.

lockType - The lock type to use with the queue.

Method Detail

capacity

public int capacity()

Returns the size of the queue.

Returns:

The size of the queue.

drainTo

public void drainTo(List<PendingMessage> pending)

Drains the list of pending messages into the provided list.

Parameters:

pending - The list to add all of the pending messages to.

isEmpty

public boolean isEmpty()

Returns true if the queue is empty. e.g., the next take position is the read before position.

Returns:

If the queue is empty.

offer

public boolean offer(PendingMessage pendingMessage)

Puts a message onto the queue. This method will block waiting for a space to add the message.

Parameters:

pendingMessage - The message to add.

Returns:

True if the message was added, false otherwise.

poll

public boolean poll(PendingMessage copyOut)

Returns the next message from the queue without blocking.

 
 PendingMessage pm = new PendingMessage();
 if( queue.poll(pm) } {
    // Handle the message copied into pm.
 }
 
 

Parameters:

copyOut - The PendingMessage to copy the pending message into.

Returns:

True if the pending message was updated.

put

public void put(Message message,
       Callback<Reply> replyCallback)
         throws InterruptedException

Puts a message onto the queue. This method will block waiting for a space to add the message.

Parameters:

message - The message to add.

replyCallback - The callback for the message to add.

Throws:

InterruptedException - If the thread is interrupted while waiting for the message. If thrown the message will not have been enqueued.

put

public void put(Message message,
       Callback<Reply> replyCallback,
       Message message2,
       Callback<Reply> replyCallback2)
         throws InterruptedException

Puts two messages onto the queue. This method will block waiting for a space to add the messages but ensures the messages are in sequence in the queue.

Parameters:

message - The first message to add.

replyCallback - The callback for the first message to add.

message2 - The second message to add.

replyCallback2 - The callback for the second message to add.

Throws:

InterruptedException - If the thread is interrupted while waiting for the message. If thrown neither message will have been enqueued.

put

public void put(PendingMessage pendingMessage)
         throws InterruptedException

Puts a message onto the queue. This method will block waiting for a space to add the message.

Parameters:

pendingMessage - The message to add.

Throws:

InterruptedException - If the thread is interrupted while waiting for the message. If thrown the message will not have been enqueued.

size

public int size()

Returns the number of messages in the queue.

Returns:

The number of messages in the queue.

take

public void take(PendingMessage copyOut)
          throws InterruptedException

Returns the next message from the queue and will block waiting for a message.

Parameters:

copyOut - The PendingMessage to copy the pending message into.

Throws:

InterruptedException - If the thread is interrupted while waiting for the message.

increment

protected int increment(int index)

Increments the index handling roll-over.

Parameters:

index - The value to increment.

Returns:

The incremented value.

markReady

protected void markReady(int index)

Marks the position as ready by incrementing the ready position to the provided position. This method uses a spin lock assuming any other threads will increment the ready position quickly to the position just before index.

Parameters:

index - The index of the ready message.

markReady2

protected void markReady2(int index)

Marks the position and the next position as ready by incrementing the ready position to the provided position + 1. This method uses a spin lock assuming any other threads will increment the ready position quickly to the position just before index.

Parameters:

index - The index of the ready message.

notifyWaiters

protected void notifyWaiters(boolean all)

Notifies the waiting threads that the state of the queue has changed.

Parameters:

all - If true then all threads will be woken. Otherwise only a single thread is woken.

offer

protected int offer()

Checks if there is remove for another message. If so returns the index of the message to update. If not return a value less then zero.

Returns:

The position of the message that can be updated or a value of less than zero if the queue is full.

offer2

protected int offer2()

Checks if there is remove for another two message. If so returns the index of the first message to update. If not return a value less then zero.

Returns:

The position of the first message that can be updated or a value of less than zero if the queue is full.