SHM handoff - #2646
Conversation
…nge and counter lease
[] Conflicts: io/zenoh-transport/src/lib.rs io/zenoh-transport/src/manager.rs
|
Any progress there ? |
Hi @doisyg ! This is still WIP, ETA this week. Not yet testable. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #2646 +/- ##
==========================================
+ Coverage 74.48% 74.69% +0.20%
==========================================
Files 423 425 +2
Lines 63056 63725 +669
==========================================
+ Hits 46967 47598 +631
- Misses 16089 16127 +38 ☔ View full report in Codecov by Harness. |
|
Hi @doisyg ! You may try SHM handoff PR since now. I'm still making some tests for it, so not guaranteed to be bug-free, but should work. |
Concurrency issue in
|
| challenge: u64, | ||
| version: u64, | ||
| protocols: [ProtocolID; 256], | ||
| shm_counters: [AtomicU32; 762], |
There was a problem hiding this comment.
I understand that within a link 1 SHM counter is used per-channel and hence per-priority. Meaning 8 counters per-link, right ?
762 counters allow to handle 95 links, limiting the number of peers in a p2p system to 96 peers.
That will not be sufficient for some ROS 2 systems where more than 200 peers are deployed.
There was a problem hiding this comment.
I was planning to make growing counter segment based on overcommit mechanics.
Now I made fixed-size array for 351 peer
|
|
||
| let c_task = task.clone(); | ||
| let c_token = token.clone(); | ||
| ZRuntime::Net.spawn(async move { |
There was a problem hiding this comment.
Does this mean a new task is created for each SHM counter ?
In large p2p systems with many links, might this result in too many tasks that will wake-up every 100ms ?
On the other hand, 100ms sleep seems quite large in case of high frequency of messages. At 100Hz it means 10 messages are retained per-cycle in SHM before to be cleaned-up.
There was a problem hiding this comment.
Made one single global task
| let smb = shmr.read_shmbuf(shmbinfo)?; | ||
|
|
||
| // Handle RX handoff | ||
| handoff.on_rx(priority); |
There was a problem hiding this comment.
If the deserialization of shminfo fails, this handoff is never called, causing a leak.
Conflicts: io/zenoh-transport/src/lib.rs io/zenoh-transport/src/manager.rs io/zenoh-transport/src/unicast/universal/link.rs
- few improvements around SHM codbase - fix handoff counter leakage in case of SHM slice mount error - increase SHM counters count to handle 352 priority-enabled peers - improve shm counters memory writes - fix SHM counters zero issue
Fixed! Made priority channel locking |
Conflicts: io/zenoh-transport/src/common/shm/interop.rs io/zenoh-transport/src/common/shm/shm_context.rs io/zenoh-transport/src/multicast/tx.rs io/zenoh-transport/src/shm.rs io/zenoh-transport/src/shm_context.rs io/zenoh-transport/src/unicast/lowlatency/tx.rs io/zenoh-transport/src/unicast/universal/tx.rs
There was a problem hiding this comment.
Pull request overview
This PR implements a shared-memory (SHM) buffer handoff protocol for reliable unicast transports so SHM-backed payloads remain valid while queued/in-flight, using per-priority shared atomic counters and transactional enqueue semantics integrated into SHM mapping and transport establishment.
Changes:
- Extends SHM negotiation to exchange metadata-segment IDs, challenges, and per-priority counter identifiers (breaking wire-format/layout changes).
- Adds TX/RX handoff tracking and transactional commit/rollback into unicast universal + low-latency paths and SHM interop mapping.
- Adds a stalled-queue SHM regression test to validate buffers remain SHM-backed under backpressure/queue stalls.
Reviewed changes
Copilot reviewed 45 out of 46 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| zenoh/src/net/runtime/mod.rs | Updates SHM provider init state import to new transport SHM module location. |
| io/zenoh-transport/tests/unicast_shm.rs | Adds stalled-queue test coverage to ensure SHM buffers survive prolonged queuing. |
| io/zenoh-transport/src/unicast/universal/tx.rs | Integrates SHM partner mapping with handoff transaction commit on successful enqueue. |
| io/zenoh-transport/src/unicast/universal/transport.rs | Updates SHM context import path for universal unicast transport. |
| io/zenoh-transport/src/unicast/universal/rx.rs | Threads per-link RX SHM handoff into SHM mounting on receive. |
| io/zenoh-transport/src/unicast/universal/link.rs | Adjusts RX loop to pass link wrapper (incl. SHM state) into message reading. |
| io/zenoh-transport/src/unicast/mod.rs | Updates SHM config import path. |
| io/zenoh-transport/src/unicast/manager.rs | Updates SHM context/config imports to new common SHM module structure. |
| io/zenoh-transport/src/unicast/lowlatency/tx.rs | Removes legacy low-latency TX scheduling path (replaced by link-based send). |
| io/zenoh-transport/src/unicast/lowlatency/transport.rs | Simplifies schedule path to use send() and updates SHM context import. |
| io/zenoh-transport/src/unicast/lowlatency/rx.rs | Adds RX handoff integration during SHM buffer mapping in low-latency receive path. |
| io/zenoh-transport/src/unicast/lowlatency/mod.rs | Drops low-latency tx module after refactor. |
| io/zenoh-transport/src/unicast/lowlatency/link.rs | Implements low-latency send with SHM mapping + handoff commit and updated RX plumbing. |
| io/zenoh-transport/src/unicast/link.rs | Adds per-link SHM handoff configuration/state to transport links. |
| io/zenoh-transport/src/unicast/establishment/open.rs | Refactors SHM establishment FSM to produce transport SHM config + per-link handoff config. |
| io/zenoh-transport/src/unicast/establishment/ext/shm/segment.rs | Introduces SHM transport metadata segment layout including counter pool. |
| io/zenoh-transport/src/unicast/establishment/ext/shm/mod.rs | Adds SHM establishment submodules (auth/handoff/segment). |
| io/zenoh-transport/src/unicast/establishment/ext/shm/handoff.rs | Implements TX/RX handoff channels, storage, and transactional semantics. |
| io/zenoh-transport/src/unicast/establishment/ext/shm/auth.rs | Expands SHM extension from scalar challenge to buffer payload carrying handoff establishment data. |
| io/zenoh-transport/src/unicast/establishment/cookie.rs | Updates cookie SHM accept/open state type paths after SHM FSM refactor. |
| io/zenoh-transport/src/unicast/establishment/accept.rs | Mirrors open-side SHM FSM refactor on accept side; extracts link handoff config. |
| io/zenoh-transport/src/multicast/tx.rs | Updates SHM partner mapping call site to new signature (handoff disabled for multicast). |
| io/zenoh-transport/src/multicast/transport.rs | Updates multicast SHM context import path. |
| io/zenoh-transport/src/multicast/rx.rs | Updates SHM RX mapping call site to new signature (handoff disabled for multicast). |
| io/zenoh-transport/src/multicast/establishment.rs | Updates multicast SHM context construction path. |
| io/zenoh-transport/src/manager.rs | Updates SHM context import path for transport manager. |
| io/zenoh-transport/src/lib.rs | Removes legacy shm/shm_context modules from crate root. |
| io/zenoh-transport/src/common/shm/shm_context.rs | Makes SHM contexts public(crate) and transitions to new common SHM module layout. |
| io/zenoh-transport/src/common/shm/mod.rs | Adds common SHM module entrypoint and re-exports provider init state. |
| io/zenoh-transport/src/common/shm/interop.rs | Updates SHM mapping interop to support handoff transactions and per-link RX acknowledgements. |
| io/zenoh-transport/src/common/shm/handoff.rs | Adds shared priority-indexed container used by handoff configuration. |
| io/zenoh-transport/src/common/mod.rs | Exposes common SHM module behind shared-memory feature. |
| io/zenoh-transport/Cargo.toml | Adds lockfree and static_init dependencies for SHM handoff reactor/storage. |
| commons/zenoh-shm/src/watchdog/confirmator.rs | Adjusts confirmed descriptor bookkeeping to use watchdog ownership consistently. |
| commons/zenoh-shm/src/version.rs | Bumps SHM metadata/layout version to reflect new transport metadata segment format. |
| commons/zenoh-shm/src/shm/mod.rs | Adds Eq/PartialEq based on segment ID to support equality checks. |
| commons/zenoh-shm/src/posix_shm/struct_in_shm.rs | Adds Eq/PartialEq impls for StructInSHM. |
| commons/zenoh-shm/src/posix_shm/segment.rs | Derives Eq/PartialEq for POSIX SHM segment wrapper. |
| commons/zenoh-shm/src/posix_shm/mod.rs | Makes segment module public and adjusts module exports. |
| commons/zenoh-shm/src/metadata/descriptor.rs | Exposes watchdog field for hard-reference construction. |
| commons/zenoh-shm/src/lib.rs | Adds ShmBufHardRef type to retain confirmed ownership of SHM allocations. |
| commons/zenoh-protocol/src/transport/open.rs | Changes SHM open extension wire type from z64 to zbuf (breaking protocol change). |
| commons/zenoh-codec/src/core/shm.rs | Adds codecs for SHM segment/StructInSHM identifiers (open/reopen on decode). |
| commons/zenoh-codec/Cargo.toml | Adds rand dependency required by new codecs. |
| Cargo.toml | Adds lockfree to workspace dependencies. |
| Cargo.lock | Records new lockfree/owned-alloc/static_init dependency resolution. |
Suppressed comments (3)
io/zenoh-transport/src/common/shm/interop.rs:311
- In
map_zmsg_to_partner, theResponsebranch gates implicit SHM optimization onpolicy.query, but replies/errors should be controlled bypolicy.reply(as documented inShmOptimizationPolicy). As written, the Reply/Err categories will ignore thereplyflag and instead follow thequerysetting.
io/zenoh-transport/src/common/shm/shm_context.rs:24 - This import block pulls
LazyShmProvider/*TransportShmConfigfrom bothcrate::common::shm::interopandcrate::shm, and also importsAuthUnicasttwice (auth::AuthUnicast, AuthUnicast). This is currently a name collision and also referencescrate::shmeven though the transport crate no longer exposes ashmmodule inlib.rs.
io/zenoh-transport/src/common/shm/interop.rs:439 - RX handoff acknowledgement is performed unconditionally for every
ShmPtrslice, even whenmap_zslice_to_shmbuffails. This can decrement the sender's handoff counter despite a mount/validation error, violating the invariant that counters are decremented only after a successful mount.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| fn protocols(&self) -> &[ProtocolID] { | ||
| &self.protocols[..self.id_count as usize] | ||
| } |
| pub fn new_rx(segment: &Arc<RXAuthSegment>, ids: HandoffCounterIds) -> Self { | ||
| match ids { | ||
| HandoffCounterIds::Disabled => Self::Disabled, | ||
| HandoffCounterIds::PerPrio(prio_container) => { | ||
| let prio_container = prio_container | ||
| .map(|counter_id| ShmRXCounterLease::new(segment.clone(), counter_id)); | ||
| Self::PerPrio(prio_container) | ||
| } | ||
| } | ||
| } |
| // SAFETY: this is safe because we store Arc to inner together with this reference. Should | ||
| // track that reference never leaves LockedTxHandoff | ||
| let lock: std::sync::MutexGuard<'static, VecDeque<ShmBufHardRef>> = | ||
| unsafe { std::mem::transmute(lock) }; | ||
| LockedTxHandoff { inner, lock } |
🏷️ Label-Based Checklist
Based on the labels applied to this PR, please complete these additional requirements:
Labels:
new feature,breaking-change,release🆕 New Feature Requirements
Since this PR adds a new feature:
Consider: Can this feature be split into smaller, incremental PRs?
💥 Breaking Change Requirements
Since this PR contains breaking changes:
Note: Breaking changes should be discussed with maintainers before implementation.
Instructions:
- [ ]to- [x])This checklist updates automatically when labels change, but preserves your checked boxes.
Summary
This PR introduces a reliable shared-memory buffer handoff mechanism for unicast transports.
The handoff ensures that a shared-memory buffer remains valid while its corresponding network message is waiting in a transport queue or otherwise in flight. The sender retains a hard reference to every transmitted SHM buffer until the receiver has successfully mounted that buffer.
This prevents SHM buffers from being reclaimed or invalidated when message delivery is delayed by:
Problem
SHM payloads are transmitted as descriptors rather than copied into the transport message.
Previously, the sender could release its last strong reference after the message had been submitted to the transport. The transport message could then remain queued for longer than the lifetime protected by the normal SHM watchdog/reference mechanism.
In that situation:
A successful enqueue therefore cannot be treated as completion of SHM ownership transfer. Ownership must remain with the sender until the receiver has actually opened and mounted the buffer.
Solution
The PR implements an explicit handoff protocol backed by shared atomic counters.
For every reliable transport direction and priority:
The sender leases a counter from its SHM transport metadata segment.
The counter identifier is exchanged with the peer during transport establishment.
Before enqueueing a message containing SHM buffers, the sender:
If the message is accepted by the transport pipeline, the transaction is committed.
If enqueueing fails or the operation is cancelled, the transaction is rolled back automatically.
The receiver mounts each SHM buffer from its descriptor.
Only after a successful mount does the receiver decrement the sender's shared counter.
A sender-side cleanup task observes the counter and releases acknowledged hard references in FIFO order.
sequenceDiagram participant TX as Sender participant C as Shared counter participant Q as Transport queue participant RX as Receiver TX->>TX: Create ShmBufHardRef TX->>C: Increment pending count TX->>Q: Enqueue message and SHM descriptor alt Enqueue succeeds TX->>TX: Commit handoff transaction Q->>RX: Deliver message RX->>RX: Open and mount SHM buffer RX->>C: Decrement pending count TX->>TX: Drop acknowledged hard reference else Enqueue fails TX->>C: Roll back counter increment TX->>TX: Drop staged hard reference endThe mechanism uses shared-memory atomics after transport establishment, so it does not require an additional acknowledgement message for every transmitted SHM buffer.
Transport metadata segment
The existing SHM authentication segment is expanded into a transport metadata segment containing:
Each process creates a local metadata segment and advertises its segment identifier during SHM negotiation.
The peer opens this segment and validates:
Counter identifiers are then exchanged during the transport open phase.
Establishment flow
SHM establishment now exchanges the information required for both authentication and handoff.
InitSynAlice sends her SHM transport metadata segment identifier.
InitAckBob:
OpenSynAlice:
OpenAckBob:
After the exchange, each endpoint has:
If the handoff state cannot be completely established, SHM handoff is disabled for that link.
Reliable links and priorities
Handoff tracking is enabled only for reliable communication.
Reliable transports can maintain separate queues for different priorities. A single counter for the entire link would not preserve the ordering relationship between those queues.
The implementation therefore allocates one handoff channel per transport priority. Each channel has:
This keeps buffer release ordering aligned with the queue in which the corresponding message was transmitted.
Best-effort communication does not use the handoff mechanism.
Transactional enqueue semantics
SHM references and counter increments are managed through a handoff transaction.
When a message is mapped for transmission, each SHM slice added to the message is registered with the current transaction.
The transaction remains uncommitted until the transport pipeline accepts the message.
Successful enqueue
When the message is accepted:
Failed or cancelled enqueue
When the message is rejected, dropped or cancelled:
The transaction automatically cancels itself when dropped without an explicit commit, which protects early-return and error paths.
Receive-side semantics
When receiving a message containing an SHM descriptor, the receiver:
The counter is decremented only after the buffer has been mounted successfully.
Consequently, a malformed descriptor, unsupported protocol or mount failure cannot incorrectly tell the sender that ownership has been transferred.
Hard references
The PR adds
ShmBufHardRef, which retains the confirmed SHM metadata/watchdog ownership required to keep an allocation valid.Hard references can be created from the supported SHM buffer wrapper types and are stored by the TX handoff queues while buffers are in flight.
The actual payload is not copied. The reference exists only to prevent premature SHM invalidation or reuse.
Sender-side cleanup
Committed references are stored in FIFO order.
A periodic cleanup task compares:
When the receiver decrements the counter, the cleanup task removes the corresponding number of references from the front of the FIFO.
This means:
If a receiver remains connected but never mounts a buffer, the reference intentionally remains retained. This favors memory safety over prematurely reclaiming an allocation that may still be consumed.
Main implementation changes
zenoh-shmShmBufHardRef.SHM codecs
Transport protocol
Unicast transport
Common transport utilities
Failure and lifecycle behavior
The implementation is designed to preserve the following invariants:
Compatibility
The SHM metadata version is increased because the shared segment layout and establishment data have changed.
Peers must agree on the SHM metadata version before using the new metadata segment. An incompatible peer must not interpret the new segment layout or use its counters.
The SHM open-extension wire representation also changes to carry the expanded negotiation payload.
These changes affect SHM negotiation rather than ordinary non-SHM payload transport. No application-level API changes are required to benefit from the handoff mechanism.
Performance and resource impact
For each reliable transport direction, the handoff introduces:
Counter acknowledgements are direct shared-memory atomic operations and do not generate additional network messages.
Best-effort links and messages without SHM buffers do not retain SHM references through this mechanism.
Memory use is proportional to the number of SHM slices that have been committed to the transport but have not yet been acknowledged by the receiver.
Reviewer focus areas
The most important areas to review are:
Expected result
After this change, an SHM buffer queued on a reliable unicast transport remains valid until the receiver has successfully mounted it, regardless of how long the message spends waiting in the transport queue.
This preserves zero-copy SHM delivery while preventing premature buffer invalidation under backpressure or prolonged queue stalls.