diff --git a/crates/hiroz-py/python/hiroz_py/__init__.pyi b/crates/hiroz-py/python/hiroz_py/__init__.pyi index a355850e5..d27c233e3 100644 --- a/crates/hiroz-py/python/hiroz_py/__init__.pyi +++ b/crates/hiroz-py/python/hiroz_py/__init__.pyi @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any, Final +from typing import Any, Final, Literal # Re-export message types from hiroz_msgs_py.types from hiroz_msgs_py import types as types @@ -183,6 +183,7 @@ class ZNode: msg_type: Any, qos: QosProfile | dict[str, object] | None = None, callback: Any | None = None, + buffer_group: Literal["auto", "cpu", "off"] = "auto", ) -> ZSubscriber: ... def create_client(self, service: str, srv_type: Any) -> ZClient: ... def create_server(self, service: str, srv_type: Any) -> ZServer: ... diff --git a/crates/hiroz-py/src/node.rs b/crates/hiroz-py/src/node.rs index 6e43eb4dc..bc63dcefb 100644 --- a/crates/hiroz-py/src/node.rs +++ b/crates/hiroz-py/src/node.rs @@ -17,6 +17,17 @@ use pyo3::prelude::*; use std::any::Any; use std::sync::Arc; +fn parse_buffer_group(group: &str) -> PyResult { + match group { + "auto" => Ok(hiroz::BufferGroup::Auto), + "cpu" => Ok(hiroz::BufferGroup::Cpu), + "off" => Ok(hiroz::BufferGroup::Off), + other => Err(pyo3::exceptions::PyValueError::new_err(format!( + "buffer_group must be 'auto', 'cpu' or 'off', got {other:?}" + ))), + } +} + /// Try to extract type info from a message class. /// /// Returns `None` if `__msgtype__` or `__hash__` are absent or not strings — @@ -209,7 +220,13 @@ impl PyZNode { /// Create a subscriber for a given topic and message type. /// /// Works with any registered message type — no factory limitations. - #[pyo3(signature = (topic, msg_type, qos=None, callback=None))] + /// + /// `buffer_group` is `'auto'`, `'cpu'` or `'off'` and decides whether the subscriber joins + /// the CPU buffer group of rmw_zenoh. The bindings have no message schema to inspect, so + /// `'auto'` and `'cpu'` do the same: the subscriber joins on every topic, unless it uses + /// transient-local QoS or a key format other than rmw_zenoh's. `'off'` never joins. Any other + /// string raises `ValueError`. + #[pyo3(signature = (topic, msg_type, qos=None, callback=None, buffer_group="auto"))] fn create_subscriber( &mut self, _py: Python, @@ -217,15 +234,18 @@ impl PyZNode { msg_type: &Bound<'_, PyAny>, qos: Option<&Bound<'_, PyAny>>, callback: Option, + buffer_group: &str, ) -> PyResult { let (msg_type_str, type_info) = extract_type_info_from_class(msg_type)?; let qos_profile = extract_qos(qos)?; + let group = parse_buffer_group(buffer_group)?; let sub_builder = self .inner .create_sub_impl::(&topic, Some(type_info)) .with_serdes::() - .with_qos(qos_profile); + .with_qos(qos_profile) + .with_buffer_group(group); if let Some(py_callback) = callback { // Callback-based subscription: no queue, callback fires on each message. diff --git a/crates/hiroz-py/tests/test_buffer_group.py b/crates/hiroz-py/tests/test_buffer_group.py new file mode 100644 index 000000000..6c1f6e15f --- /dev/null +++ b/crates/hiroz-py/tests/test_buffer_group.py @@ -0,0 +1,15 @@ +"""Tests for the buffer_group argument of create_subscriber.""" + +import pytest +from hiroz_py import std_msgs + + +@pytest.mark.parametrize("group", ["AUTO", "Off", "on", "", "cpu "]) +def test_create_subscriber_rejects_unknown_group(node, group): + with pytest.raises(ValueError, match="buffer_group"): + node.create_subscriber("/group_bad", std_msgs.String, buffer_group=group) + + +def test_create_subscriber_rejects_a_non_string_group(node): + with pytest.raises(TypeError, match="buffer_group"): + node.create_subscriber("/group_bad", std_msgs.String, buffer_group=1) diff --git a/crates/hiroz-py/tests/test_buffer_group_policy.py b/crates/hiroz-py/tests/test_buffer_group_policy.py new file mode 100644 index 000000000..7b120403e --- /dev/null +++ b/crates/hiroz-py/tests/test_buffer_group_policy.py @@ -0,0 +1,171 @@ +"""Tests that the buffer_group argument of create_subscriber sets the policy that takes effect. + +A plain Zenoh session plays the publisher. It reads the liveliness token of the subscriber, and it +writes to the group key `/_buf_cpu`. A subscriber that joined announces `backends:cpu:` in +its token and receives the group key. A subscriber that did not join keeps its token, without the +`backends:cpu:` part, and receives the base key only. + +The tests check the token and the group-key delivery of each value, in the queue form and in the +callback form. + +Each case has live controls. A second subscriber created with `buffer_group='cpu'` shows that the +token query and the group key work in the same run. A put on the base key shows that the subscriber +under test is alive. The tests cannot tell `auto` from `cpu`, because the two behave the same in +Python. +""" + +import socket +import struct +import time + +import pytest +import zenoh +from hiroz_py import ZContextBuilder, std_msgs + +DOMAIN = 0 +LISTEN_WAIT = 5.0 +QUIET = 1.5 +POLL = 0.1 + +# The default of the argument: the case passes no buffer_group at all. +OMITTED = None + +POLICIES = [ + pytest.param("off", False, id="off"), + pytest.param("cpu", True, id="cpu"), + pytest.param("auto", True, id="auto"), + pytest.param(OMITTED, True, id="omitted"), +] +FORMS = ["queue", "callback"] + + +def _free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture(scope="module") +def rig(): + """A plain Zenoh session that listens, and a hiroz node that connects to it.""" + endpoint = f"tcp/127.0.0.1:{_free_port()}" + config = zenoh.Config() + config.insert_json5("listen/endpoints", f'["{endpoint}"]') + config.insert_json5("scouting/multicast/enabled", "false") + session = zenoh.open(config) + context = ( + ZContextBuilder() + .with_domain_id(DOMAIN) + .with_connect_endpoints([endpoint]) + .disable_multicast_scouting() + .build() + ) + node = context.create_node("group_policy_node").build() + yield session, node + session.close() + + +def _cdr_string(text): + raw = text.encode() + b"\x00" + return b"\x00\x01\x00\x00" + struct.pack(" 1 or time.monotonic() >= deadline: + raise AssertionError(f"{len(hits)} tokens for {name}: {hits}") + time.sleep(POLL) + + +def _base_key(token, name): + """The key a publisher writes `name` to, read from the segments of the token.""" + parts = token.split("/") + i = parts.index(f"%{name}") + return f"{DOMAIN}/{name}/{parts[i + 1]}/{parts[i + 2]}" + + +class _Probe: + """A subscriber of either form, and what it has received.""" + + def __init__(self, node, topic, form, group): + self.form = form + self.seen = [] + kwargs = {} if group is OMITTED else {"buffer_group": group} + if form == "callback": + kwargs["callback"] = self.seen.append + self.sub = node.create_subscriber(topic, std_msgs.String, **kwargs) + + def data(self): + if self.form == "queue": + while (msg := self.sub.try_recv()) is not None: + self.seen.append(msg) + return [m.data for m in self.seen] + + +def _put_until_seen(session, key, probe, tag, limit=LISTEN_WAIT): + """Put `tag` on `key` until `probe` has received it or `limit` seconds pass.""" + end = time.monotonic() + limit + while time.monotonic() < end: + session.put(key, _cdr_string(tag)) + time.sleep(0.05) + if tag in probe.data(): + return True + return False + + +@pytest.mark.parametrize("form", FORMS) +@pytest.mark.parametrize("group,joins", POLICIES) +def test_policy_takes_effect(rig, form, group, joins): + session, node = rig + label = "omitted" if group is OMITTED else group + name = f"policy_{form}_{label}" + twin_name = f"{name}_twin" + + probe = _Probe(node, f"/{name}", form, group) + twin = _Probe(node, f"/{twin_name}", form, "cpu") + + twin_token = _token_for(session, twin_name) + assert twin_token.endswith("/backends:cpu:"), ( + f"control: the cpu twin did not announce backends:cpu: {twin_token}" + ) + token = _token_for(session, name) + assert token.endswith("/backends:cpu:") == joins, ( + f"buffer_group={label!r} ({form} form): expected joined={joins}, token is {token}" + ) + + base = _base_key(token, name) + group_key = f"{base}/_buf_cpu" + twin_group_key = f"{_base_key(twin_token, twin_name)}/_buf_cpu" + + assert _put_until_seen(session, base, probe, "base"), ( + f"control: nothing reached the {form} subscriber from {base}" + ) + assert _put_until_seen(session, twin_group_key, twin, "group"), ( + f"control: nothing reached the cpu twin from {twin_group_key}" + ) + + if joins: + assert _put_until_seen(session, group_key, probe, "group"), ( + f"buffer_group={label!r} ({form} form) joined but the group key does not deliver" + ) + else: + assert not _put_until_seen(session, group_key, probe, "group", limit=QUIET), ( + f"buffer_group={label!r} ({form} form) did not join but received the group key" + ) diff --git a/crates/hiroz-py/tests/test_pep561.py b/crates/hiroz-py/tests/test_pep561.py index 416d7bdc1..e132872d9 100644 --- a/crates/hiroz-py/tests/test_pep561.py +++ b/crates/hiroz-py/tests/test_pep561.py @@ -110,3 +110,25 @@ def test_qos_constants_are_qos_profile_instances(): def test_native_submodule_accessible(): """hiroz_py._native must be importable as a submodule.""" from hiroz_py import _native # noqa: F401 + + +def test_stub_declares_buffer_group_default_auto(): + """ZNode.create_subscriber in the stub must take buffer_group, defaulting to "auto".""" + spec = importlib.util.find_spec("hiroz_py") + assert spec is not None + stub = Path(spec.submodule_search_locations[0]) / "__init__.pyi" + tree = ast.parse(stub.read_text()) + znode = next( + n for n in ast.walk(tree) if isinstance(n, ast.ClassDef) and n.name == "ZNode" + ) + fn = next( + n + for n in znode.body + if isinstance(n, ast.FunctionDef) and n.name == "create_subscriber" + ) + names = [a.arg for a in fn.args.args] + assert "buffer_group" in names, "the stub lacks buffer_group" + default = fn.args.defaults[ + names.index("buffer_group") - (len(names) - len(fn.args.defaults)) + ] + assert isinstance(default, ast.Constant) and default.value == "auto" diff --git a/crates/hiroz-tests/tests/buffer_cpu_group.rs b/crates/hiroz-tests/tests/buffer_cpu_group.rs index 178232328..3237145a2 100644 --- a/crates/hiroz-tests/tests/buffer_cpu_group.rs +++ b/crates/hiroz-tests/tests/buffer_cpu_group.rs @@ -676,10 +676,10 @@ fn cache_join_policy() { check_rows(&raw, "cache_join_policy", &rows); } -/// The join policy of an action client. With no schema the feedback and status subscribers join on -/// topics that carry no buffer, as a typed subscriber with no schema does. A transient-local -/// feedback subscriber stays out, and the feedback subscriber of the default client joins as the -/// control. +/// The join policy of an action client. The status subscriber stays out, because its message has +/// no buffer field. A transient-local feedback subscriber stays out, as for every subscriber. The +/// feedback subscriber of the default client joins, and it is the control that shows the group +/// token is visible in this run. #[test] fn action_client_join_policy() { let router = TestRouter::new(); @@ -701,7 +701,7 @@ fn action_client_join_policy() { .unwrap(); for (topic, expected) in [ ("ac_default%_action%feedback", true), - ("ac_default%_action%status", true), + ("ac_default%_action%status", false), ("ac_transient%_action%feedback", false), ] { assert_eq!( @@ -1100,7 +1100,7 @@ fn router_allowing_cases(cases: &[AclCase], group: bool) -> TestRouter { /// subscriber declares `` and `/_buf_cpu`, so a default-deny router that /// allows `` and not the group key starves the group key, and `build` still returns /// `Ok`. The base key still flows. This holds for a buffer type and for a message with no -/// definition. A real publisher writes nothing on the group key of a topic that carries no buffer, +/// schema. A real publisher writes nothing on the group key of a topic that carries no buffer, /// so the cases for that message put on the group key only to show that the rule denies it. /// /// Controls in the same test: an `Off` subscriber receives under the allow-only rule, so the puts diff --git a/crates/hiroz/src/action/client.rs b/crates/hiroz/src/action/client.rs index e7f451bf6..65b207813 100644 --- a/crates/hiroz/src/action/client.rs +++ b/crates/hiroz/src/action/client.rs @@ -233,12 +233,13 @@ impl<'a, A: ZAction> Builder for ZActionClientBuilder<'a, A> { })?; tracing::debug!("Feedback subscriber created successfully"); - // Create status subscriber with callback for direct message routing // Use the action's status_type_info for proper ROS 2 interop let status_type_info = Some(A::status_type_info()); + // The status message has no buffer field, so the subscriber stays out of the buffer group. let mut status_sub_builder = self .node - .create_sub_impl::(&status_topic_name, status_type_info); + .create_sub_impl::(&status_topic_name, status_type_info) + .with_buffer_group(crate::BufferGroup::Off); if let Some(qos) = self.status_topic_qos { status_sub_builder.entity.qos = qos.to_protocol_qos(); } diff --git a/crates/hiroz/src/buffer_group.rs b/crates/hiroz/src/buffer_group.rs index 7a5aceccc..b2499f197 100644 --- a/crates/hiroz/src/buffer_group.rs +++ b/crates/hiroz/src/buffer_group.rs @@ -50,8 +50,8 @@ pub enum BufferGroup { /// Whether a subscriber joins. `has_buffer_fields` is `None` when hiroz has no message schema. /// -/// Only the `rmw_zenoh` key format has the group. A transient-local subscriber stays out: the -/// released `rmw_zenoh_cpp` serves its history on the base key only. +/// Only the `rmw_zenoh` key format has the group. A transient-local subscriber stays out under +/// every setting, `Cpu` included: its history comes from the base key, as in `rmw_zenoh_cpp`. pub(crate) fn joins( group: BufferGroup, has_buffer_fields: Option, diff --git a/docs/user-guide/interop.md b/docs/user-guide/interop.md index 0747b4817..493697bbe 100644 --- a/docs/user-guide/interop.md +++ b/docs/user-guide/interop.md @@ -119,14 +119,16 @@ Set the choice with `with_buffer_group` on a subscriber builder or a cache build | `BufferGroup::Off` | never | - The typed code generator maps ROS `byte`, `uint8` and `char` to one type. A typed subscriber therefore also joins for an unbounded `byte[]`. This costs one extra join and loses no data. -- A subscriber with no message schema joins on every topic, not only on buffer topics. The feedback and status subscribers of an action client and the Python bindings have no message schema. -- The action client and the Python bindings have no `Off` switch yet. +- A subscriber with no message schema joins on every topic, not only on buffer topics. The feedback subscriber of an action client and the Python bindings have no message schema. +- The Python `create_subscriber(..., buffer_group=...)` takes `'auto'`, `'cpu'` or `'off'`. `'auto'` and `'cpu'` do the same in Python. Any other string raises `ValueError`. +- The status subscriber of an action client always stays out, because its message has no buffer. The feedback subscriber has no `Off` switch yet. - A raw subscriber has no message schema, so `Auto` joins it. `create_raw_subscriber` and `create_raw_subscriber_with_qos` join, and so do the C and Go bindings that call them. `create_raw_subscriber_with_group` takes `Auto`, `Cpu` or `Off`. It exists only with the `ffi` feature, and the Go bindings have no entry point for it yet. The feedback subscriber of a raw action client reads nothing and stays out. +- A subscriber that stays out subscribes to the base key only. A typed or Python subscriber that stays out keeps its liveliness token, without the `backends:cpu:` part. A raw subscriber that stays out declares no token. - `RawSubscriber` has no public field. In release 0.2.0 the `inner` field is public, with the `ffi` feature. A caller of `raw.inner.undeclare().wait()` must call `raw.undeclare()` now. That call stops the token and both subscriptions. Dropping the wrapper stops the same handles. -- A transient-local subscriber never joins. A subscriber on a key format other than `rmw_zenoh`'s never joins. This holds under every setting. +- A transient-local subscriber never joins, and `Cpu` does not change that, because it reads its cached history from the base key. `rmw_zenoh_cpp` does the same: [0.10.6](https://github.com/ros2/rmw_zenoh/blob/4e17ccf/rmw_zenoh_cpp/src/detail/rmw_subscription_data.cpp#L99-L103) and [rolling at 678931f](https://github.com/ros2/rmw_zenoh/blob/678931f/rmw_zenoh_cpp/src/detail/rmw_subscription_data.cpp#L103-L107) disable buffer endpoints for transient-local subscribers, and rolling does so for [publishers](https://github.com/ros2/rmw_zenoh/blob/678931f/rmw_zenoh_cpp/src/detail/rmw_publisher_data.cpp#L94-L98). This limit concerns buffer transport and does not disable transient-local QoS. A subscriber on a key format other than `rmw_zenoh`'s never joins, under every setting. - `rmw-zenoh-rs` subscriptions use `Off`. -A transient-local subscriber reads its history from ``. The released 0.10.6 package gave this result in 75 cells with Image topics: five kinds of late joiner, three publish-time situations, n=5 per cell. The history arrived on `` in every cell and never on `/_buf_cpu`. An unreleased source revision serves it on the group key. No test covers history after [ros2/rmw_zenoh#1078](https://github.com/ros2/rmw_zenoh/pull/1078). +A transient-local subscriber reads its history from ``. The released 0.10.6 package gave this result in 75 cells with Image topics: five kinds of late joiner, three publish-time situations, n=5 per cell. The history arrived on `` in every cell and never on `/_buf_cpu`. No test covers history after [ros2/rmw_zenoh#1078](https://github.com/ros2/rmw_zenoh/pull/1078). If you use router access control, allow `` and `/_buf_cpu`. A joined subscriber declares only these two keys, so a rule on `/**` is not needed. A rule that allows `` and not `/_buf_cpu` stops a joined subscriber from receiving buffer messages. `build` returns `Ok` and no error appears. The base key still delivers. A subscriber receives a message that carries no buffer through the base key, so a rule on the group key does not affect that message. This holds for a subscriber with no message schema and for a subscriber set to `Cpu` on a topic that carries no buffer. `Off` stops a subscriber from joining, so `Off` needs no rule for the group key. A subscriber with no `backends:` field receives no buffer message from a buffer-aware publisher until a release carries [ros2/rmw_zenoh#1078](https://github.com/ros2/rmw_zenoh/pull/1078). `Off` therefore does not bring those messages back. hiroz logs each join at `debug` level and names both keys. diff --git a/scripts/test-python.nu b/scripts/test-python.nu index 7a21507af..0787c54fb 100755 --- a/scripts/test-python.nu +++ b/scripts/test-python.nu @@ -42,6 +42,10 @@ def setup-venv [] { run-cmd "cd crates/hiroz-py; source .venv/bin/activate && pip install -e ../hiroz-msgs/python/" --shell bash --distro (get-distro) print " Installed hiroz-msgs-py (message types)" + # Install the Zenoh Python client, which the buffer group tests use as a plain publisher + run-cmd "cd crates/hiroz-py; source .venv/bin/activate && pip install eclipse-zenoh==1.9.0" --shell bash --distro (get-distro) + print " Installed eclipse-zenoh (test publisher)" + # Install hiroz-py in editable mode using maturin run-cmd "cd crates/hiroz-py; source .venv/bin/activate && RUSTFLAGS='-D warnings' maturin develop" --shell bash --distro (get-distro) print " Installed hiroz-py (Rust bindings)"