Flow control

[andsel]

General description

Flow control permit to the client and server to define how many PUBLISH messages with QoS > 0 they could process concurrently per connection. During CONNECT the client uses the receive maximum property to communicate to the server how many publishes is can manage concurrently. On the CONNACK the server uses the same property to specify to the client how many publishes the client can send concurrently to the server. If the limit is passed by any of the peers, the offended peer DISCONNECT with reason code 0x93 (Receive Maximum exceeded).

MQTT5 specification

4.9

Clients and Servers control the number of unacknowledged PUBLISH packets they receive by using a Receive Maximum value as described in section 3.1.2.11.4 and section 3.2.2.3.2. The Receive Maximum establishes a send quota which is used to limit the number of PUBLISH QOS > 0 packets which can be sent without receiving an PUBACK (for QoS 1) or PUBCOMP (for QoS 2). The PUBACK and PUBCOMP replenish the quota in the manner described below.

The Client or Server MUST set its initial send quota to a non-zero value not exceeding the Receive Maximum [MQTT-4.9.0-1].

Each time the Client or Server sends a PUBLISH packet at QoS > 0, it decrements the send quota. If the send quota reaches zero, the Client or Server MUST NOT send any more PUBLISH packets with QoS > 0 [MQTT-4.9.0-2]. It MAY continue to send PUBLISH packets with QoS 0, or it MAY choose to suspend sending these as well. The Client and Server MUST continue to process and respond to all other MQTT Control Packets even if the quota is zero [MQTT-4.9.0-3].

The send quota is incremented by 1:

• Each time a PUBACK or PUBCOMP packet is received, regardless of whether the PUBACK or PUBCOMP carried an error code.
• Each time a PUBREC packet is received with a Return Code of 0x80 or greater.

The send quota is not incremented if it is already equal to the initial send quota. The attempt to increment above the initial send quota might be caused by the re-transmission of a PUBREL packet after a new Network Connection is established.

Refer to section 3.3.4 for a description of how Clients and Servers react if they are sent more PUBLISH packets than the Receive Maximum allows.

The send quota and Receive Maximum value are not preserved across Network Connections, and are re-initialized with each new Network Connection as described above. They are not part of the session state.

3.1.2.11.4

The Client uses this value to limit the number of QoS 1 and QoS 2 publications that it is willing to process concurrently. There is no mechanism to limit the QoS 0 publications that the Server might try to send.

The value of Receive Maximum applies only to the current Network Connection. If the Receive Maximum value is absent then its value defaults to 65535.

3.2.2.3.2

The Server uses this value to limit the number of QoS 1 and QoS 2 publications that it is willing to process concurrently for the Client. It does not provide a mechanism to limit the QoS 0 publications that the Client might try to send.

If the Receive Maximum value is absent, then its value defaults to 65535.

3.3.4

....
The Client MUST NOT send more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets for which it has not received PUBACK, PUBCOMP, or PUBREC with a Reason Code of 128 or greater from the Server [MQTT-3.3.4-7] f it receives more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets where it has not sent a PUBACK or PUBCOMP in response, the Server uses a DISCONNECT packet with Reason Code 0x93 (Receive Maximum exceeded)
...
The Client MUST NOT delay the sending of any packets other than PUBLISH packets due to having sent Receive Maximum PUBLISH packets without receiving acknowledgements for them [MQTT-3.3.4-8] The value of Receive Maximum applies only to the current Network Connection.
...
The Server MUST NOT send more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets for which it has not received PUBACK, PUBCOMP, or PUBREC with a Reason Code of 128 or greater from the Client [MQTT-3.3.4-9]. f it receives more than Receive Maximum QoS 1 and QoS 2 PUBLISH packets where it has not sent a PUBACK or PUBCOMP in response, the Client uses DISCONNECT with Reason Code 0x93 (Receive Maximum exceeded)
...
The Server MUST NOT delay the sending of any packets other than PUBLISH packets due to having sent Receive Maximum PUBLISH packets without receiving acknowledgements for them [MQTT-3.3.4-10].

Implementation

Status

As of today Moquette has a fixed window for output inflight publish messages:

• defines a constant

  moquette/broker/src/main/java/io/moquette/BrokerConstants.java

  Line 133 in 38428f6

  public static final int INFLIGHT_WINDOW_SIZE = 10;

• return in CONNACK's receive maximum

  moquette/broker/src/main/java/io/moquette/broker/MQTTConnection.java

  Line 369 in 38428f6

  .receiveMaximum(INFLIGHT_WINDOW_SIZE)

• use this value as counter limit for inflight in output

  moquette/broker/src/main/java/io/moquette/broker/Session.java

  Line 90 in 38428f6

  private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable

Problems

The value is hardcoded, can't be defined as a default in the broker configs, it's used to respond to the client to let it know how many concurrent publishes it process but in reality is used to limit the flow in output.
There isn't any control con the publishes in input.

Task list

• read the receive-maximum property from CONNECT and use that in the broker output flow to that client, if not specified use the INFLIGHT_WINDOW_SIZE. Handled client's receive maximum property to avoid publish flooding #858
• verify that the counter is incremented and decremented accordingly to the specification. Fixed by Implements management of receive_maximum parallel publishes the broekr can handle per client #852
• implement a control mechanism to limit the concurrent publishes received from the client, by now use a constant. Once surpassed DISCONNECT with proper reason code, logging the action. Fixed by Implements management of receive_maximum parallel publishes the broekr can handle per client #852.
• make the incoming receive maximum configurable by settings, updating also the fluent config. Fixed by Implements management of receive_maximum parallel publishes the broekr can handle per client #852