From 18cf20e7250db55900434d0ec633a4df2e198e50 Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Wed, 29 Jul 2026 18:27:47 +0800 Subject: [PATCH 1/3] perf(py): decode subscriber samples without copying the payload The Python callback path went through `RawBytesCdrSerdes::deserialize`, whose `Output` is an owned `RawBytesMessage` carrying no lifetime, so the whole payload was `to_vec()`d before the closure ran -- one full copy per message, scaling with payload size, discarded as soon as msgspec had decoded out of it. Adds `ZSubBuilder::build_with_sample_callback`, which hands the callback the `Sample` so it can borrow the payload and decode straight out of the network buffer, and switches `hiroz-py` to it. Everything else about the subscriber is unchanged: same encoding validation, same dispatch rules, same liveliness and graph registration. Split out of #250. It shared a call site with that PR's re-entrancy fix, which is proximity, not a reason to review them together. --- crates/hiroz-py/src/node.rs | 24 +++++++++++++++-- crates/hiroz/src/pubsub.rs | 54 +++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) diff --git a/crates/hiroz-py/src/node.rs b/crates/hiroz-py/src/node.rs index 6e43eb4dc..e186c6901 100644 --- a/crates/hiroz-py/src/node.rs +++ b/crates/hiroz-py/src/node.rs @@ -16,6 +16,7 @@ use hiroz::node::ZNode; use pyo3::prelude::*; use std::any::Any; use std::sync::Arc; +use zenoh_buffers::buffer::SplitBuffer; /// Try to extract type info from a message class. /// @@ -233,9 +234,25 @@ impl PyZNode { // matching rmw_zenoh_cpp's NodeData::subs_ pattern. The caller does not // need to assign the returned PyZSubscriber to keep the subscription active. let type_name = msg_type_str.clone(); + // Sample-level callback, not `build_with_callback`. The typed form + // would route through `RawBytesCdrSerdes::deserialize`, whose + // `Output` is an owned `RawBytesMessage` and so must `to_vec()` the + // whole payload before this closure runs — a full copy per message, + // scaling with payload size, immediately discarded once msgspec has + // decoded it. Taking the `Sample` lets the decode read straight out + // of the network buffer, and matches what the polling `recv()` path + // in `pubsub.rs` already does. let zsub = sub_builder - .build_with_callback(move |raw_msg: RawBytesMessage| { - let payload = raw_msg.0; + .build_with_sample_callback(move |sample| { + // Same zero-copy setup as `PyZSubscriber::recv`: the ZBuf is + // cheap Arc clones, and publishing it as the deserializer's + // source lets `bytes`-typed fields become sub-ZSlices of the + // received buffer instead of copies. + let payload_zbuf: zenoh_buffers::ZBuf = sample.payload().clone().into(); + hiroz_cdr::ZBUF_DESER_SOURCE.with(|cell| { + *cell.borrow_mut() = Some(payload_zbuf.clone()); + }); + let payload = payload_zbuf.contiguous(); Python::with_gil(|py| { match hiroz_msgs::deserialize_from_cdr(&type_name, py, &payload) { Ok(obj) => { @@ -248,6 +265,9 @@ impl PyZNode { } } }); + hiroz_cdr::ZBUF_DESER_SOURCE.with(|cell| { + *cell.borrow_mut() = None; + }); }) .map_err(|e| e.into_pyerr())?; diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 5b637d031..076d9660b 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -867,6 +867,60 @@ where }) } + /// Build a callback subscriber that receives the whole [`Sample`], undecoded. + /// + /// [`Self::build_with_callback`] must hand the callback an owned `S::Output`, + /// and [`ZDeserializer::Output`] carries no lifetime — so a serdes that only + /// forwards bytes (a language binding's identity codec, say) has no way to + /// express "borrow the payload", and must copy the entire message before the + /// callback has even seen it. That copy scales with payload size and is pure + /// waste when the consumer immediately re-reads the bytes into its own + /// representation. + /// + /// This entry point steps around it: the callback gets the `Sample`, so it + /// can borrow the payload (`sample.payload().to_bytes()` is a `Cow` that + /// borrows whenever the `ZBuf` is contiguous, which the receive path makes it) + /// and decode straight out of the network buffer. It can also reach the + /// sample's attachment, encoding and timestamp, which the decoded form drops. + /// + /// Everything else is identical to `build_with_callback` — same encoding + /// validation, same [`CallbackDispatcher`] handling, same liveliness and + /// graph registration. The callback is user code and is dispatched by exactly + /// the same rules. + /// + /// # Ownership + /// + /// As with `build_with_callback`, the returned [`ZSub`] must be kept alive for + /// the subscription to stay active. + pub fn build_with_sample_callback(self, callback: F) -> Result> + where + F: Fn(Sample) + Send + Sync + 'static, + S: ZDeserializer, + { + let expected_encoding = self.expected_encoding.clone(); + let callback = Arc::new(move |sample: Sample| { + if let Some(ref expected) = expected_encoding { + let encoding_str = sample.encoding().to_string(); + if let Some(received) = + crate::encoding::Encoding::from_zenoh_encoding(&encoding_str) + { + if &received != expected { + tracing::warn!( + "Encoding mismatch: expected {:?}, received {:?}", + expected, + received + ); + } + } else { + tracing::debug!("Unknown encoding format: {}", encoding_str); + } + } + callback(sample); + }); + + self.build_internal(DataHandler::Callback(callback), None) + } + /// Build a subscriber with a callback that processes deserialized messages directly. /// /// This method creates a subscriber that invokes the provided callback for each From c5ae0f5c2b97a9e45e91f23353549bd6363fe156 Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Wed, 29 Jul 2026 18:55:14 +0800 Subject: [PATCH 2/3] docs(pubsub): drop a link to a type this branch does not define The doc block moved here when this was split out of #250 referenced `CallbackDispatcher`, which #250 introduces and main does not have, so rustdoc could not resolve it and check-rustdoc-links failed. The sentence was also meaningless here for the same reason. --- crates/hiroz/src/pubsub.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 076d9660b..2a9c0e572 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -884,9 +884,8 @@ where /// sample's attachment, encoding and timestamp, which the decoded form drops. /// /// Everything else is identical to `build_with_callback` — same encoding - /// validation, same [`CallbackDispatcher`] handling, same liveliness and - /// graph registration. The callback is user code and is dispatched by exactly - /// the same rules. + /// validation, same liveliness and graph registration, same dispatch. The + /// callback is user code and is handled by exactly the same rules. /// /// # Ownership /// From 034b62d06db64d2b2a1d82f38b252eb33187657e Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Wed, 29 Jul 2026 20:45:03 +0800 Subject: [PATCH 3/3] docs(pubsub): condense the sample-callback rustdoc Half this PR was comment. The mechanism needs stating once -- owned Output, no lifetime, forced copy -- not restating three ways. --- crates/hiroz/src/pubsub.rs | 29 ++++++++--------------------- 1 file changed, 8 insertions(+), 21 deletions(-) diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 2a9c0e572..76d891da2 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -869,28 +869,15 @@ where /// Build a callback subscriber that receives the whole [`Sample`], undecoded. /// - /// [`Self::build_with_callback`] must hand the callback an owned `S::Output`, - /// and [`ZDeserializer::Output`] carries no lifetime — so a serdes that only - /// forwards bytes (a language binding's identity codec, say) has no way to - /// express "borrow the payload", and must copy the entire message before the - /// callback has even seen it. That copy scales with payload size and is pure - /// waste when the consumer immediately re-reads the bytes into its own - /// representation. + /// [`Self::build_with_callback`] hands the callback an owned `S::Output`, and + /// [`ZDeserializer::Output`] carries no lifetime — so a serdes that only + /// forwards bytes must copy the whole message before the callback sees it. + /// Taking the `Sample` instead lets the callback borrow the payload and + /// decode out of the network buffer, and reaches the attachment, encoding and + /// timestamp that the decoded form drops. /// - /// This entry point steps around it: the callback gets the `Sample`, so it - /// can borrow the payload (`sample.payload().to_bytes()` is a `Cow` that - /// borrows whenever the `ZBuf` is contiguous, which the receive path makes it) - /// and decode straight out of the network buffer. It can also reach the - /// sample's attachment, encoding and timestamp, which the decoded form drops. - /// - /// Everything else is identical to `build_with_callback` — same encoding - /// validation, same liveliness and graph registration, same dispatch. The - /// callback is user code and is handled by exactly the same rules. - /// - /// # Ownership - /// - /// As with `build_with_callback`, the returned [`ZSub`] must be kept alive for - /// the subscription to stay active. + /// Otherwise identical to `build_with_callback`, including that the returned + /// [`ZSub`] must be kept alive for the subscription to stay active. pub fn build_with_sample_callback(self, callback: F) -> Result> where F: Fn(Sample) + Send + Sync + 'static,