Class 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
    • 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
      • 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,​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

      • getMaxCredits

        public long getMaxCredits()
      • setMaxCredits

        public void setMaxCredits​(long max_credits)
      • maxCredits

        public <T extends FlowControl> T maxCredits​(long c)
      • 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)
      • maxBlockTime

        public <T extends FlowControl> T maxBlockTime​(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()
      • getReceiverCreditsFor

        public long getReceiverCreditsFor​(Address mbr)
      • printCredits

        public java.lang.String printCredits()
      • 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 JChannel.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 JChannel.connect(String) will throw an exception
      • stop

        public void stop()
        Description copied from class: Protocol
        This method is called on a JChannel.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. A protocol 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 protocol may need to add a header to it (or do nothing at all) before sending it down the stack using down_prot.down().
        Overrides:
        down in class Protocol
      • down

        public java.lang.Object down​(Message msg)
        Description copied from class: Protocol
        A message is sent down the stack. Protocols may examine the message and do something (e.g. add a header) with it before passing it down.
        Overrides:
        down in class Protocol
      • up

        public java.lang.Object up​(Event evt)
        Description copied from class: Protocol
        An event was received from the protocol below. Usually the current protocol 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
      • up

        public java.lang.Object up​(Message msg)
        Description copied from class: Protocol
        A single message was received. Protocols may examine the message and do something (e.g. add a header) with it before passing it 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​(Message msg)
      • adjustCredit

        protected long adjustCredit​(java.util.Map<Address,​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,​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,​? extends Credit> m)
      • longToBuffer

        protected static byte[] longToBuffer​(long num)
      • bufferToLong

        protected static long bufferToLong​(byte[] buf,
                                           int offset)