Merge commit '9e1c4a8680a04b34f09dd2776d07958c8aae6f92'

.github/workflows/ci.yml

@@ -86,6 +86,11 @@ jobs:
       - name: Run doctests
         run: cargo test --doc
 
+      - name: Build doc
+        run: cargo doc --no-deps --features unstable
+        env:
+          RUSTDOCFLAGS: -Dwarnings
+
       - name: Check licenses
         run: cargo deny check licenses
 

Cargo.toml

@@ -171,7 +171,7 @@ tokio-rustls = { version = "0.26.0", default-features = false }
 # tokio-vsock = see: io/zenoh-links/zenoh-link-vsock/Cargo.toml (workspaces does not support platform dependent dependencies)
 thread-priority = "1.1.0"
 typenum = "1.16.0"
-uhlc = { version = "0.7.0", default-features = false } # Default features are disabled due to usage in no_std crates
+uhlc = { version = "0.8.0", default-features = false } # Default features are disabled due to usage in no_std crates
 unwrap-infallible = "0.1.5"
 unzip-n = "0.1.2"
 url = "2.3.1"

DEFAULT_CONFIG.json5

@@ -23,9 +23,13 @@
   /// E.g. tcp/192.168.0.1:7447#iface=eth0, for connect only if the IP address is reachable via the interface eth0
   connect: {
     /// timeout waiting for all endpoints connected (0: no retry, -1: infinite timeout)
-    /// Accepts a single value or different values for router, peer and client.
+    /// Accepts a single value (e.g. timeout_ms: 0)
+    /// or different values for router, peer and client (e.g. timeout_ms: { router: -1, peer: -1, client: 0 }).
     timeout_ms: { router: -1, peer: -1, client: 0 },
 
+    /// The list of endpoints to connect to. 
+    /// Accepts a single list (e.g. endpoints: ["tcp/10.10.10.10:7447", "tcp/11.11.11.11:7447"])
+    /// or different lists for router, peer and client (e.g. endpoints: { router: ["tcp/10.10.10.10:7447"], peer: ["tcp/11.11.11.11:7447"] }).
     endpoints: [
       // "<proto>/<address>"
     ],
@@ -49,19 +53,21 @@
     },
   },
 
-  /// Which endpoints to listen on. E.g. tcp/localhost:7447.
+  /// Which endpoints to listen on. E.g. tcp/0.0.0.0:7447.
   /// By configuring the endpoints, it is possible to tell zenoh which are the endpoints that other routers,
   /// peers, or client can use to establish a zenoh session.
   /// For TCP/UDP on Linux, it is possible additionally specify the interface to be listened to:
   /// E.g. tcp/0.0.0.0:7447#iface=eth0, for listen connection only on eth0
   listen: {
     /// timeout waiting for all listen endpoints (0: no retry, -1: infinite timeout)
-    /// Accepts a single value or different values for router, peer and client.
+    /// Accepts a single value (e.g. timeout_ms: 0)
+    /// or different values for router, peer and client (e.g. timeout_ms: { router: -1, peer: -1, client: 0 }).
     timeout_ms: 0,
 
-    endpoints: [
-      // "<proto>/<address>"
-    ],
+    /// The list of endpoints to listen on. 
+    /// Accepts a single list (e.g. endpoints: ["tcp/[::]:7447", "udp/[::]:7447"])
+    /// or different lists for router, peer and client (e.g. endpoints: { router: ["tcp/[::]:7447"], peer: ["tcp/[::]:0"] }).
+    endpoints: { router: ["tcp/[::]:7447"], peer: ["tcp/[::]:0"] },
 
     /// Global listen configuration,
     /// Accepts a single value or different values for router, peer and client.
@@ -98,7 +104,8 @@
       /// The time-to-live on multicast scouting packets
       ttl: 1,
       /// Which type of Zenoh instances to automatically establish sessions with upon discovery on UDP multicast.
-      /// Accepts a single value or different values for router, peer and client.
+      /// Accepts a single value (e.g. autoconnect: "router|peer")
+      /// or different values for router, peer and client (e.g. autoconnect: { router: "", peer: "router|peer" }).
       /// Each value is bit-or-like combinations of "peer", "router" and "client".
       autoconnect: { router: "", peer: "router|peer" },
       /// Whether or not to listen for scout messages on UDP multicast and reply to them.
@@ -115,7 +122,8 @@
       /// direct connectivity with each other.
       multihop: false,
       /// Which type of Zenoh instances to automatically establish sessions with upon discovery on gossip.
-      /// Accepts a single value or different values for router, peer and client.
+      /// Accepts a single value (e.g. autoconnect: "router|peer")
+      /// or different values for router, peer and client (e.g. autoconnect: { router: "", peer: "router|peer" }).
       /// Each value is bit-or-like combinations of "peer", "router" and "client".
       autoconnect: { router: "", peer: "router|peer" },
     },
@@ -284,6 +292,8 @@
         /// Therefore, the maximum batch size is 2^16-1 (i.e. 65535).
         /// The default batch size value is the maximum batch size: 65535.
         batch_size: 65535,
+        /// Perform batching of messages if they are smaller of the batch_size
+        batching: true,
         /// Each zenoh link has a transmission queue that can be configured
         queue: {
           /// The size of each priority queue indicates the number of batches a given queue can contain.

ci/valgrind-check/src/queryable_get/bin/z_queryable_get.rs

@@ -14,7 +14,10 @@
 use std::{convert::TryFrom, time::Duration};
 
 use zenoh::{
-    config::Config, key_expr::KeyExpr, prelude::*, query::QueryTarget, selector::Selector,
+    config::Config,
+    key_expr::KeyExpr,
+    prelude::*,
+    query::{QueryTarget, Selector},
 };
 
 #[tokio::main]

commons/zenoh-buffers/src/bbuf.rs

@@ -127,15 +127,16 @@ impl Writer for &mut BBuf {
         self.capacity() - self.len()
     }
 
-    fn with_slot<F>(&mut self, len: usize, f: F) -> Result<NonZeroUsize, DidntWrite>
+    unsafe fn with_slot<F>(&mut self, len: usize, write: F) -> Result<NonZeroUsize, DidntWrite>
     where
         F: FnOnce(&mut [u8]) -> usize,
     {
         if self.remaining() < len {
             return Err(DidntWrite);
         }
 
-        let written = f(self.as_writable_slice());
+        // SAFETY: self.remaining() >= len
+        let written = write(unsafe { self.as_writable_slice().get_unchecked_mut(..len) });
         self.len += written;
 
         NonZeroUsize::new(written).ok_or(DidntWrite)

commons/zenoh-buffers/src/lib.rs

@@ -137,9 +137,14 @@ pub mod writer {
         fn can_write(&self) -> bool {
             self.remaining() != 0
         }
-        /// Provides a buffer of exactly `len` uninitialized bytes to `f` to allow in-place writing.
-        /// `f` must return the number of bytes it actually wrote.
-        fn with_slot<F>(&mut self, len: usize, f: F) -> Result<NonZeroUsize, DidntWrite>
+        /// Provides a buffer of exactly `len` uninitialized bytes to `write` to allow in-place writing.
+        /// `write` must return the number of bytes it actually wrote.
+        ///
+        /// # Safety
+        ///
+        /// Caller must ensure that `write` return an integer lesser than or equal to the length of
+        /// the slice passed in argument
+        unsafe fn with_slot<F>(&mut self, len: usize, write: F) -> Result<NonZeroUsize, DidntWrite>
         where
             F: FnOnce(&mut [u8]) -> usize;
     }

commons/zenoh-buffers/src/slice.rs

@@ -61,71 +61,39 @@ impl HasWriter for &mut [u8] {
 
 impl Writer for &mut [u8] {
     fn write(&mut self, bytes: &[u8]) -> Result<NonZeroUsize, DidntWrite> {
-        let len = bytes.len().min(self.len());
-        if len == 0 {
+        let Some(len) = NonZeroUsize::new(bytes.len().min(self.len())) else {
             return Err(DidntWrite);
-        }
-
-        // SAFETY: len is guaranteed to be the minimum between lhs and rhs length.
-        //         We early return if length is 0.
-        let lhs = crate::unsafe_slice_mut!(self, ..len);
-        let rhs = crate::unsafe_slice!(bytes, ..len);
-        lhs.copy_from_slice(rhs);
-
-        // SAFETY: len is guaranteed to be the minimum between lhs and rhs length.
-        let lhs = crate::unsafe_slice_mut!(self, len..);
-        // SAFETY: this doesn't compile with simple assignment because the compiler
-        // doesn't believe that the subslice has the same lifetime as the original slice,
-        // so we transmute to assure it that it does.
-        *self = unsafe { mem::transmute::<&mut [u8], &mut [u8]>(lhs) };
-
-        // SAFETY: this operation is safe since we check if len is non-zero.
-        Ok(unsafe { NonZeroUsize::new_unchecked(len) })
+        };
+        let (to_write, remain) = mem::take(self).split_at_mut(len.get());
+        to_write.copy_from_slice(&bytes[..len.get()]);
+        *self = remain;
+        Ok(len)
     }
 
     fn write_exact(&mut self, bytes: &[u8]) -> Result<(), DidntWrite> {
         let len = bytes.len();
         if self.len() < len {
             return Err(DidntWrite);
         }
-
-        // SAFETY: len is guaranteed to be the smaller than lhs length.
-        let lhs = crate::unsafe_slice_mut!(self, ..len);
-        let rhs = crate::unsafe_slice!(bytes, ..len);
-        lhs.copy_from_slice(rhs);
-
-        // SAFETY: len is guaranteed to be the minimum between lhs and rhs length.
-        let lhs = crate::unsafe_slice_mut!(self, len..);
-        // SAFETY: this doesn't compile with simple assignment because the compiler
-        // doesn't believe that the subslice has the same lifetime as the original slice,
-        // so we transmute to assure it that it does.
-        *self = unsafe { mem::transmute::<&mut [u8], &mut [u8]>(lhs) };
-
+        let _ = self.write(bytes);
         Ok(())
     }
 
     fn remaining(&self) -> usize {
         self.len()
     }
 
-    fn with_slot<F>(&mut self, mut len: usize, f: F) -> Result<NonZeroUsize, DidntWrite>
+    unsafe fn with_slot<F>(&mut self, len: usize, write: F) -> Result<NonZeroUsize, DidntWrite>
     where
         F: FnOnce(&mut [u8]) -> usize,
     {
         if len > self.len() {
             return Err(DidntWrite);
         }
-        // SAFETY: we early return in case len is greater than slice.len().
-        let s = crate::unsafe_slice_mut!(self, ..len);
-        len = f(s);
-        // SAFETY: we early return in case len is greater than slice.len().
-        let s = crate::unsafe_slice_mut!(self, len..);
-        // SAFETY: this doesn't compile with simple assignment because the compiler
-        // doesn't believe that the subslice has the same lifetime as the original slice,
-        // so we transmute to assure it that it does.
-        *self = unsafe { mem::transmute::<&mut [u8], &mut [u8]>(s) };
-
-        NonZeroUsize::new(len).ok_or(DidntWrite)
+        let written = write(&mut self[..len]);
+        // SAFETY: `written` < `len` is guaranteed by function contract
+        *self = unsafe { mem::take(self).get_unchecked_mut(written..) };
+        NonZeroUsize::new(written).ok_or(DidntWrite)
     }
 }
 
@@ -165,39 +133,30 @@ impl<'a> HasReader for &'a [u8] {
 
 impl Reader for &[u8] {
     fn read(&mut self, into: &mut [u8]) -> Result<NonZeroUsize, DidntRead> {
-        let len = self.len().min(into.len());
-        // SAFETY: len is guaranteed to be the smaller than lhs length.
-        let lhs = crate::unsafe_slice_mut!(into, ..len);
-        let rhs = crate::unsafe_slice!(self, ..len);
-        lhs.copy_from_slice(rhs);
-        // SAFETY: len is guaranteed to be smaller than slice.len().
-        *self = crate::unsafe_slice!(self, len..);
-        NonZeroUsize::new(len).ok_or(DidntRead)
+        let Some(len) = NonZeroUsize::new(self.len().min(into.len())) else {
+            return Err(DidntRead);
+        };
+        let (to_write, remain) = self.split_at(len.get());
+        into[..len.get()].copy_from_slice(to_write);
+        *self = remain;
+        Ok(len)
     }
 
     fn read_exact(&mut self, into: &mut [u8]) -> Result<(), DidntRead> {
         let len = into.len();
         if self.len() < len {
             return Err(DidntRead);
         }
-        // SAFETY: len is guaranteed to be the smaller than lhs length.
-        let lhs = crate::unsafe_slice_mut!(into, ..len);
-        let rhs = crate::unsafe_slice!(self, ..len);
-        lhs.copy_from_slice(rhs);
-        // SAFETY: len is guaranteed to be smaller than slice.len().
-        *self = crate::unsafe_slice!(self, len..);
+        let (to_write, remain) = self.split_at(len);
+        into[..len].copy_from_slice(to_write);
+        *self = remain;
         Ok(())
     }
 
     fn read_u8(&mut self) -> Result<u8, DidntRead> {
-        if !self.can_read() {
-            return Err(DidntRead);
-        }
-        // SAFETY: we early return in case the slice is empty.
-        //         Therefore, there is at least one element in the slice.
-        let ret = *crate::unsafe_slice!(self, 0);
-        *self = crate::unsafe_slice!(self, 1..);
-        Ok(ret)
+        let mut buf = [0; 1];
+        self.read(&mut buf)?;
+        Ok(buf[0])
     }
 
     fn read_zslices<F: FnMut(ZSlice)>(&mut self, len: usize, mut f: F) -> Result<(), DidntRead> {