deserialize frame lazily (#1856)

* deserialize frame lazily

* remove useless rewind

Author: wyfo

commons/zenoh-codec/src/network/mod.rs

@@ -19,7 +19,7 @@ mod request;
 mod response;
 
 use zenoh_buffers::{
-    reader::{DidntRead, Reader},
+    reader::{BacktrackableReader, DidntRead, Reader},
     writer::{DidntWrite, Writer},
 };
 use zenoh_protocol::{
@@ -104,6 +104,26 @@ where
     }
 }
 
+pub struct NetworkMessageIter<R> {
+    codec: Zenoh080Reliability,
+    reader: R,
+}
+
+impl<R> NetworkMessageIter<R> {
+    pub fn new(reliability: Reliability, reader: R) -> Self {
+        let codec = Zenoh080Reliability::new(reliability);
+        Self { codec, reader }
+    }
+}
+
+impl<R: BacktrackableReader> Iterator for NetworkMessageIter<R> {
+    type Item = NetworkMessage;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.codec.read(&mut self.reader).ok()
+    }
+}
+
 // Extensions: QoS
 impl<W, const ID: u8> WCodec<(ext::QoSType<{ ID }>, bool), &mut W> for Zenoh080
 where

commons/zenoh-codec/src/transport/frame.rs

@@ -11,23 +11,20 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <[email protected]>
 //
-use alloc::vec::Vec;
-
 use zenoh_buffers::{
     reader::{BacktrackableReader, DidntRead, Reader},
     writer::{DidntWrite, Writer},
 };
 use zenoh_protocol::{
     common::{iext, imsg},
     core::Reliability,
-    network::NetworkMessage,
     transport::{
         frame::{ext, flag, Frame, FrameHeader},
         id, TransportSn,
     },
 };
 
-use crate::{common::extension, RCodec, WCodec, Zenoh080, Zenoh080Header, Zenoh080Reliability};
+use crate::{common::extension, RCodec, WCodec, Zenoh080, Zenoh080Header};
 
 // FrameHeader
 impl<W> WCodec<&FrameHeader, &mut W> for Zenoh080
@@ -146,9 +143,7 @@ where
         self.write(&mut *writer, &header)?;
 
         // Body
-        for m in payload.iter() {
-            self.write(&mut *writer, m)?;
-        }
+        writer.write_zslice(payload)?;
 
         Ok(())
     }
@@ -175,20 +170,7 @@ where
 
     fn read(self, reader: &mut R) -> Result<Frame, Self::Error> {
         let header: FrameHeader = self.read(&mut *reader)?;
-
-        let rcode = Zenoh080Reliability::new(header.reliability);
-        let mut payload = Vec::new();
-        while reader.can_read() {
-            let mark = reader.mark();
-            let res: Result<NetworkMessage, DidntRead> = rcode.read(&mut *reader);
-            match res {
-                Ok(m) => payload.push(m),
-                Err(_) => {
-                    reader.rewind(mark);
-                    break;
-                }
-            }
-        }
+        let payload = reader.read_zslice(reader.remaining())?;
 
         Ok(Frame {
             reliability: header.reliability,

commons/zenoh-protocol/src/transport/frame.rs

@@ -11,9 +11,9 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <[email protected]>
 //
-use alloc::vec::Vec;
+use zenoh_buffers::ZSlice;
 
-use crate::{core::Reliability, network::NetworkMessage, transport::TransportSn};
+use crate::{core::Reliability, transport::TransportSn};
 
 pub mod flag {
     pub const R: u8 = 1 << 5; // 0x20 Reliable      if R==1 then the frame is reliable
@@ -72,7 +72,7 @@ pub struct Frame {
     pub reliability: Reliability,
     pub sn: TransportSn,
     pub ext_qos: ext::QoSType,
-    pub payload: Vec<NetworkMessage>,
+    pub payload: ZSlice,
 }
 
 // Extensions
@@ -93,12 +93,7 @@ impl Frame {
         let reliability = Reliability::rand();
         let sn: TransportSn = rng.gen();
         let ext_qos = ext::QoSType::rand();
-        let mut payload = vec![];
-        for _ in 0..rng.gen_range(1..4) {
-            let mut m = NetworkMessage::rand();
-            m.reliability = reliability;
-            payload.push(m);
-        }
+        let payload = ZSlice::rand(rng.gen_range(8..128));
 
         Frame {
             reliability,

io/zenoh-transport/src/common/pipeline.rs

@@ -993,10 +993,10 @@ mod tests {
         reader::{DidntRead, HasReader},
         ZBuf,
     };
-    use zenoh_codec::{RCodec, Zenoh080};
+    use zenoh_codec::{network::NetworkMessageIter, RCodec, Zenoh080};
     use zenoh_config::{QueueAllocConf, QueueAllocMode};
     use zenoh_protocol::{
-        core::{Bits, CongestionControl, Encoding, Priority},
+        core::{Bits, CongestionControl, Encoding, Priority, Reliability},
         network::{ext, Push},
         transport::{BatchSize, Fragment, Frame, TransportBody, TransportSn},
         zenoh::{PushBody, Put},
@@ -1100,8 +1100,10 @@ mod tests {
                     match res {
                         Ok(msg) => {
                             match msg.body {
-                                TransportBody::Frame(Frame { payload, .. }) => {
-                                    msgs += payload.len()
+                                TransportBody::Frame(Frame { mut payload, .. }) => {
+                                    msgs +=
+                                        NetworkMessageIter::new(Reliability::DEFAULT, &mut payload)
+                                            .count();
                                 }
                                 TransportBody::Fragment(Fragment { more, .. }) => {
                                     fragments += 1;

io/zenoh-transport/src/multicast/rx.rs

@@ -13,6 +13,7 @@
 //
 use std::sync::MutexGuard;
 
+use zenoh_codec::network::NetworkMessageIter;
 use zenoh_core::{zlock, zread};
 use zenoh_protocol::{
     core::{Locator, Priority, Reliability},
@@ -170,7 +171,7 @@ impl TransportMulticastInner {
             // Drop invalid message and continue
             return Ok(());
         }
-        for msg in payload.drain(..) {
+        for msg in NetworkMessageIter::new(reliability, &mut payload) {
             self.trigger_callback(msg, peer)?;
         }
 

io/zenoh-transport/src/unicast/universal/rx.rs

@@ -13,6 +13,7 @@
 //
 use std::sync::MutexGuard;
 
+use zenoh_codec::network::NetworkMessageIter;
 use zenoh_core::{zlock, zread};
 use zenoh_link::Link;
 use zenoh_protocol::{
@@ -103,7 +104,7 @@ impl TransportUnicastUniversal {
         }
         let callback = zread!(self.callback).clone();
         if let Some(callback) = callback.as_ref() {
-            for msg in payload.drain(..) {
+            for msg in NetworkMessageIter::new(reliability, &mut payload) {
                 self.trigger_callback(callback.as_ref(), msg)?;
             }
         } else {