Class PerDestinationBundler

java.lang.Object

org.jgroups.protocols.BaseBundler

org.jgroups.protocols.PerDestinationBundler

All Implemented Interfaces:

Runnable, Bundler

public class PerDestinationBundler
extends BaseBundler
implements Runnable

Queues messages per destination ('null' is a special destination). Uses 1 thread per destination to process queued messages, so it won't scale to many cluster members (unless virtual threads are used).
See https://issues.redhat.com/browse/JGRP-2639 for details.

Since:

5.2.7

Author:

Bela Ban

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected class	PerDestinationBundler.SendBuffer

Field Summary

Fields

Modifier and Type	Field	Description
protected final Map<Address, PerDestinationBundler.SendBuffer>	dests
protected Address	local_addr
protected final Semaphore	msgs_available
protected static final Address	NULL
protected Runner	single_thread_runner
protected static final String	THREAD_NAME

Fields inherited from class BaseBundler

avg_fill_count, avg_remove_queue_size, avg_send_time, capacity, count, drop_when_full, FMT, FUNC, lock, log, max_size, members, msg_processing_policy, msg_stats, msgs, num_batches_sent, num_drops_on_full_queue, num_sends_because_full_queue, num_sends_because_no_msgs, num_single_msgs_sent, output, remove_delay, remove_queue_capacity, suppress_log, suppress_log_timeout, total_msgs_sent, transport, use_ringbuffer, use_single_sender_thread

Constructor Summary

Constructors

Constructor	Description
PerDestinationBundler()

Method Summary

Modifier and Type	Method	Description
String	active()
String	dests()
void	doSend(Message msg)
String	dump()
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
boolean	isRunning()
int	messagesAvailable()
protected void	removeLeftMembers(List<Address> left_mbrs)
void	run()	Blocks on the semaphore until messages are available, then iterates through the send buffers and sends all messages.
void	send(Message msg)
int	size()	Returns the total number of messages in the hashmap
void	start()	Called after Bundler.init(TP)
void	startSingleThreadRunner()
void	stop()
void	stopSingleThreadRunner()
void	viewChange(View view)

Methods inherited from class BaseBundler

addMessage, avgBatchSize, dropWhenFull, dropWhenFull, getCapacity, getMaxSize, loopback, loopback, loopbackUnlessDontLoopbackIsSet, printBuffers, removeDelay, removeDelay, removeQueueCapacity, removeQueueCapacity, resetStats, sendBundledMessages, sendMessageList, sendMultiple, sendSingle, sendSingleMessage, setCapacity, setMaxSize, suppressLogTimeout, suppressLogTimeout, useRingBuffer, useRingBuffer, useSingleSenderThread, useSingleSenderThread

Methods inherited from class Object

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

Methods inherited from interface Bundler

destroy, renameThread

Field Details

local_addr

protected Address local_addr

dests

protected final Map<Address, PerDestinationBundler.SendBuffer> dests

NULL

protected static final Address NULL

single_thread_runner

protected Runner single_thread_runner

THREAD_NAME

protected static final String THREAD_NAME

See Also:

• Constant Field Values

msgs_available

protected final Semaphore msgs_available

Constructor Details

PerDestinationBundler

public PerDestinationBundler()

Method Details

isRunning

public boolean isRunning()

messagesAvailable

public int messagesAvailable()

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

dump

public String dump()

active

public String active()

dests

public String dests()

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

startSingleThreadRunner

public void startSingleThreadRunner()

stopSingleThreadRunner

public void stopSingleThreadRunner()

send

public void send(Message msg)
          throws Exception

Specified by:

send in interface Bundler

Overrides:

send in class BaseBundler

Throws:

Exception

doSend

public void doSend(Message msg)
            throws Exception

Throws:

Exception

run

public void run()

Blocks on the semaphore until messages are available, then iterates through the send buffers and sends all messages. When an iteration found no messages to send, the runner calls run() again, blocking on the semaphore until new messages are available in any of the buffers. This is the single_sender_thread (use_single_sender_thread=true)

Specified by:

run in interface Runnable

viewChange

public void viewChange(View view)

Specified by:

viewChange in interface Bundler

Overrides:

viewChange in class BaseBundler

removeLeftMembers

protected void removeLeftMembers(List<Address> left_mbrs)