Package org.jgroups.protocols

Class RemoveQueueBundler

java.lang.Object

org.jgroups.protocols.BaseBundler

org.jgroups.protocols.RemoveQueueBundler

All Implemented Interfaces:

Bundler

public class RemoveQueueBundler
extends BaseBundler

Bundler implementation which sends message batches (or single messages) as soon as the remove queue is full (or max_bundler_size would be exceeded).

Messages are removed from the main queue and processed as follows (assuming they all fit into the remove queue):

A B B C C A causes the following sends: {AA} -> {CC} -> {BB}

Note that null is also a valid destination (send-to-all).

Contrary to TransferQueueBundler, this bundler uses a RingBuffer rather than an ArrayBlockingQueue and the size of the remove queue is fixed. TransferQueueBundler increases the size of the remove queue dynamically, which leads to higher latency if the remove queue grows too much.

JIRA: https://issues.redhat.com/browse/JGRP-2171

Since:

4.0.4

Field Summary

Fields

Modifier and Type	Field	Description
protected final AverageMinMax	avg_batch_size
protected int	queue_size
protected RingBuffer<Message>	rb
protected Message[]	remove_queue
protected Runner	runner
protected static final String	THREAD_NAME

Fields inherited from class org.jgroups.protocols.BaseBundler

avg_send_time, capacity, count, lock, log, max_size, msgs, output, transport

Constructor Summary

Constructors

Constructor	Description
RemoveQueueBundler()

Method Summary

Modifier and Type	Method	Description
String	avgBatchSize()
int	getQueueSize()	If the bundler has a queue and it should be managed by a queuing discipline (like Random Early Detection), then return the number of elements in the queue, else -1.
void	init(TP transport)	Called after creation of the bundler
void	renameThread()
void	resetStats()
int	ringBufferSize()
void	run()
void	send(Message msg)
protected void	sendMessageList(Address dest, Address src, List<Message> list)
int	size()	Returns the total number of messages in the hashmap
void	start()	Called after Bundler.init(TP)
void	stop()

Methods inherited from class org.jgroups.protocols.BaseBundler

addMessage, getCapacity, getMaxSize, sendBundledMessages, sendSingleMessage, setCapacity, setMaxSize, viewChange

Methods inherited from class java.lang.Object

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

Field Details

rb

protected RingBuffer<Message> rb

runner

protected Runner runner

remove_queue

protected Message[] remove_queue

avg_batch_size

protected final AverageMinMax avg_batch_size

THREAD_NAME

protected static final String THREAD_NAME

See Also:

• Constant Field Values

queue_size

protected int queue_size

Constructor Details

RemoveQueueBundler

public RemoveQueueBundler()

Method Details

avgBatchSize

public String avgBatchSize()

ringBufferSize

public int ringBufferSize()

resetStats

public void resetStats()

Specified by:

resetStats in interface Bundler

Overrides:

resetStats in class BaseBundler

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

renameThread

public void renameThread()

send

public void send(Message msg)
          throws Exception

Specified by:

send in interface Bundler

Overrides:

send in class BaseBundler

Throws:

Exception

run

public void run()

getQueueSize

public int getQueueSize()

Description copied from interface: Bundler

If the bundler has a queue and it should be managed by a queuing discipline (like Random Early Detection), then return the number of elements in the queue, else -1. In the latter case, the queue won't be managed.

This method needs to be fast as it might get called on every message to be sent.

Specified by:

getQueueSize in interface Bundler

Overrides:

getQueueSize in class BaseBundler

size

public int size()

Description copied from class: BaseBundler

Returns the total number of messages in the hashmap

Specified by:

size in interface Bundler

Overrides:

size in class BaseBundler

sendMessageList

protected void sendMessageList(Address dest,
 Address src,
 List<Message> list)

Overrides:

sendMessageList in class BaseBundler