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
Modifier and Type Class and Description

protected class FlowControl.Credit

Nested Classes
Modifier and Type	Class and Description
`protected class`	`FlowControl.Credit`

Field Summary

Fields
Modifier and Type	Field and Description
`protected int`	`frag_size`
`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 Address`	`local_addr`
`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 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
after_creation_hook, down_prot, ergonomics, id, log, name, stack, stats, up_prot

Constructor Summary

Constructors
Constructor and Description

FlowControl()

Constructors
Constructor and Description
`FlowControl()`

Method Summary

All Methods Static Methods Instance Methods Abstract Methods Concrete Methods
Modifier and Type	Method and Description
`protected long`	`adjustCredit(java.util.Map<Address,FlowControl.Credit> map, Address sender, int length)` Check whether sender has enough credits left.
`protected static long`	`bufferToLong(byte[] buf, int offset)`
`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()`
`abstract double`	`getAverageTimeBlocked()`
`protected abstract Header`	`getCreditRequestHeader()`
`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()`
`protected abstract Header`	`getReplenishHeader()`
`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`	`handleUpEvent(Message msg, FcHeader hdr)`
`protected void`	`handleViewChange(java.util.List<Address> mbrs)`
`void`	`init()` Called after instance has been created (null constructor) and before protocol is started.
`protected static byte[]`	`longToBuffer(long num)`
`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, long credits_needed)` We cannot send this request as OOB message, 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.
`void`	`up(MessageBatch batch)` Sends up a multiple messages in a `MessageBatch`.

Methods inherited from class org.jgroups.stack.Protocol
accept, afterCreationHook, destroy, enableStats, getConfigurableObjects, getDownProtocol, getDownServices, getId, getIdsAbove, getLevel, getLog, getName, getProtocolStack, getSocketFactory, getThreadFactory, getTransport, getUpProtocol, getUpServices, getValue, isErgonomics, level, parse, 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
  - 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.
  - local_addr
```
protected Address local_addr
```
  - running
```
protected volatile boolean running
```
    Whether FlowControl is still running, this is set to false when the protocol terminates (on stop())
  - frag_size
```
protected int frag_size
```
- 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)
```
  - getMaxBlockTime
```
public long getMaxBlockTime()
```
  - setMaxBlockTime
```
public void setMaxBlockTime(long t)
```
  - setMaxBlockTimes
```
public void setMaxBlockTimes(java.lang.String str)
```
  - getMaxBlockTimes
```
public java.lang.String getMaxBlockTimes()
```
  - getNumberOfBlockings
```
public abstract int getNumberOfBlockings()
```
  - getAverageTimeBlocked
```
public abstract 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)
```
  - getReplenishHeader
```
protected abstract Header getReplenishHeader()
```
  - getCreditRequestHeader
```
protected abstract Header getCreditRequestHeader()
```
  - 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
  - handleUpEvent
```
protected void handleUpEvent(Message msg,
                             FcHeader hdr)
```
  - up
```
public void up(MessageBatch batch)
```
    Description copied from class: Protocol
    
    Sends up a multiple messages in a MessageBatch. The sender of the batch is always the same, and so is the destination (null == multicast messages). Messages in a batch can be OOB messages, regular messages, or mixed messages, although the transport itself will create initial MessageBatches that contain only either OOB or regular messages.
    The default processing below sends messages up the stack individually, based on a matching criteria (calling Protocol.accept(org.jgroups.Message)), and - if true - calls Protocol.up(org.jgroups.Event) for that message and removes the message. If the batch is not empty, it is passed up, or else it is dropped.
    Subclasses should check if there are any messages destined for them (e.g. using MessageBatch.getMatchingMessages(short,boolean)), then possibly remove and process them and finally pass the batch up to the next protocol. Protocols can also modify messages in place, e.g. ENCRYPT could decrypt all encrypted messages in the batch, not remove them, and pass the batch up when done.
    
    Overrides:
    
    up in class Protocol
    
    Parameters:
    
    batch - The message batch
  - 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,
                                 long credits_needed)
```
    We cannot send this request as OOB message, 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)
```
  - longToBuffer
```
protected static byte[] longToBuffer(long num)
```
  - bufferToLong
```
protected static long bufferToLong(byte[] buf,
                                   int offset)
```

Class FlowControl

Nested Class Summary

Field Summary

Fields inherited from class org.jgroups.stack.Protocol

Constructor Summary

Method Summary

Methods inherited from class org.jgroups.stack.Protocol

Methods inherited from class java.lang.Object

Field Detail

max_credits

max_block_time

max_block_times

min_threshold

min_credits

ignore_synchronous_response

num_credit_requests_received

num_credit_requests_sent

num_credit_responses_sent

num_credit_responses_received

received

local_addr

running

frag_size

Constructor Detail

FlowControl

Method Detail

resetStats

getMaxCredits

setMaxCredits

getMinThreshold

setMinThreshold

getMinCredits

setMinCredits

getMaxBlockTime

setMaxBlockTime

setMaxBlockTimes

getMaxBlockTimes

getNumberOfBlockings

getAverageTimeBlocked

getNumberOfCreditRequestsReceived

getNumberOfCreditRequestsSent

getNumberOfCreditResponsesReceived

getNumberOfCreditResponsesSent

printSenderCredits

printReceiverCredits

printCredits

dumpStats

getMaxBlockTime

handleMulticastMessage

handleCredit

getReplenishHeader

getCreditRequestHeader

unblock

init

start

stop

down

up

handleUpEvent

up

handleConfigEvent

handleDownMessage

adjustCredit

handleCreditRequest

sendCredit

sendCreditRequest

handleViewChange

printMap

longToBuffer

bufferToLong