SHM handoff

Cargo.toml

@@ -132,6 +132,7 @@ lazy_static = "1.5.0"
 leb128 = "0.2"
 libc = "0.2.175"
 libloading = "0.8.9"
+lockfree = "0.5"
 lru = "0.16.2"
 lz4_flex = "0.10.0"
 mime = "0.3.17"

commons/zenoh-codec/Cargo.toml

@@ -39,6 +39,7 @@ shared-memory = [
 std = ["tracing", "uhlc/std", "zenoh-buffers/std", "zenoh-protocol/std"]
 
 [dependencies]
+rand = { workspace = true }
 tracing = { workspace = true, optional = true }
 uhlc = { workspace = true }
 zenoh-buffers = { workspace = true, default-features = false }
@@ -48,7 +49,6 @@ zenoh-shm = { workspace = true, optional = true }
 # INFO: May cause problems when testing no_std stuff. Check this tool: https://docs.rs/crate/cargo-no-dev-deps/0.1.0
 [dev-dependencies]
 criterion = { workspace = true }
-
 rand = { workspace = true, features = ["default"] }
 zenoh-protocol = { workspace = true, features = ["test"] }
 zenoh-util = { workspace = true }

commons/zenoh-codec/src/core/shm.rs

@@ -18,7 +18,11 @@ use zenoh_buffers::{
     writer::{DidntWrite, Writer},
 };
 use zenoh_shm::{
-    api::provider::chunk::ChunkDescriptor, metadata::descriptor::MetadataDescriptor, ShmBufInfo,
+    api::provider::chunk::ChunkDescriptor,
+    metadata::descriptor::MetadataDescriptor,
+    posix_shm::{segment::Segment, struct_in_shm::StructInSHM},
+    shm::SegmentID,
+    ShmBufInfo,
 };
 
 use crate::{RCodec, WCodec, Zenoh080};
@@ -143,3 +147,61 @@ where
         Ok(shm_info)
     }
 }
+
+impl<'a, W, ID> WCodec<&'a Segment<ID>, &'a mut W> for Zenoh080
+where
+    W: Writer,
+    ID: SegmentID,
+    rand::distributions::Standard: rand::distributions::Distribution<ID>,
+    Zenoh080: WCodec<ID, &'a mut W>,
+{
+    type Output = <Zenoh080 as WCodec<ID, &'a mut W>>::Output;
+
+    fn write(self, writer: &'a mut W, x: &Segment<ID>) -> Self::Output {
+        self.write(&mut *writer, x.id())
+    }
+}
+
+impl<'a, R, ID> RCodec<Segment<ID>, &'a mut R> for Zenoh080
+where
+    R: Reader,
+    ID: SegmentID,
+    rand::distributions::Standard: rand::distributions::Distribution<ID>,
+    Zenoh080: RCodec<ID, &'a mut R>,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &'a mut R) -> Result<Segment<ID>, Self::Error> {
+        let id = self.read(&mut *reader).map_err(|_| DidntRead)?;
+        Segment::open(id).map_err(|_| DidntRead)
+    }
+}
+
+impl<'a, W, ID, Elem> WCodec<&'a StructInSHM<ID, Elem>, &'a mut W> for Zenoh080
+where
+    W: Writer,
+    ID: SegmentID,
+    rand::distributions::Standard: rand::distributions::Distribution<ID>,
+    Zenoh080: WCodec<ID, &'a mut W>,
+{
+    type Output = <Zenoh080 as WCodec<ID, &'a mut W>>::Output;
+
+    fn write(self, writer: &'a mut W, x: &StructInSHM<ID, Elem>) -> Self::Output {
+        self.write(&mut *writer, x.id())
+    }
+}
+
+impl<'a, R, ID, Elem> RCodec<StructInSHM<ID, Elem>, &'a mut R> for Zenoh080
+where
+    R: Reader,
+    ID: SegmentID,
+    rand::distributions::Standard: rand::distributions::Distribution<ID>,
+    Zenoh080: RCodec<ID, &'a mut R>,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &'a mut R) -> Result<StructInSHM<ID, Elem>, Self::Error> {
+        let id = self.read(&mut *reader).map_err(|_| DidntRead)?;
+        StructInSHM::open(id).map_err(|_| DidntRead)
+    }
+}

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

@@ -108,7 +108,7 @@ pub mod ext {
     /// # Shm extension
     /// Used as challenge for probing shared memory capabilities
     #[cfg(feature = "shared-memory")]
-    pub type Shm = zextz64!(0x2, false);
+    pub type Shm = zextzbuf!(0x2, false);
 
     /// # Auth extension
     /// Used as challenge for probing authentication rights
@@ -155,7 +155,7 @@ impl OpenSyn {
         let cookie = ZSlice::rand(rng.gen_range(MIN..=MAX));
         let ext_qos = rng.gen_bool(0.5).then_some(ZExtUnit::rand());
         #[cfg(feature = "shared-memory")]
-        let ext_shm = rng.gen_bool(0.5).then_some(ZExtZ64::rand());
+        let ext_shm = rng.gen_bool(0.5).then_some(ZExtZBuf::rand());
         let ext_auth = rng.gen_bool(0.5).then_some(ZExtZBuf::rand());
         let ext_mlink = rng.gen_bool(0.5).then_some(ZExtZBuf::rand());
         let ext_lowlatency = rng.gen_bool(0.5).then_some(ZExtUnit::rand());
@@ -219,7 +219,7 @@ impl OpenAck {
         let initial_sn: TransportSn = rng.gen();
         let ext_qos = rng.gen_bool(0.5).then_some(ZExtUnit::rand());
         #[cfg(feature = "shared-memory")]
-        let ext_shm = rng.gen_bool(0.5).then_some(ZExtZ64::rand());
+        let ext_shm = rng.gen_bool(0.5).then_some(ZExtZBuf::rand());
         let ext_auth = rng.gen_bool(0.5).then_some(ZExtZBuf::rand());
         let ext_mlink = rng.gen_bool(0.5).then_some(ZExtUnit::rand());
         let ext_lowlatency = rng.gen_bool(0.5).then_some(ZExtUnit::rand());

commons/zenoh-shm/src/lib.rs

@@ -61,9 +61,9 @@ pub mod init;
 pub mod metadata;
 pub mod posix_shm;
 pub mod reader;
+pub mod shm;
 pub mod version;
 pub mod watchdog;
-tested_crate_module!(shm);
 
 /// Information about a [`ShmBufInner`].
 ///
@@ -95,6 +95,37 @@ impl ShmBufInfo {
     }
 }
 
+/// A hard shared reference to SHM buffer
+#[derive(Debug, Clone)]
+#[allow(unused)]
+pub struct ShmBufHardRef(ConfirmedDescriptor);
+
+impl From<&ShmBufInner> for ShmBufHardRef {
+    fn from(value: &ShmBufInner) -> Self {
+        Self(value.metadata.clone())
+    }
+}
+impl From<&ZShm> for ShmBufHardRef {
+    fn from(value: &ZShm) -> Self {
+        (&value.inner).into()
+    }
+}
+impl From<&zshm> for ShmBufHardRef {
+    fn from(value: &zshm) -> Self {
+        (&value.inner).into()
+    }
+}
+impl From<&ZShmMut> for ShmBufHardRef {
+    fn from(value: &ZShmMut) -> Self {
+        (&value.inner).into()
+    }
+}
+impl From<&zshmmut> for ShmBufHardRef {
+    fn from(value: &zshmmut) -> Self {
+        (&value.inner).into()
+    }
+}
+
 /// A zenoh buffer in shared memory.
 pub struct ShmBufInner {
     pub(crate) metadata: ConfirmedDescriptor,

commons/zenoh-shm/src/metadata/descriptor.rs

@@ -113,7 +113,7 @@ impl OwnedWatchdog {
 pub struct OwnedMetadataDescriptor {
     pub(crate) segment: Arc<MetadataSegment>,
     header: &'static ChunkHeaderType,
-    watchdog: OwnedWatchdog,
+    pub(crate) watchdog: OwnedWatchdog,
 }
 
 impl Hash for OwnedMetadataDescriptor {

commons/zenoh-shm/src/posix_shm/mod.rs

@@ -13,6 +13,6 @@
 //
 
 pub mod array;
-pub mod struct_in_shm;
-tested_crate_module!(segment);
 pub(crate) mod cleanup;
+pub mod segment;
+pub mod struct_in_shm;

commons/zenoh-shm/src/posix_shm/segment.rs

@@ -22,6 +22,7 @@ use crate::{cleanup::CLEANUP, shm};
 const SEGMENT_DEDICATE_TRIES: usize = 100;
 
 /// Segment of shared memory identified by an ID
+#[derive(PartialEq, Eq)]
 pub struct Segment<ID>
 where
     rand::distributions::Standard: rand::distributions::Distribution<ID>,

commons/zenoh-shm/src/posix_shm/struct_in_shm.rs

@@ -36,6 +36,23 @@ where
     _phantom: PhantomData<Elem>,
 }
 
+impl<ID, Elem: Sync> PartialEq for StructInSHM<ID, Elem>
+where
+    rand::distributions::Standard: rand::distributions::Distribution<ID>,
+    ID: shm::SegmentID,
+{
+    fn eq(&self, other: &Self) -> bool {
+        self.inner == other.inner
+    }
+}
+
+impl<ID, Elem: Sync> Eq for StructInSHM<ID, Elem>
+where
+    rand::distributions::Standard: rand::distributions::Distribution<ID>,
+    ID: shm::SegmentID,
+{
+}
+
 unsafe impl<ID, Elem: Sync> Sync for StructInSHM<ID, Elem>
 where
     rand::distributions::Standard: rand::distributions::Distribution<ID>,

commons/zenoh-shm/src/shm/mod.rs

@@ -52,6 +52,14 @@ pub struct Segment<ID: SegmentID> {
     inner: platform::SegmentImpl<ID>,
 }
 
+impl<ID: SegmentID> Eq for Segment<ID> {}
+
+impl<ID: SegmentID> PartialEq for Segment<ID> {
+    fn eq(&self, other: &Self) -> bool {
+        self.inner.id() == other.inner.id()
+    }
+}
+
 impl<ID: SegmentID> std::fmt::Debug for Segment<ID> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_tuple("Segment").field(&self.inner).finish()

commons/zenoh-shm/src/version.rs

@@ -12,4 +12,4 @@
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
 
-pub const SHM_VERSION: u64 = 1;
+pub const SHM_VERSION: u64 = 2;

commons/zenoh-shm/src/watchdog/confirmator.rs

@@ -41,19 +41,20 @@ pub struct ConfirmedDescriptor {
 
 impl Clone for ConfirmedDescriptor {
     fn clone(&self) -> Self {
+        // NOTE: need `ConfirmedDescriptor::new` to add the descriptor to the confirmed segment
         ConfirmedDescriptor::new(self.owned.clone(), self.confirmed.clone())
     }
 }
 
 impl Drop for ConfirmedDescriptor {
     fn drop(&mut self) {
-        self.confirmed.remove(self.owned.clone());
+        self.confirmed.remove(self.owned.deref().clone());
     }
 }
 
 impl ConfirmedDescriptor {
     fn new(owned: OwnedMetadataDescriptor, confirmed: Arc<ConfirmedSegment>) -> Self {
-        confirmed.add(owned.clone());
+        confirmed.add(owned.deref().clone());
         Self { owned, confirmed }
     }
 }
@@ -99,12 +100,12 @@ impl ConfirmedSegment {
         }
     }
 
-    fn add(&self, descriptor: OwnedMetadataDescriptor) {
-        self.make_transaction(Transaction::Add(descriptor.deref().clone()));
+    fn add(&self, descriptor: OwnedWatchdog) {
+        self.make_transaction(Transaction::Add(descriptor));
     }
 
-    fn remove(&self, descriptor: OwnedMetadataDescriptor) {
-        self.make_transaction(Transaction::Remove(descriptor.deref().clone()));
+    fn remove(&self, descriptor: OwnedWatchdog) {
+        self.make_transaction(Transaction::Remove(descriptor));
     }
 
     fn make_transaction(&self, transaction: Transaction) {

io/zenoh-transport/Cargo.toml

@@ -29,6 +29,8 @@ auth_pubkey = ["rsa", "transport_auth"]
 auth_usrpwd = ["transport_auth"]
 default = ["test", "transport_multilink"]
 shared-memory = [
+  "lockfree",
+  "static_init",
   "zenoh-buffers/shared-memory",
   "zenoh-codec/shared-memory",
   "zenoh-protocol/shared-memory",
@@ -59,12 +61,14 @@ crossbeam-utils = { workspace = true }
 flume = { workspace = true }
 futures = { workspace = true }
 lazy_static = { workspace = true }
+lockfree = { workspace = true, optional = true }
 lz4_flex = { workspace = true }
 rand = { workspace = true, features = ["default"] }
 ringbuffer-spsc = { workspace = true }
 rsa = { workspace = true, optional = true }
 serde = { workspace = true, features = ["default"] }
 sha3 = { workspace = true }
+static_init = { workspace = true, optional = true }
 tokio = { workspace = true, features = [
   "fs",
   "io-util",

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

@@ -16,3 +16,5 @@ pub(crate) mod defragmentation;
 pub(crate) mod pipeline;
 pub(crate) mod priority;
 pub(crate) mod seq_num;
+#[cfg(feature = "shared-memory")]
+pub mod shm;