feat(pubsub): intra-process delivery hands the subscriber an Arc<T> - #340
Open
YuanYuYuan wants to merge 37 commits into
Open
feat(pubsub): intra-process delivery hands the subscriber an Arc<T>#340YuanYuYuan wants to merge 37 commits into
YuanYuYuan wants to merge 37 commits into
Conversation
The subscriber side already had `with_locality` (it forwards to zenoh's `allowed_origin`). The publisher side had no equivalent, so the one setting that matters for intra-process delivery was unreachable: `Locality::SessionLocal` on a publisher makes zenoh's `resolve_put` skip `primitives.send_push_consume` entirely. No link, no wire encode, no shared memory — the payload goes straight to the matching local subscriber callbacks. Every node created from one ZContext shares one zenoh session, so this reaches sibling nodes in the same process. `Locality` is re-exported from the crate root and the prelude. Callers of the subscriber-side `with_locality` previously had to add a direct `zenoh` dependency to name the type, which also let them pick a different zenoh version than the one hiroz links. This removes the transport, not the serialization: zenoh payloads are bytes, so the message is still CDR-encoded on publish and decoded on receive. The doc comment says so, and says that publisher SHM is wasted work under SessionLocal. Measured on hiroz-bench (viz-grid, two nodes on one ZContext against the two-process baseline): 263 -> 25 microseconds p50 at 64 B. What fails without this: `publisher_locality.rs` does not compile, since `with_locality` is what it calls. As a property detector it does work in both directions — swapping `Locality::SessionLocal` for `Locality::Any` in the test makes it fail on "SessionLocal publisher leaked to another context" (measured, rc=101), and back again makes it pass in 1.18 s. The test publishes on two topics from one context and asserts a control topic DOES cross to a second context over the same router within the same deadline. Without that control, "the other context received nothing" would be indistinguishable from a broken router or a wrong topic.
…zation
Locality::SessionLocal removed the transport but not the encoding: zenoh
payloads are bytes, so CDR still ran on every message even when both endpoints
were in one process. On hiroz-bench that left 64 B at 23 microseconds against
copper's 0.68 — the remainder being CDR plus hiroz's own per-message work.
This adds an opt-in path beside the wire path, the same shape as rclcpp's
intra-process comm:
ZPub::publish_shared(Arc<T>) hands every same-session subscriber
of this exact Rust type a refcount
bump. Nothing is encoded or copied.
ZSubBuilder::build_with_shared_callback registers on the bus AND declares a
zenoh subscriber with
allowed_origin(Remote), so remote
traffic still arrives normally and a
same-session publisher cannot deliver
the same message twice.
ZPubBuilder::with_intra_process_only keeps publish_shared off zenoh
entirely.
local_bus keys entries by (session zid, qualified topic). The zid is what makes
"same session" true rather than merely "same process" — two ZContexts in one
process open two sessions and must not see each other's traffic. Both ZPub and
ZSub already hold an Arc<Session>, so this needs no plumbing through the node
tree.
Callbacks are never invoked under the registry lock. A subscriber callback
commonly publishes, which re-enters the bus; the matching callbacks are cloned
out and the guard dropped first. The map is nested rather than keyed by a tuple
so the publisher does not allocate a String to look itself up, and a single
subscriber is dispatched without allocating a Vec — at 64 B both were a
measurable fraction of the path.
Measured on hiroz-bench at 64 B, p50: 273 microseconds on the wire, 23.3 with
SessionLocal, 5.7 here. At 1 MiB: 1966 / 1371 / 140.
What fails without this: tests/intra_process.rs does not compile, since it calls
the new API. As property detectors the four tests do work:
- same_arc_reaches_a_same_session_subscriber asserts Arc::ptr_eq between the
published and received Arc. Pointer identity cannot survive an encode/decode
round trip, so this is the assertion that distinguishes the fast path from a
correct-looking slow one. Swapping publish_shared for publish makes it fail
with "subscriber did not receive the message" (measured, rc=101) — proof
that the bus, not zenoh, is what delivered.
- intra_process_only_publisher_does_not_reach_another_context carries a
control topic that MUST cross to the second context over the same router
within the same deadline. Without it a zero would be indistinguishable from
a broken router.
- a_different_rust_type_on_the_same_topic_is_not_delivered pins the TypeId
gate.
- dropping_the_subscriber_unregisters_it asserts delivery works first, then
that it stops.
Known limitations, all disclosed in the module docs: the publisher is told
whether to use the wire rather than reading it off the graph, so an
intra-process-only publisher with no local subscriber drops the message instead
of falling back; and every receiver shares one read-only Arc rather than a
unique payload being moved to a sole receiver.
The registry was keyed by (zid, topic), so every publish_shared hashed a fully-qualified ROS key expression — a long string — to find its own subscriber list. At 64 B payloads, where the whole round trip is around a microsecond, two such lookups per round trip are a visible share of it. A publisher and a subscriber now resolve an Arc<Channel> when they are built and never touch the registry again. Publishing is a lock, a TypeId filter and the callback; no hashing, no string. Channels are created on demand and never removed. One empty Channel per (session, topic) ever used is bounded and trivial, and keeping them means a publisher's handle stays valid across a subscriber coming and going — which a Weak-based registry would not give without re-resolving. The single-subscriber case still avoids allocating a Vec, and callbacks are still invoked with no lock held. The four tests in tests/intra_process.rs pass unchanged, including the Arc::ptr_eq identity assertion — this is a lookup change, not a delivery change.
A publisher on the intra-process path delivers an Arc<T> without serializing, but still allocates and fills a payload buffer per send. as_mut_slice is what lets one buffer serve many sends: it hands back this buffer's own bytes when writing them cannot be observed by anyone else, and None otherwise, so the caller allocates only when it must. It delegates to a windowed accessor added to zenoh-buffers, pinned here by a patch until that lands upstream.
A [patch.crates-io] entry is the worst place for a dependency that can move under you: nothing about the build changes when it does. The rev also carries the SHM-pending fix, which the branch tip did not when this was first pinned.
A subscriber built with a shared callback forced its wire half to Locality::Remote, to stop a same-session publish_shared arriving twice. That reasoning only covers publish_shared. A plain publish has no bus delivery to duplicate, so the filter discarded it outright: an ordinary publisher and an ordinary subscriber in one process, both matched in the graph, and nothing ever arrived. Suppression moves to the publisher, which is the only side that knows whether it used both paths. publish_shared now takes the bus only when its wire half cannot also reach this session — with_intra_process_only, or a Remote locality. Otherwise it sends on the wire alone: correct, and not zero-copy, which is the honest trade for a publisher that never said which audience it wanted. Also bounds intra-process delivery depth. Delivery is inline on the publishing thread, so a callback that publishes onto its own topic recursed until the stack ran out. The same shape on the wire is an endless stream of messages, which is survivable. It is now refused past a fixed depth and logged. Fixes #39. Fixes #40.
Delivery held a read lock over the subscriber list, and cloned each matching callback so the guard could be dropped before any callback ran. The clone was not optional: invoking a callback under the guard is the re-entrancy deadlock this workspace has fixed repeatedly. A snapshot removes both. A publisher loads the current list and calls through it; a subscriber coming or going swaps in a new list and leaves a publish already in flight running against the old one. Same visibility the clone gave, no atomics, and nothing is held so it cannot deadlock. Two more refcount pairs go with it. The erased callback now takes the payload by value, so a sole subscriber is handed the only reference rather than a clone of it, and publish_shared moves the message instead of cloning when the wire will not be used.
The two limitations this prototype was filed with. The publisher was told whether to use the wire, so publish_shared on a publisher whose only subscriber was an ordinary one delivered nothing at all. It now reads its audience off the graph per message and takes the bus only when every subscriber that could receive is on it, which is the rule rclcpp uses. with_intra_process_only stays as an explicit override. Every receiver also shared one read-only Arc, with no way to express the move rclcpp performs when a topic has exactly one taker. publish_owned hands a sole owning subscriber the value itself, and hands it back untouched when it cannot, so the caller falls back rather than losing it. Fixes the two known limitations on #36.
The previous change added a graph lookup that publish_shared could not reach: a plain publisher returned at the bus_delivery guard above it, and intra_process_only short-circuited below it, leaving the lookup live only for a Locality::Remote publisher. Three reverts left the new tests green, which is what exposed it. Drop the guard. #39 requires that one message never take both the bus and the wire, and each branch still takes exactly one. It does not require taking the wire unconditionally, which is what #36 filed. Refs #36, #39
Found by adversarial review of #130, all introduced by this branch. An owned subscriber built its wire half as a no-op closure, so every message that did not arrive by publish_owned on the bus was discarded -- including from every remote publisher -- while the subscription still advertised itself to the graph. Both halves now run the same callback. Channel::publish reported depth exhaustion and 'nobody wanted it' as the same zero, and publish_shared read zero as 'fall back to the wire'. The wire re-enters the callback on a zenoh runtime thread, which then publishes from inside that runtime and does not return. It now returns Delivery, and only NoTaker falls through. Routing never consulted the publisher's own locality. A Remote wire half cannot reach this session, so bus and wire address disjoint audiences and both must run; taking the wire alone left same-session subscribers with nothing. The recursion test publishes behind a watchdog. Reverting the second fix makes the publish never return, and a count assertion cannot see that -- the count never gets to climb. The timeout is explicit in the test so the failure names the property instead of reporting as a hung suite. Refs #36
The [patch.crates-io] entry applies only from the root workspace being built, so it never reached a downstream consumer -- who resolved the registry zenoh-buffers and failed on a missing as_mut_slice. cargo publish strips the patch from the published manifest but verifies with the workspace manifest, so verification passed and the published crate would have been broken for everyone, silently. No hiroz code path needs the accessor; it exists for callers that pool their own payload buffers. It is now behind an off-by-default pooled-payload feature, and a workspace that enables it supplies its own patch. The default build resolves released zenoh-buffers. Also document on publish_shared that the bus carries no QoS: transient local is violated rather than unsupported, reliability and history have no meaning without a queue, and the attachment is absent. Refs #130
G2 (#134). bus_can_serve_everyone inferred 'everyone is local' from the ROS liveliness graph, which does not hold: a plain zenoh subscriber on the same keyexpr -- z_sub, a storage or REST plugin, a native recorder -- declares no ROS token and is invisible to it. Such a publisher skipped the wire and that subscriber received nothing, silently. There is no count to subtract our own subscribers from, so the condition cannot be repaired and the inference is withdrawn. The bus is now taken only when the caller has asserted the audience, with with_intra_process_only() or Locality::Remote. This also closes the discovery-lag window (#135): the Remote path always publishes on the wire, so a remote subscriber whose liveliness token has not yet arrived is still served. G1 (#133). TRANSIENT_LOCAL lives in the wire publisher's cache and an intra-process-only publisher has no wire, so serving the bus would let a late-joining subscriber be handed a history the message is missing from. That combination is refused. The Remote path is unaffected: its wire publish populates the cache as before. Refs #132, #133, #134, #135
…hrough the re-entrancy check Four adversarial reviews. Three defects and two gaps. A panicking subscriber callback unwound into the application's publishing thread and skipped every subscriber after it in the snapshot. On the wire a panic kills one zenoh task; synchronous bus delivery made it a blast-radius regression, and delivery order is snapshot order, so which siblings were censored varied between runs. Each callback now runs inside catch_unwind, is logged, and delivery continues. The crate contract says every user-code invocation routes through the re-entrancy assertion so that calling out under a tracked lock is caught in debug builds. The bus routed none, while making that hazard more reachable than the path it replaces, not less. Both call sites now do. publish_owned returned Err(payload) for depth exhaustion as well as for 'no owning receiver'. The caller falls back on that, which for a Locality::Remote publisher put a message on the wire that the caller asked to hand to one local owner. Depth exhaustion is now a deliberate drop, matching publish at the same depth. Also: the doc claim that a dropped subscriber cannot be called back into was false. Drop unregisters, it does not quiesce, and a delivery already in flight runs to completion. That is memory-safe because the snapshot owns the closure and its captures, but it is not a barrier, and callers were told it was. The registry doc undercounted what it retains: the outer key is a per-context ZenohId, so the session dimension is unbounded. Tests: a positive control for the type-mismatch test, which a dead bus satisfied; the fan-out branch, which had no coverage at all and which a stub calling only the first subscriber would have passed; publish_owned's missing durability refusal, which is a live defect with a failing baseline today; and MAX_DELIVERY_DEPTH is exported so the recursion test asserts the exact bound instead of a ceiling twice its size. Refs #132, #133
…d could not say why
Three defects on one call path, found by an adversarial audit.
publish_owned on a Locality::Remote publisher took the bus and returned,
never touching the wire. publish_shared gets this right one method away,
and its comment says why: the two routes address disjoint audiences and
BOTH must run. The move cannot run both - the wire half needs the value in
order to serialize it - so a Remote publisher now shares instead of moving.
That also closes the durability hole. refuse_durable_bus permits a
TRANSIENT_LOCAL Remote publisher on the grounds that the wire still runs,
which was false on exactly this path, so a late joiner was served a history
the message was missing from.
And both methods returned usize, collapsing NoTaker, DepthExceeded and
'there is no bus on this publisher' into 0. Delivery exists to keep the
first two apart: a caller may fall back to the wire on NoTaker, and loops
forever if it does so on DepthExceeded. Both now return Published { Wire,
Bus(Delivery), BusAndWire(Delivery) }, and the bus's own publish_owned
returns Delivery rather than reporting a depth-drop as a delivery.
Every publish_owned test used with_intra_process_only(), which is the
failure mode #132 warns about in as many words.
publish_shared returns Published, which contains Delivery, so a crate private Delivery is a private type in the public API. The same fix landed later in the stack when local_bus became a pub module; it belongs here now that the publisher's return type exposes it.
publish_owned_on_a_remote_publisher_still_reaches_the_wire is the detector that did not exist. Every other publish_owned test used with_intra_process_only(), so none could see the Remote arm take the bus and return. The far-subscriber assertion is the point; the near one passes against the defect. The other three pin what a count could not express: Wire against Bus(NoTaker) - a plain publisher never asks the bus, an asserted one asks and finds nobody - and a depth refusal reported as dropped rather than as one receiver taking it.
publish_shared now returns Published, so twelve assertions had to change. Each was converted by what its test is actually about rather than by a blanket rewrite: the three that asserted zero become exact, because zero was the value that hid the reason. Two of them are Published::Wire - the bus was never asked - and one is Bus(NoTaker), where it was asked and found nobody. A count could not tell those apart, which is the defect this change exists to fix.
The Remote arm of publish_moved shared the value instead of moving it, so the wire ran but the shared path filters on is_shared() and never reached an owned subscriber. Publish to the wire first from a reference, then give the value away.
The two publishing methods each decided the route themselves, in the
opposite order, so a publisher carrying both assertions routed one way
through publish_shared and the other through publish_owned. That is the
drift that produced the Remote defect, in a configuration nothing
covered. Route::{BusOnly,BusAndWire,WireOnly} is resolved in one place
and both methods match on it; a test pins the precedence.
The publish_shared table still promised a count and still described the
graph inference that was withdrawn. Qualify thirteen bare issue refs
that number hiroz-bench issues: hiroz has its own #36 and #40, so each
one silently pointed at a stranger.
An edit left the head of build_with_shared_callback's documentation stranded above build_with_owned_callback, so rustdoc showed readers of the owned method a table of two registrations it does not make, and left the shared method with only the trailing paragraph. The sentence about the origin filter had also been spliced mid-clause and contradicted the table two lines above it; it now states what the code does.
cargo doc exits 0 over an unresolved link, so these passed every gate. A path in a //! doc resolves at the pub mod line, where the module's own types are not in scope, which is why TypeId and Channel failed there.
hiroz never called it. It existed so an out-of-tree caller could rewrite a ZBuf in place, but ZBuf is a newtype with a public field and a DerefMut, so such a caller reaches zenoh's opt_mut_slice directly and needs nothing from hiroz. What it cost to keep: the feature's patched zenoh-buffers lives on a tailnet-only forge, so CI could not compile the test behind the gate. A cfg-gated suite that never builds is the documented way for a test to rot unnoticed. Removing it makes 'hiroz needs no patched zenoh' unconditional rather than feature-qualified.
… publish on the wire first Two defects found by adversarial review, each reported independently by more than one reviewer. An owned subscriber was starved in silence. Channel::publish filters on is_shared(), so it is invisible to the shared path, and an intra_process_only publisher has no wire behind the bus. A registered receiver of the right type got nothing while the caller was told Ok(Bus(NoTaker)) - reachable through publish_shared with an owned subscriber, and through publish_owned whenever there is not exactly one owning receiver. It cannot be repaired by serving a clone, because ZMessage is Send + Sync + Sized and not Clone, so the publisher refuses instead - the way refuse_durable_bus already refuses a durable publisher rather than quietly violating its contract. The two methods also ordered bus and wire oppositely. publish_shared ran the bus first, so a wire failure returned Err after every local subscriber had already been called, and Result<Published> has no partial-success value to say so: a caller that retried delivered twice locally. Both now publish on the wire first, which is the order publish_owned already argued for in its own comment. Also drops leftovers from the feature removal: an orphaned comment that had re-attached itself to config-builders, which is on by default and so told a reader the default build needs a patched zenoh-buffers, and a stray blank line that put zbuf.rs in the diff for no reason.
… reclaim channels Three more findings from the adversarial reviews. The panic isolation swallowed the crate's own re-entrancy assertion. It reports by panicking, and a nested delivery raises it from inside the outer catch_unwind - so the detector built to catch 'a user callback ran while a lock was held' was downgraded to a log line, in exactly the case the bus makes most reachable: a callback that publishes. It is now re-raised in debug builds, told apart from a user panic by a stable prefix rather than by guessing. Delivery::Sent(n) counted subscribers invoked, not subscribers that returned. A sole subscriber panicking on every message reported Sent(1) forever, so a caller falling back on NoTaker could not see a total delivery failure. The registry only ever grew. channel() runs in every publisher build, so this was one entry per topic for the process lifetime, not one per ZContext as #152 says. It now reclaims what no endpoint holds, which is safe only because a strong count of one proves no ZPub or ZSub can still reach the channel.
…aph token
The previous commit did not compile: the reentrancy import landed inside
a use std::{..} brace group, and the new constant was inserted between
an #[inline(always)] attribute and the function it belongs to. Both are
mechanical slips from anchor-based editing, and ran=0 was the tell - not
the exit code, which cargo returns for a build failure and a test
failure alike.
Also documents the two review findings that are deliberately not code
changes. MAX_DELIVERY_DEPTH now says what it does not bound: it is
thread-local, so a callback that spawns a thread escapes it, and a topic
cycle across the wire is not bounded at all. Suppressing the wire half
for nested deliveries was considered and rejected - a nested publish is
a distinct message, so dropping it trades a loud problem for a silent
one. And with_intra_process_only now records that such a publisher still
declares a liveliness token, so other nodes' publisher counts and
matched events include an endpoint that can reach nobody.
Both tests pinned the old Delivery::Sent semantics - one asserted Sent(1) for a sole subscriber that panicked, the other Sent(3) where two of three returned. Under the corrected contract those are NoTaker and Sent(2), and the counter assertions are what carry the real property in each case. The duplicate test added alongside the fix is dropped: the existing sole-subscriber test covers it once its assertion is right.
It read 'subscribers that returned normally, not returned normally' - the replacement rewrote the first line and left the second, so the line documenting what Delivery::Sent counts said nothing.
YuanYuYuan
changed the title
There was a problem hiding this comment.
🟡 Changes recommended
Locality filtering, panic accounting, registry reclamation, and several concurrency-sensitive tests have unresolved correctness issues.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds an opt-in intra-process bus that delivers shared ROS messages as Arc<T> without serialization.
Changes:
- Adds local shared/owned delivery with routing and recursion safeguards.
- Adds publisher/subscriber locality APIs.
- Adds extensive integration coverage.
File summaries
| File | Description |
|---|---|
crates/hiroz/src/reentrancy.rs |
Identifies re-entrancy violations. |
crates/hiroz/src/pubsub.rs |
Integrates bus routing and callbacks. |
crates/hiroz/src/prelude.rs |
Re-exports new API types. |
crates/hiroz/src/parameter/service.rs |
Initializes new builder fields. |
crates/hiroz/src/node.rs |
Initializes publisher routing defaults. |
crates/hiroz/src/local_bus.rs |
Implements the intra-process bus. |
crates/hiroz/src/lib.rs |
Exposes the bus and locality API. |
crates/hiroz/src/dynamic/tests/pubsub_tests.rs |
Updates dynamic builder fixtures. |
crates/hiroz/Cargo.toml |
Adds arc-swap. |
crates/hiroz-tests/tests/publisher_locality.rs |
Tests publisher locality. |
crates/hiroz-tests/tests/intra_process.rs |
Tests bus behavior and routing. |
crates/hiroz-tests/Cargo.toml |
Documents feature forwarding. |
Review details
Suppressed comments (1)
crates/hiroz-tests/tests/intra_process.rs:981
- This second unsynchronized mutation of the process-global panic hook can interleave with the earlier panic test and restore the wrong hook. Please route both tests through one locked RAII helper (or leave the hook unchanged) so parallel test execution cannot contaminate the rest of the binary.
let prev = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let delivered = publisher.publish_shared(Arc::new(RosString {
data: "boom".to_owned(),
}));
std::panic::set_hook(prev);
- Files reviewed: 12/12 changed files
- Comments generated: 11
- Review effort level: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
The re-entrancy re-raise is behind cfg(debug_assertions), so in a release build both the REENTRANCY_VIOLATION import and the caught payload are unused - errors under -D warnings, which is what CI builds with. My local gate could not have caught this: it builds debug, where both are live. Gate the import and underscore the binding.
…nting fixes From Copilot's review. Each was real. The panic classifier matched a public message prefix, so a subscriber could panic with a String beginning with it and force resume_unwind - defeating the isolation the containment exists to provide. It is now a private type no code outside the crate can construct. publish_moved discarded invoke_isolated's result and reported Sent(1) for a callback that panicked; the shared path was fixed this morning and the owned path was missed. A fan-out where every callback panicked returned Sent(0) while the sole-subscriber arm returned NoTaker for the same outcome, so a caller branching on NoTaker missed one spelling. Reclamation swept only the session being touched, so each retired session kept its final channel for the process lifetime - the session-churn half of the leak rather than a fix for it. A Locality::Remote subscriber was still registered on the same-session bus, receiving exactly what its own allowed_origin filter excludes. The wire half honoured the restriction; the bus half ignored it. Also removes two doc claims that still pointed at the withdrawn graph inference.
The Remote-subscriber guard read self.locality after build_with_callback had taken self by value. Also strengthens three tests Copilot flagged as unable to detect what they were named for: the ordering test now induces a real wire failure by closing the session and asserts no local subscriber ran, the registry test is serialised and uses saturating arithmetic against a process-global counter, and the two panic tests are serialised because the panic hook is process-global.
clippy::manual_non_exhaustive, and the attribute is the better spelling: it blocks external construction the same way, without a field that reads as an accident.
should_panic can only match a string payload, so the typed payload broke it. Matching the message is also what made the old classifier forgeable - the type is both stronger and what local_bus keys on.
Three lints CI caught and my gate could not: collapsible_if twice and type_complexity on the erased republish hook. The gap is that I linted 'cargo clippy -p hiroz --all-targets', which covers hiroz's own targets and not the hiroz-tests crate at all - the documented default-members blind spot. The gate now lints both.
These pointed at an issue tracker that is not this repository, so on a public remote they name nothing a reader can follow. Where the sentence carried meaning beyond the pointer it is kept; where the pointer was the sentence it is removed.
The last clippy::type_complexity CI reported. Same cause as the one in intra_process: ZSub carries a serializer parameter, so spelling it at a return position trips the lint.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
A same-session publisher hands the subscriber an
Arc<T>instead of serializing it over zenoh, and a pooled payload buffer removes the per-send allocation.Measured as part of a decomposition against copper-rs and rclcpp, in which each step prices exactly one thing.
Note
This PR was split. It now carries the intra-process bus only — +2103/−0 across 12 files. The pool is #341 and the publish API is #342, each stacked on this one and reviewable on its own diff.
Before and after
publishpublish_sharedArc::ptr_eqholds)with_intra_process_only()orLocality::RemoteAt 64 B / 200 Hz, sweep F: 106.4 µs on the wire → 3.79 µs with
Locality::SessionLocal→ 0.40 µs on the bus. Pooling then takes 1 MiB from 54.1 µs to 0.42 µs.Does this need a zenoh fork?
No — unconditionally, with no feature qualifier. The manifest has no
[patch.crates-io]and asks for releasedzenoh 1.9.0/zenoh-buffers 1.9.0. Verified by building and testing against the registry crate.An earlier revision carried a
pooled-payloadfeature wrapping an unreleasedZSlice::opt_mut_slice(raised separately against zenoh). It is removed. hiroz never called it, and it was not what gave anyone access:hiroz::ZBufis a newtype with a public field and aDerefMut, so a workspace that patcheszenoh-buffersreaches the accessor directly. The wrapper added a feature flag and no capability.Note
Keeping it had a cost worth naming. The patched crate lives on a private forge, so CI could not compile the one test behind the gate — a
#[cfg(feature)]suite that never builds is the standard way for a test to rot unnoticed. The suite is now 23 of 23 with nothing gated, rather than 23 of 24.Defects found by review, and fixed here
Four adversarial reviews (soundness, correctness, compatibility, concurrency) after the branch was first pushed. Each fix has a detector proven to fail without it.
|_m| {}— every message from off-session discardedwith_intra_process_only(), so the wire half never ranLocality::Remoteplus a mixed audience lost the same-session subscriberself.localitypublish_ownedreached the bus around the transient-local refusalpublish_ownedon aLocality::Remotepublisher took the bus and returned — every off-session subscriber got nothingwith_intra_process_only(), so the Remote arm was never reachedis_shared()intra_process_onlypublisher has no wire behind the buspublish_sharedran bus-then-wire whilepublish_ownedran wire-then-bus, so a wire failure returnedErrafter local subscribers had runLocality::RemotepublisherThree of those are the same defect twice.
publish_sharedandpublish_ownedeach decided the route themselves, so a fix to one did not reach the other and a configuration only one of them handled went unnoticed. The route is now resolved in one place and both methods match on it, so they cannot disagree; a test pins the precedence, and flipping it reds that test alone.A design decision was also withdrawn.
bus_can_serve_everyoneinferred "everyone is local" from the ROS liveliness graph. A plain zenoh subscriber declares no token, so it is invisible there and its messages were silently dropped — and with no count to subtract our own subscribers from, the predicate is unanswerable rather than miscoded. The bus is now taken only when the caller asserts the audience. Cost: a plain publisher no longer gets zero-copy automatically.Evidence
cargo test -p hiroz-tests --test intra_process --features ros-msgs,jazzy— 26 tests, compile-gated so an empty binary cannot read as a pass. Reverting each fix reds exactly one test, none overlapping.Each revert is reported with its own compile-error count, because
cargoexits101for a build failure and a test failure alike — a revert that never compiled is otherwise indistinguishable from a detector firing. Every red below built cleanly (errors=0) and ran the same 26 tests.Arc::ptr_eqseparates the bus from the wire (a CDR round trip must allocate). The owned test asserts mutation, which a sharedArc<T>cannot express. The recursion test publishes behind a watchdog, because reverting that fix makes the publish never return and a count assertion cannot see it.What the bus is worth
At 64 B / 200 Hz, sweep F on a pinned host, median of three interleaved runs:
hiroz— zenoh, two processeshiroz(local)—Locality::SessionLocal, no transport, still CDRhiroz(zc)— theArc<T>bus, no transport, no CDRRemoving the serialization is worth more than removing the transport, in every cell of the grid.
For context, rclcpp's own intra-process path reads 13.0 µs at 64 B on the same host, and the shape across payloads — 13.0 → 20.5 → 53.1 µs for a 16,384× payload increase — says that gap is two executor wake-ups per round trip, not the handover. rclcpp queues into a ring buffer and trips a guard condition; this bus calls the subscriber inline on the publishing thread. That is a design trade, not a quality gap: inline delivery is exactly what forced the depth guard and the panic isolation above.
Known limitations
TRANSIENT_LOCAL+intra_process_onlyros2/rclcpp#2303) — refusing is a choice, not a necessitySessionLocalpublisher is invisible off-processros2 topic echosees the topic and no dataZenohId, so repeatedZContextcreation leaks a channel map per sessionpanic = "unwind"panic = "abort"gets noneOpen items are tracked on a meta issue with four children.
Breaking changes
local_bus::channel(..)runs in everyZPubBuilder::build()RwLockwrite on first use of a topicThat one is not opt-in, which is why it is stated first. The API additions themselves are —
publish(&T)is untouched, and a publisher that never callspublish_sharedorpublish_ownedbehaves exactly as before.The
publishsignature change belongs to the stacked #342.Note
Four adversarial reviews ran against
2ff4f5c0— concurrency, contract-vs-code, test quality, ROS integration. Two defects were fixed here; three of the four reviewers found the owned-subscriber starvation independently. The starvation cannot be repaired by serving a clone, becauseZMessageisSend + Sync + Sizedand notClone, so the publisher refuses — the same stancerefuse_durable_busalready takes. Reverting the refusal reds exactly the two new tests at 0 compile errors.Findings deliberately not fixed here, as disclosure rather than code: a nested delivery's re-entrancy assertion is swallowed by the panic isolation's
catch_unwind;Delivery::Sent(n)counts subscribers invoked, not subscribers that returned; the depth guard bounds the stack while aBusAndWirepublisher still emits one wire message per nesting level; the registry leaks per topic, not merely per session (the follow-up issue understates it); and the liveliness token is declared before locality is consulted, so a bus-only publisher makes other nodes' publisher counts and matched-events wrong.