Renaming & Dev

.github/workflows/ci.yml

@@ -67,7 +67,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: clippy
-          args: --all-targets --features shared-memory --features transport_shm -- -D warnings
+          args: --all-targets --features shared-memory --features transport_unixpipe -- -D warnings
 
   test:
     name: Run tests on ${{ matrix.os }}
@@ -107,7 +107,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: nextest
-          args: run -F shared-memory -F transport_shm -p zenoh-transport
+          args: run -F shared-memory -F transport_unixpipe -p zenoh-transport
         env:
           CARGO_REGISTRIES_CRATES_IO_PROTOCOL: sparse
           ASYNC_STD_THREAD_COUNT: 4

Cargo.toml

@@ -37,7 +37,7 @@ members = [
   "io/zenoh-links/zenoh-link-udp/",
   "io/zenoh-links/zenoh-link-unixsock_stream/",
   "io/zenoh-links/zenoh-link-ws/",
-  "io/zenoh-links/zenoh-link-shm/",
+  "io/zenoh-links/zenoh-link-unixpipe/",
   "io/zenoh-transport",
   "plugins/example-plugin",
   "plugins/zenoh-backend-traits",
@@ -180,7 +180,7 @@ zenoh-link-unixsock_stream = { version = "0.10.0-dev", path = "io/zenoh-links/ze
 zenoh-link-quic = { version = "0.10.0-dev", path = "io/zenoh-links/zenoh-link-quic" }
 zenoh-link-udp = { version = "0.10.0-dev", path = "io/zenoh-links/zenoh-link-udp" }
 zenoh-link-ws = { version = "0.10.0-dev", path = "io/zenoh-links/zenoh-link-ws" }
-zenoh-link-shm = { version = "0.10.0-dev", path = "io/zenoh-links/zenoh-link-shm" }
+zenoh-link-unixpipe = { version = "0.10.0-dev", path = "io/zenoh-links/zenoh-link-unixpipe" }
 zenoh-link-serial = { version = "0.10.0-dev", path = "io/zenoh-links/zenoh-link-serial" }
 zenoh-link = { version = "0.10.0-dev", path = "io/zenoh-link" }
 zenoh-link-commons = { version = "0.10.0-dev", path = "io/zenoh-link-commons" }

DEFAULT_CONFIG.json5

@@ -125,6 +125,14 @@
       max_sessions: 1000,
       /// Maximum number of incoming links that are admitted per session
       max_links: 1,
+      /// Enables the LowLatency transport
+      /// This option does not make LowLatency transport mandatory, the actual implementation of transport
+      /// used will depend on Establish procedure and other party's settings
+      ///
+      /// NOTE: Currently, the LowLatency transport doesn't preserve QoS prioritization. 
+      /// NOTE: Due to the note above, 'lowlatency' is incompatible with 'qos' option, so in order to
+      ///       enable 'lowlatency' you need to explicitly disable 'qos'.
+      lowlatency: false,
     },
     qos: {
       enabled: true,

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

@@ -45,7 +45,8 @@ where
         let mut n_exts = (x.ext_qos.is_some() as u8)
             + (x.ext_shm.is_some() as u8)
             + (x.ext_auth.is_some() as u8)
-            + (x.ext_mlink.is_some() as u8);
+            + (x.ext_mlink.is_some() as u8)
+            + (x.ext_lowlatency.is_some() as u8);
         if n_exts != 0 {
             header |= flag::Z;
         }
@@ -87,6 +88,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (mlink, n_exts != 0))?;
         }
+        if let Some(lowlatency) = x.ext_lowlatency.as_ref() {
+            n_exts -= 1;
+            self.write(&mut *writer, (lowlatency, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -144,6 +149,7 @@ where
         let mut ext_shm = None;
         let mut ext_auth = None;
         let mut ext_mlink = None;
+        let mut ext_lowlatency = None;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -170,6 +176,11 @@ where
                     ext_mlink = Some(a);
                     has_ext = ext;
                 }
+                ext::LowLatency::ID => {
+                    let (q, ext): (ext::LowLatency, bool) = eodec.read(&mut *reader)?;
+                    ext_lowlatency = Some(q);
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "InitSyn", ext)?;
                 }
@@ -186,6 +197,7 @@ where
             ext_shm,
             ext_auth,
             ext_mlink,
+            ext_lowlatency,
         })
     }
 }
@@ -206,7 +218,8 @@ where
         let mut n_exts = (x.ext_qos.is_some() as u8)
             + (x.ext_shm.is_some() as u8)
             + (x.ext_auth.is_some() as u8)
-            + (x.ext_mlink.is_some() as u8);
+            + (x.ext_mlink.is_some() as u8)
+            + (x.ext_lowlatency.is_some() as u8);
         if n_exts != 0 {
             header |= flag::Z;
         }
@@ -251,6 +264,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (mlink, n_exts != 0))?;
         }
+        if let Some(lowlatency) = x.ext_lowlatency.as_ref() {
+            n_exts -= 1;
+            self.write(&mut *writer, (lowlatency, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -311,6 +328,7 @@ where
         let mut ext_shm = None;
         let mut ext_auth = None;
         let mut ext_mlink = None;
+        let mut ext_lowlatency = None;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -337,6 +355,11 @@ where
                     ext_mlink = Some(a);
                     has_ext = ext;
                 }
+                ext::LowLatency::ID => {
+                    let (q, ext): (ext::LowLatency, bool) = eodec.read(&mut *reader)?;
+                    ext_lowlatency = Some(q);
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "InitAck", ext)?;
                 }
@@ -354,6 +377,7 @@ where
             ext_shm,
             ext_auth,
             ext_mlink,
+            ext_lowlatency,
         })
     }
 }

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

@@ -25,50 +25,48 @@ use zenoh_buffers::{
     reader::{BacktrackableReader, DidntRead, Reader},
     writer::{DidntWrite, Writer},
 };
-#[cfg(feature = "shared-memory")]
-use zenoh_protocol::network::NetworkMessage;
 use zenoh_protocol::{
     common::{imsg, ZExtZ64},
+    network::NetworkMessage,
     transport::*,
 };
 
-// TransportMessageShm
-#[cfg(feature = "shared-memory")]
-impl<W> WCodec<&TransportMessageShm, &mut W> for Zenoh080
+// TransportMessageLowLatency
+impl<W> WCodec<&TransportMessageLowLatency, &mut W> for Zenoh080
 where
     W: Writer,
 {
     type Output = Result<(), DidntWrite>;
 
-    fn write(self, writer: &mut W, x: &TransportMessageShm) -> Self::Output {
+    fn write(self, writer: &mut W, x: &TransportMessageLowLatency) -> Self::Output {
         match &x.body {
-            TransportBodyShm::Network(b) => self.write(&mut *writer, b.as_ref()),
-            TransportBodyShm::KeepAlive(b) => self.write(&mut *writer, b),
-            TransportBodyShm::Close(b) => self.write(&mut *writer, b),
+            TransportBodyLowLatency::Network(b) => self.write(&mut *writer, b),
+            TransportBodyLowLatency::KeepAlive(b) => self.write(&mut *writer, b),
+            TransportBodyLowLatency::Close(b) => self.write(&mut *writer, b),
         }
     }
 }
-#[cfg(feature = "shared-memory")]
-impl<R> RCodec<TransportMessageShm, &mut R> for Zenoh080
+
+impl<R> RCodec<TransportMessageLowLatency, &mut R> for Zenoh080
 where
     R: Reader + BacktrackableReader,
 {
     type Error = DidntRead;
 
-    fn read(self, reader: &mut R) -> Result<TransportMessageShm, Self::Error> {
+    fn read(self, reader: &mut R) -> Result<TransportMessageLowLatency, Self::Error> {
         let header: u8 = self.read(&mut *reader)?;
 
         let codec = Zenoh080Header::new(header);
         let body = match imsg::mid(codec.header) {
-            id::KEEP_ALIVE => TransportBodyShm::KeepAlive(codec.read(&mut *reader)?),
-            id::CLOSE => TransportBodyShm::Close(codec.read(&mut *reader)?),
+            id::KEEP_ALIVE => TransportBodyLowLatency::KeepAlive(codec.read(&mut *reader)?),
+            id::CLOSE => TransportBodyLowLatency::Close(codec.read(&mut *reader)?),
             _ => {
                 let nw: NetworkMessage = codec.read(&mut *reader)?;
-                TransportBodyShm::Network(Box::new(nw))
+                TransportBodyLowLatency::Network(nw)
             }
         };
 
-        Ok(TransportMessageShm { body })
+        Ok(TransportMessageLowLatency { body })
     }
 }
 

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

@@ -43,7 +43,8 @@ where
         let mut n_exts = (x.ext_qos.is_some() as u8)
             + (x.ext_shm.is_some() as u8)
             + (x.ext_auth.is_some() as u8)
-            + (x.ext_mlink.is_some() as u8);
+            + (x.ext_mlink.is_some() as u8)
+            + (x.ext_lowlatency.is_some() as u8);
         if n_exts != 0 {
             header |= flag::Z;
         }
@@ -75,6 +76,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (mlink, n_exts != 0))?;
         }
+        if let Some(lowlatency) = x.ext_lowlatency.as_ref() {
+            n_exts -= 1;
+            self.write(&mut *writer, (lowlatency, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -119,6 +124,7 @@ where
         let mut ext_shm = None;
         let mut ext_auth = None;
         let mut ext_mlink = None;
+        let mut ext_lowlatency = None;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -145,6 +151,11 @@ where
                     ext_mlink = Some(a);
                     has_ext = ext;
                 }
+                ext::LowLatency::ID => {
+                    let (q, ext): (ext::LowLatency, bool) = eodec.read(&mut *reader)?;
+                    ext_lowlatency = Some(q);
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "OpenSyn", ext)?;
                 }
@@ -159,6 +170,7 @@ where
             ext_shm,
             ext_auth,
             ext_mlink,
+            ext_lowlatency,
         })
     }
 }
@@ -181,7 +193,8 @@ where
         let mut n_exts = (x.ext_qos.is_some() as u8)
             + (x.ext_shm.is_some() as u8)
             + (x.ext_auth.is_some() as u8)
-            + (x.ext_mlink.is_some() as u8);
+            + (x.ext_mlink.is_some() as u8)
+            + (x.ext_lowlatency.is_some() as u8);
         if n_exts != 0 {
             header |= flag::Z;
         }
@@ -212,6 +225,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (mlink, n_exts != 0))?;
         }
+        if let Some(lowlatency) = x.ext_lowlatency.as_ref() {
+            n_exts -= 1;
+            self.write(&mut *writer, (lowlatency, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -255,6 +272,7 @@ where
         let mut ext_shm = None;
         let mut ext_auth = None;
         let mut ext_mlink = None;
+        let mut ext_lowlatency = None;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -281,6 +299,11 @@ where
                     ext_mlink = Some(a);
                     has_ext = ext;
                 }
+                ext::LowLatency::ID => {
+                    let (q, ext): (ext::LowLatency, bool) = eodec.read(&mut *reader)?;
+                    ext_lowlatency = Some(q);
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "OpenAck", ext)?;
                 }
@@ -294,6 +317,7 @@ where
             ext_shm,
             ext_auth,
             ext_mlink,
+            ext_lowlatency,
         })
     }
 }

commons/zenoh-config/src/defaults.rs

@@ -106,6 +106,7 @@ impl Default for TransportUnicastConf {
             accept_pending: 100,
             max_sessions: 1_000,
             max_links: 1,
+            lowlatency: false,
         }
     }
 }