org.jgroups.protocols
Class FlowControl

java.lang.Object
  org.jgroups.stack.Protocol
      org.jgroups.protocols.FlowControl

Direct Known Subclasses:

MFC, UFC

public abstract class FlowControl
extends Protocol

Simple flow control protocol based on a credit system. Each sender has a number of credits (bytes to send). When the credits have been exhausted, the sender blocks. Each receiver also keeps track of how many credits it has received from a sender. When credits for a sender fall below a threshold, the receiver sends more credits to the sender.

Author:

Bela Ban

Nested Class Summary

protected class

FlowControl.Credit

Field Summary

protected static FcHeader	CREDIT_REQUEST_HDR
protected boolean	frag_size_received
protected boolean	ignore_synchronous_response Whether an up thread that comes back down should be allowed to bypass blocking if all credits are exhausted.
protected java.lang.ThreadLocal<java.lang.Boolean>	ignore_thread Thread that carries messages through up() and shouldn't be blocked in down() if ignore_synchronous_response==true.
protected long	max_block_time Max time (in milliseconds) to block.
protected java.util.Map<java.lang.Long,java.lang.Long>	max_block_times Defines the max number of milliseconds for a message to block before being sent, based on the length of the message.
protected long	max_credits Max number of bytes to send per receiver until an ack must be received before continuing sending
protected long	min_credits Computed as max_credits times min_theshold.
protected double	min_threshold If we're down to (min_threshold * max_credits) bytes for P, we send more credits to P.
protected int	num_credit_requests_received
protected int	num_credit_requests_sent
protected int	num_credit_responses_received
protected int	num_credit_responses_sent
protected java.util.Map<Address,FlowControl.Credit>	received Keeps track of credits per member at the receiver.
protected static FcHeader	REPLENISH_HDR
protected boolean	running Whether FlowControl is still running, this is set to false when the protocol terminates (on stop())

Fields inherited from class org.jgroups.stack.Protocol

down_prot, ergonomics, id, log, name, stack, stats, up_prot

Constructor Summary

FlowControl()

Method Summary

protected long	adjustCredit(java.util.Map<Address,FlowControl.Credit> map, Address sender, int length) Check whether sender has enough credits left.
java.lang.Object	down(Event evt) An event is to be sent down the stack.
java.util.Map<java.lang.String,java.lang.Object>	dumpStats()
double	getAverageTimeBlocked()
long	getMaxBlockTime()
protected long	getMaxBlockTime(long length)
java.lang.String	getMaxBlockTimes()
long	getMaxCredits()
long	getMinCredits()
double	getMinThreshold()
abstract int	getNumberOfBlockings()
int	getNumberOfCreditRequestsReceived()
int	getNumberOfCreditRequestsSent()
int	getNumberOfCreditResponsesReceived()
int	getNumberOfCreditResponsesSent()
abstract long	getTotalTimeBlocked()
protected void	handleConfigEvent(java.util.Map<java.lang.String,java.lang.Object> info)
protected abstract void	handleCredit(Address sender, long increase)
protected void	handleCreditRequest(java.util.Map<Address,FlowControl.Credit> map, Address sender, long requested_credits)
protected abstract java.lang.Object	handleDownMessage(Event evt, Message msg, Address dest, int length)
protected abstract boolean	handleMulticastMessage() Whether the protocol handles message with dest == null || dest.isMulticastAddress()
protected void	handleViewChange(java.util.List<Address> mbrs)
void	init() Called after instance has been created (null constructor) and before protocol is started.
java.lang.String	printCredits()
protected static java.lang.String	printMap(java.util.Map<Address,FlowControl.Credit> m)
java.lang.String	printReceiverCredits()
abstract java.lang.String	printSenderCredits()
void	resetStats()
protected void	sendCredit(Address dest, long credits)
protected void	sendCreditRequest(Address dest, java.lang.Long credits_needed) We cannot send this request as OOB messages, as the credit request needs to queue up behind the regular messages; if a receiver cannot process the regular messages, that is a sign that the sender should be throttled !
void	setMaxBlockTime(long t)
void	setMaxBlockTimes(java.lang.String str)
void	setMaxCredits(long max_credits)
void	setMinCredits(long min_credits)
void	setMinThreshold(double min_threshold)
void	start() This method is called on a Channel.connect(String).
void	stop() This method is called on a Channel.disconnect().
void	unblock() Allows to unblock all blocked senders from an external program, e.g.
java.lang.Object	up(Event evt) An event was received from the layer below.

Methods inherited from class org.jgroups.stack.Protocol

destroy, enableStats, getConfigurableObjects, getDownProtocol, getDownServices, getId, getIdsAbove, getLevel, getName, getProtocolStack, getSocketFactory, getThreadFactory, getTransport, getUpProtocol, getUpServices, getValue, isErgonomics, printStats, providedDownServices, providedUpServices, requiredDownServices, requiredUpServices, resetStatistics, setDownProtocol, setErgonomics, setId, setLevel, setProtocolStack, setSocketFactory, setUpProtocol, setValue, setValues, statsEnabled

Methods inherited from class java.lang.Object

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

Field Detail

REPLENISH_HDR

protected static final FcHeader REPLENISH_HDR

CREDIT_REQUEST_HDR

protected static final FcHeader CREDIT_REQUEST_HDR

max_credits

protected long max_credits

Max number of bytes to send per receiver until an ack must be received before continuing sending

max_block_time

protected long max_block_time

Max time (in milliseconds) to block. If credit hasn't been received after max_block_time, we send a REPLENISHMENT request to the members from which we expect credits. A value <= 0 means to wait forever.

max_block_times

protected java.util.Map<java.lang.Long,java.lang.Long> max_block_times

Defines the max number of milliseconds for a message to block before being sent, based on the length of the message. The property is defined as a comma-separated list of values (separated by ':'), where the key is the size in bytes and the value is the number of milliseconds to block. Example: max_block_times="50:1,500:3,1500:5,10000:10,100000:100". This means that messages up to 50 bytes wait 1 ms max until they get sent, messages up to 500 bytes 3 ms, and so on. If a message's length (size of the payload in bytes) is for example 15'000 bytes, FlowControl blocks it for a max of 100 ms.

min_threshold

protected double min_threshold

If we're down to (min_threshold * max_credits) bytes for P, we send more credits to P. Example: if max_credits is 1'000'000, and min_threshold 0.25, then we send ca. 250'000 credits to P once we've got only 250'000 credits left for P (we've received 750'000 bytes from P).

min_credits

protected long min_credits

Computed as max_credits times min_theshold. If explicitly set, this will override the above computation

ignore_synchronous_response

protected boolean ignore_synchronous_response

Whether an up thread that comes back down should be allowed to bypass blocking if all credits are exhausted. Avoids JGRP-465. Set to false by default in 2.5 because we have OOB messages for credit replenishments - this flag should not be set to true if the concurrent stack is used

num_credit_requests_received

protected int num_credit_requests_received

num_credit_requests_sent

protected int num_credit_requests_sent

num_credit_responses_sent

protected int num_credit_responses_sent

num_credit_responses_received

protected int num_credit_responses_received

received

protected final java.util.Map<Address,FlowControl.Credit> received

Keeps track of credits per member at the receiver. For each message, the credits for the sender are decremented by the size of the received message. When the credits fall below the threshold, we refill and send a REPLENISH message to the sender.

running

protected volatile boolean running

Whether FlowControl is still running, this is set to false when the protocol terminates (on stop())

frag_size_received

protected boolean frag_size_received

ignore_thread

protected final java.lang.ThreadLocal<java.lang.Boolean> ignore_thread

Thread that carries messages through up() and shouldn't be blocked in down() if ignore_synchronous_response==true. JGRP-465.

Constructor Detail

FlowControl

public FlowControl()

Method Detail

resetStats

public void resetStats()

Overrides:

resetStats in class Protocol

getMaxCredits

public long getMaxCredits()

setMaxCredits

public void setMaxCredits(long max_credits)

getMinThreshold

public double getMinThreshold()

setMinThreshold

public void setMinThreshold(double min_threshold)

getMinCredits

public long getMinCredits()

setMinCredits

public void setMinCredits(long min_credits)

getNumberOfBlockings

public abstract int getNumberOfBlockings()

getMaxBlockTime

public long getMaxBlockTime()

setMaxBlockTime

public void setMaxBlockTime(long t)

setMaxBlockTimes

public void setMaxBlockTimes(java.lang.String str)

getMaxBlockTimes

public java.lang.String getMaxBlockTimes()

getTotalTimeBlocked

public abstract long getTotalTimeBlocked()

getAverageTimeBlocked

public double getAverageTimeBlocked()

getNumberOfCreditRequestsReceived

public int getNumberOfCreditRequestsReceived()

getNumberOfCreditRequestsSent

public int getNumberOfCreditRequestsSent()

getNumberOfCreditResponsesReceived

public int getNumberOfCreditResponsesReceived()

getNumberOfCreditResponsesSent

public int getNumberOfCreditResponsesSent()

printSenderCredits

public abstract java.lang.String printSenderCredits()

printReceiverCredits

public java.lang.String printReceiverCredits()

printCredits

public java.lang.String printCredits()

dumpStats

public java.util.Map<java.lang.String,java.lang.Object> dumpStats()

Overrides:

dumpStats in class Protocol

getMaxBlockTime

protected long getMaxBlockTime(long length)

handleMulticastMessage

protected abstract boolean handleMulticastMessage()

Whether the protocol handles message with dest == null || dest.isMulticastAddress()

Returns:

handleCredit

protected abstract void handleCredit(Address sender,
                                     long increase)

unblock

public void unblock()

Allows to unblock all blocked senders from an external program, e.g. JMX

init

public void init()
          throws java.lang.Exception

Description copied from class: Protocol

Called after instance has been created (null constructor) and before protocol is started. Properties are already set. Other protocols are not yet connected and events cannot yet be sent.

Overrides:

init in class Protocol

Throws:

java.lang.Exception - Thrown if protocol cannot be initialized successfully. This will cause the ProtocolStack to fail, so the channel constructor will throw an exception

start

public void start()
           throws java.lang.Exception

Description copied from class: Protocol

This method is called on a Channel.connect(String). Starts work. Protocols are connected and queues are ready to receive events. Will be called from bottom to top. This call will replace the START and START_OK events.

Overrides:

start in class Protocol

Throws:

java.lang.Exception - Thrown if protocol cannot be started successfully. This will cause the ProtocolStack to fail, so Channel.connect(String) will throw an exception

stop

public void stop()

Description copied from class: Protocol

This method is called on a Channel.disconnect(). Stops work (e.g. by closing multicast socket). Will be called from top to bottom. This means that at the time of the method invocation the neighbor protocol below is still working. This method will replace the STOP, STOP_OK, CLEANUP and CLEANUP_OK events. The ProtocolStack guarantees that when this method is called all messages in the down queue will have been flushed

Overrides:

stop in class Protocol

down

public java.lang.Object down(Event evt)

Description copied from class: Protocol

An event is to be sent down the stack. The layer may want to examine its type and perform some action on it, depending on the event's type. If the event is a message MSG, then the layer may need to add a header to it (or do nothing at all) before sending it down the stack using down_prot.down(). In case of a GET_ADDRESS event (which tries to retrieve the stack's address from one of the bottom layers), the layer may need to send a new response event back up the stack using up_prot.up().

Overrides:

down in class Protocol

up

public java.lang.Object up(Event evt)

Description copied from class: Protocol

An event was received from the layer below. Usually the current layer will want to examine the event type and - depending on its type - perform some computation (e.g. removing headers from a MSG event type, or updating the internal membership list when receiving a VIEW_CHANGE event). Finally the event is either a) discarded, or b) an event is sent down the stack using down_prot.down() or c) the event (or another event) is sent up the stack using up_prot.up().

Overrides:

up in class Protocol

handleConfigEvent

protected void handleConfigEvent(java.util.Map<java.lang.String,java.lang.Object> info)

handleDownMessage

protected abstract java.lang.Object handleDownMessage(Event evt,
                                                      Message msg,
                                                      Address dest,
                                                      int length)

adjustCredit

protected long adjustCredit(java.util.Map<Address,FlowControl.Credit> map,
                            Address sender,
                            int length)

Check whether sender has enough credits left. If not, send it some more

Parameters:

map - The hashmap to use

sender - The address of the sender

length - The number of bytes received by this message. We don't care about the size of the headers for the purpose of flow control

Returns:

long Number of credits to be sent. Greater than 0 if credits needs to be sent, 0 otherwise

handleCreditRequest

protected void handleCreditRequest(java.util.Map<Address,FlowControl.Credit> map,
                                   Address sender,
                                   long requested_credits)

Parameters:

map - The map to modify

sender - The sender who requests credits

requested_credits - Number of bytes that the sender has left to send messages to us

sendCredit

protected void sendCredit(Address dest,
                          long credits)

sendCreditRequest

protected void sendCreditRequest(Address dest,
                                 java.lang.Long credits_needed)

We cannot send this request as OOB messages, as the credit request needs to queue up behind the regular messages; if a receiver cannot process the regular messages, that is a sign that the sender should be throttled !

Parameters:

dest - The member to which we send the credit request

credits_needed - The number of bytes (of credits) left for dest

handleViewChange

protected void handleViewChange(java.util.List<Address> mbrs)

printMap

protected static java.lang.String printMap(java.util.Map<Address,FlowControl.Credit> m)