org.jgroups.protocols

Class RingBufferBundler

java.lang.Object

org.jgroups.protocols.BaseBundler

org.jgroups.protocols.RingBufferBundler

All Implemented Interfaces:

Bundler

public class RingBufferBundler
extends BaseBundler

Bundler which uses RingBuffer to store messages. The difference to TransferQueueBundler is that RingBuffer uses a wait strategy (to for example spinning) before blocking. Also, the hashmap of the superclass is not used, but the array of the RingBuffer is used directly to bundle and send messages, minimizing memory allocation.

Field Summary

Fields

Modifier and Type	Field and Description
protected Runner	bundler_thread
protected int	capacity
protected int	num_spins
protected static BiConsumer<java.lang.Integer,java.lang.Integer>	PARK
protected RingBuffer<Message>	rb
protected java.lang.Runnable	run_function
protected static BiConsumer<java.lang.Integer,java.lang.Integer>	SPIN
protected static BiConsumer<java.lang.Integer,java.lang.Integer>	SPIN_PARK
protected static BiConsumer<java.lang.Integer,java.lang.Integer>	SPIN_YIELD
protected java.lang.Runnable	stop_function
protected static java.lang.String	THREAD_NAME
protected BiConsumer<java.lang.Integer,java.lang.Integer>	wait_strategy
protected static BiConsumer<java.lang.Integer,java.lang.Integer>	YIELD

Fields inherited from class org.jgroups.protocols.BaseBundler

count, lock, log, msgs, output, transport

Constructor Summary

Constructors

Modifier	Constructor and Description
	RingBufferBundler()
	RingBufferBundler(int capacity)
protected	RingBufferBundler(RingBuffer<Message> rb)

Method Summary

Modifier and Type	Method and Description
protected int	advance(int index)
protected static int	assertPositive(int value, java.lang.String message)
RingBuffer<Message>	buf()
protected BiConsumer<java.lang.Integer,java.lang.Integer>	createWaitStrategy(java.lang.String st, BiConsumer<java.lang.Integer,java.lang.Integer> default_wait_strategy)
java.lang.Thread	getThread()
protected int	index(int idx)
void	init(TP transport) Called after creation of the bundler
protected int	marshalMessagesToSameDestination(Address dest, Message[] buf, int start_index, int end_index, int max_bundle_size)
int	numSpins()
RingBufferBundler	numSpins(int n)
protected static java.lang.String	print(BiConsumer<java.lang.Integer,java.lang.Integer> wait_strategy)
protected void	readMessages()
void	send(Message msg)
void	sendBundledMessages(Message[] buf, int read_index, int available_msgs) Read and send messages in range [read-index ..
int	size() The number of unsent messages in the bundler
void	start() Called after Bundler.init(TP)
void	stop()
java.lang.String	waitStrategy()
RingBufferBundler	waitStrategy(java.lang.String st)

Methods inherited from class org.jgroups.protocols.BaseBundler

addMessage, checkForSharedTransport, clearMessages, sendBundledMessages, sendMessageList, sendSingleMessage, viewChange

Methods inherited from class java.lang.Object

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

Field Detail

rb

protected RingBuffer<Message> rb

bundler_thread

protected Runner bundler_thread

num_spins

protected int num_spins

THREAD_NAME

protected static final java.lang.String THREAD_NAME

See Also:

Constant Field Values

wait_strategy

protected BiConsumer<java.lang.Integer,java.lang.Integer> wait_strategy

capacity

protected int capacity

run_function

protected final java.lang.Runnable run_function

stop_function

protected final java.lang.Runnable stop_function

SPIN

protected static final BiConsumer<java.lang.Integer,java.lang.Integer> SPIN

YIELD

protected static final BiConsumer<java.lang.Integer,java.lang.Integer> YIELD

PARK

protected static final BiConsumer<java.lang.Integer,java.lang.Integer> PARK

SPIN_PARK

protected static final BiConsumer<java.lang.Integer,java.lang.Integer> SPIN_PARK

SPIN_YIELD

protected static final BiConsumer<java.lang.Integer,java.lang.Integer> SPIN_YIELD

Constructor Detail

RingBufferBundler

public RingBufferBundler()

RingBufferBundler

protected RingBufferBundler(RingBuffer<Message> rb)

RingBufferBundler

public RingBufferBundler(int capacity)

Method Detail

buf

public RingBuffer<Message> buf()

getThread

public java.lang.Thread getThread()

size

public int size()

Description copied from interface: Bundler

The number of unsent messages in the bundler

Specified by:

size in interface Bundler

Overrides:

size in class BaseBundler

numSpins

public int numSpins()

numSpins

public RingBufferBundler numSpins(int n)

waitStrategy

public java.lang.String waitStrategy()

waitStrategy

public RingBufferBundler waitStrategy(java.lang.String st)

init

public void init(TP transport)

Description copied from interface: Bundler

Called after creation of the bundler

Specified by:

init in interface Bundler

Overrides:

init in class BaseBundler

Parameters:

transport - the transport, for further reference

start

public void start()

Description copied from interface: Bundler

Called after Bundler.init(TP)

Specified by:

start in interface Bundler

Overrides:

start in class BaseBundler

stop

public void stop()

Specified by:

stop in interface Bundler

Overrides:

stop in class BaseBundler

send

public void send(Message msg)
          throws java.lang.Exception

Specified by:

send in interface Bundler

Overrides:

send in class BaseBundler

Throws:

java.lang.Exception

sendBundledMessages

public void sendBundledMessages(Message[] buf,
                                int read_index,
                                int available_msgs)

Read and send messages in range [read-index .. read-index+available_msgs-1]

marshalMessagesToSameDestination

protected int marshalMessagesToSameDestination(Address dest,
                                               Message[] buf,
                                               int start_index,
                                               int end_index,
                                               int max_bundle_size)
                                        throws java.lang.Exception

Throws:

java.lang.Exception

readMessages

protected void readMessages()

advance

protected final int advance(int index)

index

protected final int index(int idx)

print

protected static java.lang.String print(BiConsumer<java.lang.Integer,java.lang.Integer> wait_strategy)

createWaitStrategy

protected BiConsumer<java.lang.Integer,java.lang.Integer> createWaitStrategy(java.lang.String st,
                                                                             BiConsumer<java.lang.Integer,java.lang.Integer> default_wait_strategy)

assertPositive

protected static int assertPositive(int value,
                                    java.lang.String message)