feat: buffer `Framed<.., Codec>` packets #826

de-sh · 2024-03-20T15:55:32Z

resolves #810 on top of #825

Defaults to buffering 10 packets before each flush, behavior can be changed with MqttOptions.set_max_request_batch(). Instantly flushes outgoing PingReq as well(required to not close the connection).

Type of change

Checklist:

Formatted with cargo fmt
Make an entry to CHANGELOG.md if it's relevant to the users of the library. If it's not relevant mention why.

* refactor: implement size for all v5 packet types * refactor: implement `Packet.size()` for v4

* refactor: `Packet::read` * refactor: `Packet::write` * test: fix changes in refactor

Defaults to 10

flxo

Thanks for the update!

I don't understand why there's homebrewn counting of the outgoing packets. The counter also includes packets generated in response to incoming packets, so the name of the configuration is misleading.

The manual count + flush also works against the capacity of Framed (which defaults here). Wouldn't it be better to leave the need to a implicit flush to Framed? Flushing must still happen after each batch of incoming packets and requests.

I totally agree with wrapping of SinkExt::flush with the timeout.

My suggestion is:

rename max_request_batch to io_buffer_capacity. Apply that when Framed is constructed.
Remove Network::write and use SinkExt::feed instead (or write a wrapper if you wan't to hide theSink)
Keep Network::flush with the timeout feature.

cheers

@flxo

flxo · 2024-03-21T12:16:18Z

rumqttc/src/eventloop.rs

@@ -185,7 +185,7 @@ impl EventLoop {
            o = network.read() => {
                let incoming = o?;
                if let Some(packet) = self.state.handle_incoming_packet(incoming)? {
-                    network.send(packet).await?;
+                    network.write(packet).await?;


This packet is never flushed unless there's a ping or options.max_request_batch - 1 more packets "sent".

Had the same doubt, was using tests as a crutch, but I see how even the tests aren't complete.

What type of acks should we instantly respond to, I guess only PingResp?

I'm not so deep in the MQTT standard but think it's just PingResp.

de-sh · 2024-03-21T21:30:13Z

Wouldn't it be better to leave the need to a implicit flush to Framed?

How does Framed do implicit flushing again? just calling. I have validated that using framed.feed() doesn't flush on breaching buffer, what could I be doing wrong?

Remove Network::write and use SinkExt::feed instead (or write a wrapper if you wan't to hide theSink)

I would like to keep write and wrap the internal workings for the time being, mainly because this isn't directly exposed to the user and readability of code is important for maintenance.

flxo

Left some comments. My suggestion is to move the "when to flush" logic to EventLoop. See comments.

flxo · 2024-03-22T07:11:22Z

rumqttc/src/eventloop.rs

@@ -185,7 +185,7 @@ impl EventLoop {
            o = network.read() => {
                let incoming = o?;
                if let Some(packet) = self.state.handle_incoming_packet(incoming)? {
-                    network.send(packet).await?;
+                    network.write(packet).await?;


I'm not so deep in the MQTT standard but think it's just PingResp.

flxo · 2024-03-22T07:30:50Z

rumqttc/src/framed.rs

+            .await
+            .map_err(StateError::Deserialization)?;
+
+        if should_flush || self.framed.write_buffer().len() + packet_size >= self.buffer_capacity {


FramedWrite does that job for you here. I think removing the manual buffer size check is good and avoid not necessary complexity here.

I'd also do the "when to flush" logic in EventLoop and not in here because it must reflect MQTT and the IO part should be agnostic.

Add a loop (up to e.g 10) around the select! that breaks if pending or request_rx is empty and the rx buffer of Framed is empty after a fill_buf. There is no need for extra logic in Network.

like this:

for _ in 0..10 { select! { r = next_request() => { packet = state.process(r) network.feed(packet) } p = network.read() => { packet = state.handle_incoming(p) network.feed(packet) if matches(packet, PingResp) { // Ensure the flush by breaking. break; } } _ = ping_interval.tick() => { packet = PingReq network.feed(packet) } } let data_pending = network.fill_rx_buf(); // get inner of framed; get read_buffer_mut, fill it. // Break if there is nothing to do. if pending.is_empty() && requests_rx.is_empty() && !data_pending() { break; } } network.flush()

flxo · 2024-03-22T07:31:50Z

rumqttc/src/v5/framed.rs

+    pub async fn write(&mut self, packet: Packet) -> Result<(), StateError> {
+        let packet_size = packet.size();
+        let should_flush = match packet {
+            Packet::Connect(..) | Packet::PingReq(_) | Packet::PingResp(_) => true,


Same as above.

de-sh · 2024-03-25T10:24:27Z

closed in favor of #823

Devdutt Shenoi added 17 commits March 16, 2024 20:14

feat: implement Packet::size for v4/v5 (#817)

719633a

* refactor: implement size for all v5 packet types * refactor: implement `Packet.size()` for v4

refactor: read and write methods on v4::Packet (#821)

cfdd394

* refactor: `Packet::read` * refactor: `Packet::write` * test: fix changes in refactor

refactor: AsyncReadWrite (#822)

8b4f96d

feat: MQTT Codec decoder

abf416c

feat: MQTT Codec encoder

0842f73

refactor: move write buffer into Network

036e1f2

refactor: testing improvements

8a220c3

fix: readb should block for 1 packet (#824)

a5b2717

fix: Network::connect should flush

303d8c2

test: fix network expectations for EventLoop::clean

7fee5c1

refactor: use Framed to encode/decode

75efeaa

deprecate Network::readb

041c6c8

refactor: v5 implementation

5dbe5a1

doc: changelog entry

c861a95

allow configuring MqttOptions.max_request_batch

6c9a8d9

Defaults to 10

feat: buffer packets

33abd75

test: fix by avoiding buffer

8b0aaa1

de-sh mentioned this pull request Mar 20, 2024

refactor: simplify Networkwith Framed<.., Codec> #825

Merged

2 tasks

de-sh changed the title ~~feat: buffer packets~~ feat: buffer Framed<.., Codec> packets Mar 20, 2024

flxo reviewed Mar 21, 2024

View reviewed changes

Devdutt Shenoi added 3 commits March 21, 2024 17:05

fix: network buffering based on bytes written

f71665b

fix: don't return Error::InsufficientBytes

83ed8c4

fix: buffer til capacity breach

aa08d34

flxo reviewed Mar 22, 2024

View reviewed changes

de-sh force-pushed the codec branch from c861a95 to 200f865 Compare March 25, 2024 07:11

Base automatically changed from codec to main March 25, 2024 10:02

de-sh closed this Mar 25, 2024

de-sh deleted the buffers branch July 31, 2024 06:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: buffer `Framed<.., Codec>` packets #826

feat: buffer `Framed<.., Codec>` packets #826

de-sh commented Mar 20, 2024 •

edited

Loading

flxo left a comment

flxo Mar 21, 2024

de-sh Mar 21, 2024

flxo Mar 22, 2024

de-sh commented Mar 21, 2024 •

edited

Loading

flxo left a comment

flxo Mar 22, 2024

flxo Mar 22, 2024

flxo Mar 22, 2024

de-sh commented Mar 25, 2024

feat: buffer Framed<.., Codec> packets #826

feat: buffer Framed<.., Codec> packets #826

Conversation

de-sh commented Mar 20, 2024 • edited Loading

Type of change

Checklist:

flxo left a comment

Choose a reason for hiding this comment

flxo Mar 21, 2024

Choose a reason for hiding this comment

de-sh Mar 21, 2024

Choose a reason for hiding this comment

flxo Mar 22, 2024

Choose a reason for hiding this comment

de-sh commented Mar 21, 2024 • edited Loading

flxo left a comment

Choose a reason for hiding this comment

flxo Mar 22, 2024

Choose a reason for hiding this comment

flxo Mar 22, 2024

Choose a reason for hiding this comment

flxo Mar 22, 2024

Choose a reason for hiding this comment

de-sh commented Mar 25, 2024

feat: buffer `Framed<.., Codec>` packets #826

feat: buffer `Framed<.., Codec>` packets #826

de-sh commented Mar 20, 2024 •

edited

Loading

de-sh commented Mar 21, 2024 •

edited

Loading