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

Nested Classes
`protected class` `FlowControl.Credit`

### Field Summary

Fields
`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

Constructors
Constructor and Description
`FlowControl()`

### Method Summary

Methods
`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.Vector<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, downThreadEnabled, enableStats, getConfigurableObjects, getDownProtocol, getId, getLevel, getName, getProperties, getProtocolStack, getSocketFactory, getThreadFactory, getTransport, getUpProtocol, getValue, isErgonomics, printStats, providedDownServices, providedUpServices, requiredDownServices, requiredUpServices, setDownProtocol, setErgonomics, setId, setLevel, setProperties, setPropertiesInternal, setProperty, setProtocolStack, setSocketFactory, setUpProtocol, setValue, setValues, statsEnabled, upThreadEnabled`

### 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.Vector<Address> mbrs)

#### printMap

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