Default to Zenoh transport on macOS and document replay workflow - #2106
Default to Zenoh transport on macOS and document replay workflow#2106bogwi wants to merge 41 commits into
Conversation
Greptile SummaryThis PR integrates a Zenoh transport backend alongside LCM, defaults to Zenoh on macOS when
Confidence Score: 4/5Safe to merge for Linux users; macOS users forcing --transport=lcm may silently lose the pSHM color_image override if the blueprint module was pre-imported. The module-level dimos/robot/unitree/go2/blueprints/basic/unitree_go2_basic.py — module-level transport check should be deferred to blueprint-build time rather than evaluated once at import. Important Files Changed
Reviews (4): Last reviewed commit: "codecov: add zenoh" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| if "pubsubs" in fields_set and pubsubs is not None: | ||
| is_legacy_default = len(pubsubs) == 1 and isinstance(pubsubs[0], LCM) | ||
| if not is_legacy_default: | ||
| return pubsubs | ||
| return _default_pubsubs(getattr(config, "g", config)) |
There was a problem hiding this comment.
pubsubs=[LCM()]
_resolve_pubsubs cannot distinguish between the old default pubsubs=[LCM()] and a caller who explicitly sets pubsubs=[LCM()] to opt out of Zenoh. On macOS with Zenoh installed, any explicit pubsubs=[LCM()] will be silently replaced by [Zenoh(), LCM()] — the user's intent to stay on LCM-only is ignored with no warning. The only escape hatch is to also set transport="lcm" globally, but that changes module stream transport as well, not just the Rerun bridge's subscription backend. A narrower fix would be to respect the explicit override if the global transport is also lcm-forced, or to add a dedicated force_pubsubs field so callers can opt out cleanly.
| @property | ||
| def session(self) -> zenoh.Session: | ||
| """Get the Zenoh Session instance for this service's config.""" | ||
| key = self.config.session_key | ||
| if key not in _sessions: | ||
| raise RuntimeError("Zenoh session not initialized — call start() first") | ||
| return _sessions[key] |
There was a problem hiding this comment.
session property reads _sessions without holding _sessions_lock — there is a TOCTOU between the if key not in _sessions guard and the return _sessions[key] line. If close_all_zenoh_sessions() runs between those two statements on another thread, the dict will be empty and the return would raise a KeyError rather than the expected RuntimeError. Wrapping the property body in _sessions_lock (or using dict.get) eliminates the race.
| @property | |
| def session(self) -> zenoh.Session: | |
| """Get the Zenoh Session instance for this service's config.""" | |
| key = self.config.session_key | |
| if key not in _sessions: | |
| raise RuntimeError("Zenoh session not initialized — call start() first") | |
| return _sessions[key] | |
| @property | |
| def session(self) -> zenoh.Session: | |
| """Get the Zenoh Session instance for this service's config.""" | |
| key = self.config.session_key | |
| with _sessions_lock: | |
| session = _sessions.get(key) | |
| if session is None: | |
| raise RuntimeError("Zenoh session not initialized — call start() first") | |
| return session |
| class ZenohTransport(PubSubTransport[T]): | ||
| """Zenoh transport with LCM encoding for typed DimosMsg.""" | ||
|
|
||
| _started: bool = False | ||
|
|
||
| def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no-untyped-def] | ||
| super().__init__(LCMTopic(topic, type)) | ||
| self.zenoh = Zenoh(**kwargs) | ||
| self._start_lock = threading.RLock() |
There was a problem hiding this comment.
_started declared as a class-level attribute — _started: bool = False defines a class variable. The first self._started = True assignment silently creates a per-instance attribute that shadows it, which works in CPython but is a code smell: mypy treats the annotation as a class-level field, and pickling via __reduce__ on a freshly reconstructed object relies on the class-level default (which is correct here, but fragile). Initialising _started inside __init__ makes the per-instance ownership explicit and avoids the ambiguity. The same applies to pZenohTransport.
| class ZenohTransport(PubSubTransport[T]): | |
| """Zenoh transport with LCM encoding for typed DimosMsg.""" | |
| _started: bool = False | |
| def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no-untyped-def] | |
| super().__init__(LCMTopic(topic, type)) | |
| self.zenoh = Zenoh(**kwargs) | |
| self._start_lock = threading.RLock() | |
| class ZenohTransport(PubSubTransport[T]): | |
| """Zenoh transport with LCM encoding for typed DimosMsg.""" | |
| def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no-untyped-def] | |
| super().__init__(LCMTopic(topic, type)) | |
| self.zenoh = Zenoh(**kwargs) | |
| self._start_lock = threading.RLock() | |
| self._started: bool = False |
Prepare the codebase for Zenoh integration without changing behavior. All existing tests pass (1401 passed, 3 xfailed for Phase 2 stubs). - Add `transport` field to GlobalConfig (default: "lcm") - Add ZENOH_AVAILABLE guard in transport.py - Branch _get_transport_for() on global_config.transport - Gate LCM configurators to only run when transport is "lcm" - Add ZenohTransport/pZenohTransport behind ZENOH_AVAILABLE guard - Add zenohpubsub.py stub (raises NotImplementedError) - Add `zenoh` optional dependency group in pyproject.toml - Add test_zenoh_transport.py covering all new conditional branches
TDD: tests written first, then implementation. Follows DDSService pattern — module-level session dict with lock. - ZenohConfig with mode/connect/listen fields and session_key - ZenohService.start() opens session if not exists for config - ZenohService.stop() does NOT close shared session - session property raises RuntimeError if not started - Two services with same config share one session (8 tests pass)
TDD: tests written first, then implementation. - ZenohPubSubBase(ZenohService, AllPubSub[Topic, bytes]) - Publisher caching per key expression (avoids re-declaring) - Subscriber tracking for cleanup on stop() - Idempotent unsubscribe (guards against Zenoh ZError) - subscribe_all() via dimos/** wildcard - Zenoh and PickleZenoh composed classes (encoder mixins) - 7 unit tests pass
Both encoder-composed variants pass all spec conformance tests: - test_store, test_multiple_subscribers, test_unsubscribe - test_multiple_messages, test_async_iterator - 25 total tests pass (10 new Zenoh tests)
Remove xfail markers — Phase 2 stubs are now real implementations. Add transport wrapper integration tests for broadcast/subscribe. - ZenohTransport wraps Zenoh (LCM-encoded) with DDSTransport pattern - pZenohTransport wraps PickleZenoh with Topic wrapping for pubsub layer - Auto-start on first broadcast, stop/restart lifecycle - 16 tests pass (4 new wrapper tests)
Zenoh appears alongside LCM, SHM in benchmark heatmaps. Results: competitive with LCM for localhost — 82-149k msgs/sec for small messages, 0% message loss, <1ms latency.
Transport-level errors (session closed, invalid key expression) are logged but not raised. Delivery guarantees are handled by Zenoh's reliability protocol, not by exception propagation.
- Fix #3: unsubscribe() now only calls undeclare() if it successfully removed the subscriber from the list. If stop() already cleared the list, unsubscribe() returns without double-undeclaring. - Fix #5: on_sample callback wraps payload.to_bytes() in try/except to prevent malformed payloads from crashing Zenoh's internal thread.
Check membership before removing instead of catching ValueError. Reads more clearly and avoids using exceptions for control flow.
Two issues prevented the Rerun bridge from showing data over Zenoh: 1. The bridge hardcoded LCM() as its pubsub. Now resolves lazily at start() using self.config.g.transport from the worker's GlobalConfig. 2. Zenoh key expressions cannot contain '#' (forbidden character). Type info is now embedded as a '/' segment in the key expression (e.g., dimos/pointcloud/sensor_msgs.PointCloud2). _key_expr_to_topic reconstructs the Topic with lcm_type for subscribe_all decoding. Also fixes entity path mapping to strip the dimos/ prefix so Zenoh entity paths match LCM paths in the Rerun viewer.
- typed_out/untyped_out → typed_data/untyped_data - Use TypedMsg instead of Image for blueprint integration tests - Image still used in transport wrapper test (real LCM round-trip)
Replace raw time.sleep() calls with named helpers that document intent. wait_for_subscribers() explains Zenoh has no "subscriber ready" signal.
Replace manual if-both-received check with threading.Barrier(2). The previous approach could miss the event if both callbacks ran concurrently and checked the other's list before it was populated.
8 new tests covering: - Typed/untyped topic → key expression conversion - Key expression → topic with known/unknown/missing type - Default lcm_type fallback - Round-trip typed and untyped Also documents known limitation: if a topic's base path ends with a segment matching a registered DimosMsg type name, _key_expr_to_topic will incorrectly split it. In practice this doesn't happen because stream names (cmd_vel, lidar) don't match type names.
Existing blueprints pass pubsubs=[LCM()] to RerunBridgeModule. Removing the field caused a Pydantic ValidationError (extra_forbidden). Keep the field but document that it's ignored — start() resolves the pubsub backend from global_config.transport instead.
TF (transform frames) is hardcoded to LCM in the Module base class. When transport=zenoh, module streams use Zenoh but TF stays on LCM. The bridge now listens on both so the robot pose updates in the viewer.
Zenoh tests used time.sleep() to wait for subscriber propagation, which is either too slow or too flaky in CI. Replace with _retry_until() that re-publishes in a tight loop until the subscriber's Event fires.
Calls zenoh.init_log_from_env_or("warn") at module load so that
RUST_LOG=debug surfaces Zenoh's Rust-side transport logs (including
SHM negotiation). Defaults to warn to avoid noise.
uv sync --extra zenoh would resolve dimos[dev] from PyPI instead of the local project, uninstalling other dependencies. The zenoh extra only needs eclipse-zenoh — base deps are already installed.
… missing Align with module_coordinator._get_transport_for: raise RuntimeError instead of silently falling back to LCM when transport is zenoh and eclipse-zenoh is not installed.
…v pip instead of uv sync --extra zenoh
Agent and MCP conftest subscribe on LCM while the coordinator uses global_config.transport, which defaults to Zenoh on Darwin when Zenoh is installed. Set transport to lcm so spies and modules share the same backend. The module reloading test runs a separate Python process; set transport to lcm in the REPL before ModuleCoordinator.build for the same reason.
f7959b2 to
056ddcd
Compare
| **Two ways to override for one run or for your shell:** | ||
|
|
||
| 1. **CLI:** `dimos --transport=zenoh ...` or `dimos --transport=lcm ...` (see [CLI](/docs/usage/cli.md) for precedence with `.env` and blueprints). | ||
| 2. **Environment:** `DIMOS_TRANSPORT=zenoh` or `DIMOS_TRANSPORT=lcm`. |
There was a problem hiding this comment.
DIMOS_TRANSPORT env var is not recognized — should be TRANSPORT
GlobalConfig is a pydantic_settings.BaseSettings with no env_prefix configured in its model_config. Without env_prefix = "DIMOS_", pydantic-settings resolves each field from the uppercased field name directly — so transport reads from TRANSPORT, not DIMOS_TRANSPORT. A user who follows the docs and sets export DIMOS_TRANSPORT=zenoh will silently get no effect; their stack will still start with the platform default transport. The same issue applies to the DIMOS_TRANSPORT reference in docs/installation/osx.md. The correct env var to document is TRANSPORT=zenoh.
056ddcd to
91c3af2
Compare
| return autoconnect() | ||
|
|
||
|
|
||
| _transports_base = _platform_transports() |
There was a problem hiding this comment.
_transports_base = _platform_transports() is evaluated exactly once — when this module is first imported — and its global_config.transport read is frozen at that point. If the module is imported before the CLI (or a test fixture) calls global_config.update(transport="lcm"), the pSHM override for color_image is silently skipped and never re-applied for the lifetime of the process.
The original line only called platform.system() (constant), so it was safe to evaluate at import time. The new code adds a dependency on global_config.transport, which can change after import. Any pre-import of this module — e.g., via dimos.robot.unitree.go2.blueprints.smart.unitree_go2, which is itself discoverable by pytest collection — will freeze the pSHM decision before the fixture's global_config.update(transport="lcm") runs.
|
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. |
|
Closed this old PR due to inactivity. Can be reopened later. |
3 participants
Supersedes #1906 so self-hosted CI can run. Prior review discussion is preserved there.
Supersedes #1787.
This PR carries forward the Zenoh transport integration from #1787 and wraps it into a merge-ready branch that fixes the remaining macOS Big Office replay gap.
Validation
Typical replay on macOS when Zenoh is installed (default is already Zenoh, so no transport flag is required):
The same workload on Linux (default remains
lcmuntil you opt in):Notes