Migrate SHM (#1798)

* WIP on SHM migration

* posix implementation

* tests for shm api abstraction

* - move advisory locking functionality to shm segment class
- some good refactoring

* fix lints

* fix workspace clippy

* handle flock errors more precisely

* use advisory-lock-based persistency suppression only on linux, mac has (and always had) some issues to fix

* Update unix.rs

* Update unix.rs

* Update unix.rs

* remove unsupported index for shm segment

* Update shm.rs

* rough windows support for shm

* Update unix.rs

* - format fix
- disable some shm tests only for macos

* add missing win_sys

* fix errors on win

* Update unix.rs and windows.rs

* - get rid of some unwrap
- fix clippy
- take new elem_count() behavior into account (elem_count may be bigger then the one requested upon segment creation)

* Update shm.rs

* Update shm.rs

* Update array.rs and shm.rs

* Update array.rs

* Update array.rs

* Get real segment size for created POSIX segment

* fix bug with SHM protocol handshaking

* Update unix.rs and windows.rs

* fix posix segment tests

* Update Cargo.lock, Cargo.toml, and 2 more files...

* Update windows.rs

* fix posix shm provider tests

* Update windows.rs

* Update posix_shm_provider_backend.rs, windows.rs, and posix_shm_provider.rs

* fix allocator test

* WIP on mac persistency problem (flock() doesn't always work on mac tmpfs, so need other bsd-specific mechanism)

* Make BSD and non-BSD implementation for shared memory cleanup

* Add config aliases to properly detect BSD systems

* - fix CI
- add comments

* Update unix.rs

* Apply advisory locking on non-tmpfs for BSD

* fix issues on mac

* Update unix.rs

* Update unix.rs

* Update windows.rs

* Update unix.rs and windows.rs

* - review fixes
- fix segment collision from different runs

* Update shm.rs

* Update shm.rs

* Review fixes

Cargo.toml

@@ -90,7 +90,6 @@ bytes = "1.7.1"
 clap = { version = "4.5.17", features = ["derive"] }
 console-subscriber = "0.4.0"
 const_format = "0.2.33"
-crc = "3.2.1"
 criterion = "0.5"
 crossbeam-utils = "0.8.20"
 crossbeam-queue = "0.3.12"
@@ -163,7 +162,6 @@ serde_yaml = "0.9.34"
 static_init = "1.0.3"
 stabby = "36.1.1"
 sha3 = "0.10.8"
-shared_memory = "0.12.4"
 shellexpand = "3.1.0"
 socket2 = { version = "0.5.7", features = ["all"] }
 stop-token = "0.7.0"
@@ -188,6 +186,7 @@ uuid = { version = "1.10.0", default-features = false, features = [
 validated_struct = "2.1.0"
 vec_map = "0.8.2"
 webpki-roots = "0.26.5"
+win-sys = "0.3"
 winapi = { version = "0.3.9", features = ["iphlpapi", "winerror"] }
 x509-parser = "0.16.0"
 z-serial = "0.3.1"

commons/zenoh-shm/Cargo.toml

@@ -33,15 +33,13 @@ test = ["num_cpus"]
 
 [dependencies]
 async-trait = { workspace = true }
-crc = { workspace = true }
 tracing = { workspace = true }
-shared_memory = { workspace = true }
 tokio = { workspace = true }
 zenoh-result = { workspace = true }
 zenoh-core = { workspace = true }
 zenoh-macros = { workspace = true }
 zenoh-buffers = { workspace = true }
-rand = { workspace = true }
+rand = { workspace = true, features = ["std", "std_rng"] }
 static_init = { workspace = true }
 num-traits = { workspace = true }
 num_cpus = { workspace = true, optional = true }
@@ -51,6 +49,14 @@ crossbeam-queue = { workspace = true }
 
 [target.'cfg(unix)'.dependencies]
 advisory-lock = { workspace = true }
+nix = { workspace = true, features = ["fs", "mman"] }
+
+[target.'cfg(windows)'.dependencies]
+win-sys = { workspace = true }
+winapi = { workspace = true }
 
 [dev-dependencies]
 libc = { workspace = true }
+
+[build-dependencies]
+cfg_aliases = "0.2.1"

commons/zenoh-shm/build.rs

@@ -0,0 +1,49 @@
+//
+// Copyright (c) 2025 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+
+use cfg_aliases::cfg_aliases;
+
+fn main() {
+    // these aliases should at least be included in the same aliases of Nix crate:
+    // ___________________
+    // |                 |
+    // |  Nix aliases    |
+    // |  ___________    |
+    // |  |   Our   |    |
+    // |  | aliases |    |
+    // |  |_________|    |
+    // |_________________|
+    cfg_aliases! {
+        dragonfly: { target_os = "dragonfly" },
+        ios: { target_os = "ios" },
+        freebsd: { target_os = "freebsd" },
+        macos: { target_os = "macos" },
+        netbsd: { target_os = "netbsd" },
+        openbsd: { target_os = "openbsd" },
+        watchos: { target_os = "watchos" },
+        tvos: { target_os = "tvos" },
+        visionos: { target_os = "visionos" },
+
+        apple_targets: { any(ios, macos, watchos, tvos, visionos) },
+        bsd: { any(freebsd, dragonfly, netbsd, openbsd, apple_targets) },
+
+        // we use this alias to detect platforms that
+        // don't support advisory file locking on tmpfs
+        shm_external_lockfile: { any(bsd, target_os = "redox") },
+    }
+
+    println!("cargo:rustc-check-cfg=cfg(apple_targets)");
+    println!("cargo:rustc-check-cfg=cfg(bsd)");
+    println!("cargo:rustc-check-cfg=cfg(shm_external_lockfile)");
+}

commons/zenoh-shm/src/api/protocol_implementations/posix/posix_shm_provider_backend.rs

@@ -133,21 +133,28 @@ impl PosixShmProviderBackend {
     fn new(layout: &MemoryLayout) -> ZResult<Self> {
         let segment = PosixShmSegment::create(layout.size())?;
 
+        // because of platform specific, our shm segment is >= requested size, so in order to utilize
+        // additional memory we re-layout the size
+        let real_size = segment.segment.elem_count().get();
+        let aligned_size = (real_size
+            - (real_size % layout.alignment().get_alignment_value().get()))
+        .try_into()?;
+
         let mut free_list = BinaryHeap::new();
         let root_chunk = Chunk {
             offset: 0,
-            size: layout.size(),
+            size: aligned_size,
         };
         free_list.push(root_chunk);
 
         tracing::trace!(
-            "Created PosixShmProviderBackend id {}, layout {:?}",
+            "Created PosixShmProviderBackend id {}, layout {:?}, aligned size {aligned_size}",
             segment.segment.id(),
             layout
         );
 
         Ok(Self {
-            available: AtomicUsize::new(layout.size().get()),
+            available: AtomicUsize::new(aligned_size.get()),
             segment,
             free_list: Mutex::new(free_list),
             alignment: layout.alignment(),

commons/zenoh-shm/src/api/protocol_implementations/posix/posix_shm_segment.rs

@@ -24,21 +24,19 @@ use crate::{
     posix_shm::array::ArrayInSHM,
 };
 
-const POSIX_SHM_SEGMENT_PREFIX: &str = "posix_shm_provider_segment";
-
 #[derive(Debug)]
 pub(crate) struct PosixShmSegment {
     pub(crate) segment: ArrayInSHM<SegmentID, u8, ChunkID>,
 }
 
 impl PosixShmSegment {
     pub(crate) fn create(alloc_size: NonZeroUsize) -> ZResult<Self> {
-        let segment = ArrayInSHM::create(alloc_size.get(), POSIX_SHM_SEGMENT_PREFIX)?;
+        let segment = ArrayInSHM::create(alloc_size)?;
         Ok(Self { segment })
     }
 
     pub(crate) fn open(id: SegmentID) -> ZResult<Self> {
-        let segment = ArrayInSHM::open(id, POSIX_SHM_SEGMENT_PREFIX)?;
+        let segment = ArrayInSHM::open(id)?;
         Ok(Self { segment })
     }
 }

commons/zenoh-shm/src/lib.rs

@@ -60,6 +60,7 @@ pub mod posix_shm;
 pub mod reader;
 pub mod version;
 pub mod watchdog;
+tested_crate_module!(shm);
 
 /// Information about a [`ShmBufInner`].
 ///

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

@@ -19,12 +19,10 @@ use zenoh_result::ZResult;
 use super::descriptor::{MetadataIndex, MetadataSegmentID};
 use crate::{header::chunk_header::ChunkHeaderType, posix_shm::struct_in_shm::StructInSHM};
 
-const METADATA_SEGMENT_PREFIX: &str = "metadata";
-
 #[stabby::stabby]
 pub struct Metadata<const S: usize> {
     headers: [ChunkHeaderType; S],
-    watchdogs: [AtomicU64; S], // todo: replace with (S + 63) / 64 when Rust supports it
+    watchdogs: [AtomicU64; S], // TODO: replace with (S + 63) / 64 when Rust supports it
 }
 
 impl<const S: usize> Metadata<S> {
@@ -59,12 +57,12 @@ pub struct MetadataSegment<const S: usize = 32768> {
 
 impl<const S: usize> MetadataSegment<S> {
     pub fn create() -> ZResult<Self> {
-        let data = StructInSHM::create(METADATA_SEGMENT_PREFIX)?;
+        let data = StructInSHM::create()?;
         Ok(Self { data })
     }
 
     pub fn open(id: MetadataSegmentID) -> ZResult<Self> {
-        let data = StructInSHM::open(id, METADATA_SEGMENT_PREFIX)?;
+        let data = StructInSHM::open(id)?;
         Ok(Self { data })
     }
 }