POC: SHM watchdog

.github/workflows/sync-lockfiles.yml

@@ -0,0 +1,103 @@
+name: Sync Cargo lockfile with Zenoh's dependants
+
+on:
+  schedule:
+    - cron: "0 0 * * *" # At the end of every day
+  workflow_dispatch:
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  fetch:
+    name: Fetch Zenoh's lockfile
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Zenoh
+        uses: actions/checkout@v4
+        with:
+          repository: eclipse-zenoh/zenoh
+
+      - name: Upload lockfile
+        uses: actions/upload-artifact@v3
+        with:
+          name: Cargo.lock
+          path: Cargo.lock
+
+  sync:
+    name: Sync Cargo lockfile with ${{ matrix.dependant }}
+    needs: fetch
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        dependant:
+          - zenoh-c
+          - zenoh-python
+          - zenoh-java
+          - zenoh-kotlin
+          - zenoh-plugin-dds
+          - zenoh-plugin-mqtt
+          - zenoh-plugin-ros1
+          - zenoh-plugin-ros2dds
+          - zenoh-plugin-webserver
+          - zenoh-backend-filesystem
+          - zenoh-backend-influxdb
+          - zenoh-backend-rocksdb
+          - zenoh-backend-s3
+    steps:
+      - name: Checkout ${{ matrix.dependant }}
+        uses: actions/checkout@v4
+        with:
+          repository: eclipse-zenoh/${{ matrix.dependant }}
+          submodules: true
+          token: ${{ secrets.BOT_TOKEN_WORKFLOW }}
+
+      - name: Install Rust toolchain
+        # NOTE: Showing the active Rust toolchain (defined by the rust-toolchain.toml file)
+        # will have the side effect of installing it; if it's not installed already.
+        run: rustup show
+
+      # NOTE: Not all Zenoh dependants have their Cargo manifest and lockfile
+      # at the repository's toplevel. The only exception being zenoh-kotlin and
+      # zenoh-java. Thus the need for this ugly workaround.
+      - name: Compute crate path of ${{ matrix.dependant }}
+        id: crate-path
+        run: |
+          if [[ "${{ matrix.dependant }}" =~ zenoh-(java|kotlin) ]]; then
+            echo "value=zenoh-jni" >> $GITHUB_OUTPUT
+          else
+            echo "value=." >> $GITHUB_OUTPUT
+          fi
+
+      - name: Override ${{ matrix.dependant }} lockfile with Zenoh's
+        uses: actions/download-artifact@v3
+        with:
+          name: Cargo.lock
+          path: ${{ steps.crate-path.outputs.value }}
+
+      - name: Rectify lockfile
+        # NOTE: Checking the package for errors will rectify the Cargo.lock while preserving
+        # the dependency versions fetched from source.
+        run: cargo check --manifest-path ${{ steps.crate-path.outputs.value }}/Cargo.toml
+
+      - name: Create/Update a pull request if the lockfile changed
+        id: cpr
+        # NOTE: If there is a pending PR, this action will simply update it with a forced push.
+        uses: peter-evans/create-pull-request@v5
+        with:
+          title: Sync lockfile with Zenoh's
+          body: Automated synchronization of the Cargo lockfile with Zenoh. This is done to ensure plugin ABI compatibility.
+          commit-message: "chore: Sync Cargo lockfile with Zenoh's"
+          committer: eclipse-zenoh-bot <[email protected]>
+          author: eclipse-zenoh-bot <[email protected]>
+          branch: eclipse-zenoh-bot/sync-lockfile
+          delete-branch: true
+          labels: dependencies
+          token: ${{ secrets.BOT_TOKEN_WORKFLOW }}
+
+      - name: Enable auto merge for the pull request
+        if: steps.cpr.outputs.pull-request-operation == 'created'
+        run: gh pr merge -R "eclipse-zenoh/${{ matrix.dependant }}" --merge --auto "${{ steps.cpr.outputs.pull-request-number }}"
+        env:
+          GH_TOKEN: ${{ secrets.BOT_TOKEN_WORKFLOW }}

Cargo.toml

@@ -109,6 +109,7 @@ log = "0.4.17"
 lz4_flex = "0.11"
 nix = { version = "0.27", features = ["fs"] }
 num_cpus = "1.15.0"
+num-traits = { version = "0.2.17", default-features = false }
 ordered-float = "4.1.1"
 panic-message = "0.3.0"
 paste = "1.0.12"

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

@@ -16,7 +16,50 @@ use zenoh_buffers::{
     reader::{DidntRead, Reader},
     writer::{DidntWrite, Writer},
 };
-use zenoh_shm::SharedMemoryBufInfo;
+use zenoh_shm::{
+    api::provider::chunk::ChunkDescriptor, header::descriptor::HeaderDescriptor,
+    watchdog::descriptor::Descriptor, SharedMemoryBufInfo,
+};
+
+impl<W> WCodec<&Descriptor, &mut W> for Zenoh080
+where
+    W: Writer,
+{
+    type Output = Result<(), DidntWrite>;
+
+    fn write(self, writer: &mut W, x: &Descriptor) -> Self::Output {
+        self.write(&mut *writer, x.id)?;
+        self.write(&mut *writer, x.index_and_bitpos)?;
+        Ok(())
+    }
+}
+
+impl<W> WCodec<&HeaderDescriptor, &mut W> for Zenoh080
+where
+    W: Writer,
+{
+    type Output = Result<(), DidntWrite>;
+
+    fn write(self, writer: &mut W, x: &HeaderDescriptor) -> Self::Output {
+        self.write(&mut *writer, x.id)?;
+        self.write(&mut *writer, x.index)?;
+        Ok(())
+    }
+}
+
+impl<W> WCodec<&ChunkDescriptor, &mut W> for Zenoh080
+where
+    W: Writer,
+{
+    type Output = Result<(), DidntWrite>;
+
+    fn write(self, writer: &mut W, x: &ChunkDescriptor) -> Self::Output {
+        self.write(&mut *writer, x.segment)?;
+        self.write(&mut *writer, x.chunk)?;
+        self.write(&mut *writer, x.len)?;
+        Ok(())
+    }
+}
 
 impl<W> WCodec<&SharedMemoryBufInfo, &mut W> for Zenoh080
 where
@@ -26,33 +69,96 @@ where
 
     fn write(self, writer: &mut W, x: &SharedMemoryBufInfo) -> Self::Output {
         let SharedMemoryBufInfo {
-            offset,
-            length,
-            shm_manager,
-            kind,
+            data_descriptor,
+            shm_protocol,
+            data_len,
+            watchdog_descriptor,
+            header_descriptor,
+            generation,
         } = x;
 
-        self.write(&mut *writer, offset)?;
-        self.write(&mut *writer, length)?;
-        self.write(&mut *writer, shm_manager.as_str())?;
-        self.write(&mut *writer, kind)?;
+        self.write(&mut *writer, data_descriptor)?;
+        self.write(&mut *writer, shm_protocol)?;
+        self.write(&mut *writer, data_len)?;
+        self.write(&mut *writer, watchdog_descriptor)?;
+        self.write(&mut *writer, header_descriptor)?;
+        self.write(&mut *writer, generation)?;
         Ok(())
     }
 }
 
+impl<R> RCodec<Descriptor, &mut R> for Zenoh080
+where
+    R: Reader,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &mut R) -> Result<Descriptor, Self::Error> {
+        let id = self.read(&mut *reader)?;
+        let index_and_bitpos = self.read(&mut *reader)?;
+
+        Ok(Descriptor {
+            id,
+            index_and_bitpos,
+        })
+    }
+}
+
+impl<R> RCodec<HeaderDescriptor, &mut R> for Zenoh080
+where
+    R: Reader,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &mut R) -> Result<HeaderDescriptor, Self::Error> {
+        let id = self.read(&mut *reader)?;
+        let index = self.read(&mut *reader)?;
+
+        Ok(HeaderDescriptor { id, index })
+    }
+}
+
+impl<R> RCodec<ChunkDescriptor, &mut R> for Zenoh080
+where
+    R: Reader,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &mut R) -> Result<ChunkDescriptor, Self::Error> {
+        let segment = self.read(&mut *reader)?;
+        let chunk = self.read(&mut *reader)?;
+        let len = self.read(&mut *reader)?;
+
+        Ok(ChunkDescriptor {
+            segment,
+            chunk,
+            len,
+        })
+    }
+}
+
 impl<R> RCodec<SharedMemoryBufInfo, &mut R> for Zenoh080
 where
     R: Reader,
 {
     type Error = DidntRead;
 
     fn read(self, reader: &mut R) -> Result<SharedMemoryBufInfo, Self::Error> {
-        let offset: usize = self.read(&mut *reader)?;
-        let length: usize = self.read(&mut *reader)?;
-        let shm_manager: String = self.read(&mut *reader)?;
-        let kind: u8 = self.read(&mut *reader)?;
+        let data_descriptor = self.read(&mut *reader)?;
+        let shm_protocol = self.read(&mut *reader)?;
+        let data_len = self.read(&mut *reader)?;
+        let watchdog_descriptor = self.read(&mut *reader)?;
+        let header_descriptor = self.read(&mut *reader)?;
+        let generation = self.read(&mut *reader)?;
 
-        let shm_info = SharedMemoryBufInfo::new(offset, length, shm_manager, kind);
+        let shm_info = SharedMemoryBufInfo::new(
+            data_descriptor,
+            shm_protocol,
+            data_len,
+            watchdog_descriptor,
+            header_descriptor,
+            generation,
+        );
         Ok(shm_info)
     }
 }

commons/zenoh-codec/tests/codec.rs

@@ -325,15 +325,24 @@ fn codec_encoding() {
 #[cfg(feature = "shared-memory")]
 #[test]
 fn codec_shm_info() {
-    use zenoh_shm::SharedMemoryBufInfo;
+    use zenoh_shm::api::provider::chunk::ChunkDescriptor;
+    use zenoh_shm::header::descriptor::HeaderDescriptor;
+    use zenoh_shm::{watchdog::descriptor::Descriptor, SharedMemoryBufInfo};
 
     run!(SharedMemoryBufInfo, {
         let mut rng = rand::thread_rng();
-        let len = rng.gen_range(0..16);
         SharedMemoryBufInfo::new(
+            ChunkDescriptor::new(rng.gen(), rng.gen(), rng.gen()),
             rng.gen(),
             rng.gen(),
-            Alphanumeric.sample_string(&mut rng, len),
+            Descriptor {
+                id: rng.gen(),
+                index_and_bitpos: rng.gen(),
+            },
+            HeaderDescriptor {
+                id: rng.gen(),
+                index: rng.gen(),
+            },
             rng.gen(),
         )
     });

commons/zenoh-shm/Cargo.toml

@@ -28,10 +28,23 @@ categories = { workspace = true }
 description = "Internal crate for zenoh."
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[features]
+test = ["num_cpus"]
+
 [dependencies]
 bincode = { workspace = true }
 log = { workspace = true }
 serde = { workspace = true, features = ["default"] }
 shared_memory = { workspace = true }
 zenoh-buffers = { workspace = true }
 zenoh-result = { workspace = true }
+rand = { workspace = true }
+lazy_static = { workspace = true }
+num-traits = { workspace = true }
+num_cpus = { workspace = true, optional = true }
+thread-priority = "0.15.1"
+lockfree = "0.5.1"
+
+[dev-dependencies]
+zenoh-shm = { workspace = true, features = ["test"] }
+libc = { workspace = true }

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

@@ -0,0 +1,16 @@
+//
+// Copyright (c) 2023 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]>
+//
+
+pub mod shared_memory_client;
+pub mod shared_memory_segment;

commons/zenoh-shm/src/api/client/shared_memory_client.rs

@@ -0,0 +1,29 @@
+//
+// Copyright (c) 2023 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 std::fmt::Debug;
+
+use std::sync::Arc;
+
+use zenoh_result::ZResult;
+
+use crate::api::common::types::SegmentID;
+
+use super::shared_memory_segment::SharedMemorySegment;
+
+// SharedMemoryClient - client factory implementation for particular shared memory protocol
+pub trait SharedMemoryClient: Debug + Send + Sync {
+    // Attach to particular shared memory segment
+    fn attach(&self, segment: SegmentID) -> ZResult<Arc<dyn SharedMemorySegment>>;
+}