From d2dc12c1ec3d792c84397b4c99b981231d2ddd65 Mon Sep 17 00:00:00 2001 From: bogwi Date: Mon, 27 Jul 2026 11:12:23 +0900 Subject: [PATCH 1/3] Shard LCM stream channels across dedicated multicast buses. --- dimos/core/transport.py | 10 ++-- dimos/e2e_tests/lcm_spy.py | 23 ++++++++- dimos/protocol/service/lcmservice.py | 61 +++++++++++++++++++++++ dimos/utils/cli/agentspy/demo_agentspy.py | 5 +- dimos/utils/cli/spy/core.py | 28 +++++++++-- dimos/visualization/rerun/bridge.py | 4 +- 6 files changed, 116 insertions(+), 15 deletions(-) diff --git a/dimos/core/transport.py b/dimos/core/transport.py index a66d3c7d6a..25b561433f 100644 --- a/dimos/core/transport.py +++ b/dimos/core/transport.py @@ -45,6 +45,7 @@ Topic as ZenohTopic, Zenoh, ) +from dimos.protocol.service.lcmservice import lcm_url_for_channel from dimos.utils.logging_config import setup_logger logger = setup_logger() @@ -108,15 +109,13 @@ class pLCMTransport(PubSubTransport[T]): def __init__(self, topic: str, **kwargs) -> None: # type: ignore[no-untyped-def] super().__init__(topic) + kwargs.setdefault("url", lcm_url_for_channel(topic)) self.lcm = PickleLCM(**kwargs) def __reduce__(self): # type: ignore[no-untyped-def] return (pLCMTransport, (self.topic,)) def broadcast(self, _: Out[T] | None, msg: T) -> None: - if not self._started: - self.start() - self.lcm.publish(self.topic, msg) def subscribe( @@ -141,6 +140,7 @@ class LCMTransport(PubSubTransport[T]): def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no-untyped-def] super().__init__(LCMTopic(topic, type)) if not hasattr(self, "lcm"): + kwargs.setdefault("url", lcm_url_for_channel(str(self.topic))) self.lcm = LCM(**kwargs) def start(self) -> None: @@ -155,9 +155,6 @@ def __reduce__(self): # type: ignore[no-untyped-def] return (LCMTransport, (self.topic.topic, self.topic.lcm_type)) def broadcast(self, _, msg) -> None: # type: ignore[no-untyped-def] - if not self._started: - self.start() - self.lcm.publish(self.topic, msg) def subscribe( @@ -174,6 +171,7 @@ def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no JpegLCM, ) # ~330ms: deferred to avoid pulling in Image/cv2/rerun + kwargs.setdefault("url", lcm_url_for_channel(str(LCMTopic(topic, type)))) self.lcm = JpegLCM(**kwargs) # type: ignore[assignment] super().__init__(topic, type) diff --git a/dimos/e2e_tests/lcm_spy.py b/dimos/e2e_tests/lcm_spy.py index cec34324f3..dfccd7ef5f 100644 --- a/dimos/e2e_tests/lcm_spy.py +++ b/dimos/e2e_tests/lcm_spy.py @@ -23,7 +23,11 @@ from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped from dimos.msgs.protocol import DimosMsg -from dimos.protocol.service.lcmservice import LCMService +from dimos.protocol.service.lcmservice import ( + LCMService, + lcm_bus_urls, + lcm_url_for_channel, +) from dimos.utils.testing.waiting import wait_until @@ -39,6 +43,11 @@ class LcmSpy(LCMService): def __init__(self, **kwargs: Any) -> None: super().__init__(**kwargs) self.l = lcm.LCM() + # Stream channels are sharded across dedicated buses; tap those too so + # the spy sees the whole system, not just the default control bus. + self._stream_services = [ + LCMService(url=url) for url in lcm_bus_urls() if url != self.config.url + ] self.messages = {} self._messages_lock = threading.Lock() self._saved_topics = set() @@ -50,9 +59,15 @@ def start(self) -> None: super().start() if self.l: self.l.subscribe(".*", self.msg) + for service in self._stream_services: + service.start() + if service.l: + service.l.subscribe(".*", self.msg) def stop(self) -> None: super().stop() + for service in self._stream_services: + service.stop() def msg(self, topic: str, data: bytes) -> None: with self._saved_topics_lock: @@ -67,6 +82,12 @@ def msg(self, topic: str, data: bytes) -> None: listener(data) def publish(self, topic: str, msg: Any) -> None: + # Publish on the bus the channel's exact subscribers joined. + url = lcm_url_for_channel(topic) + for service in self._stream_services: + if service.config.url == url and service.l is not None: + service.l.publish(topic, msg.lcm_encode()) + return self.l.publish(topic, msg.lcm_encode()) def save_topic(self, topic: str) -> None: diff --git a/dimos/protocol/service/lcmservice.py b/dimos/protocol/service/lcmservice.py index f63a1e3b09..ab3c570b79 100644 --- a/dimos/protocol/service/lcmservice.py +++ b/dimos/protocol/service/lcmservice.py @@ -15,11 +15,14 @@ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor +from ipaddress import IPv4Address import os import platform import threading import traceback from typing import Any +from urllib.parse import urlsplit +import zlib import lcm as lcm_mod @@ -39,6 +42,64 @@ ) +def _multicast_bus_url(base_url: str, offset: int) -> str: + """Derive an adjacent multicast group with its own port. + + The dedicated port is what guarantees socket-level separation: BSD/macOS + demultiplexes multicast by bound port, so two buses sharing a port would + leak traffic into each other's sockets. Non-udpm and non-multicast URLs + (memq, unicast) are returned unchanged, collapsing the pool to one bus. + """ + parsed = urlsplit(base_url) + if parsed.scheme != "udpm" or parsed.hostname is None or parsed.port is None: + return base_url + try: + address = IPv4Address(parsed.hostname) + except ValueError: + return base_url + if not address.is_multicast: + return base_url + + octets = list(address.packed) + octets[-1] = ((octets[-1] - 1 + offset) % 254) + 1 + host = str(IPv4Address(bytes(octets))) + port_span = 65535 - 1024 + 1 + port = ((parsed.port - 1024 + offset) % port_span) + 1024 + return parsed._replace(netloc=f"{host}:{port}").geturl() + + +_STREAM_BUS_COUNT = max(0, int(os.getenv("LCM_STREAM_BUSES", "16"))) + +_STREAM_BUS_URLS = tuple( + dict.fromkeys(_multicast_bus_url(_DEFAULT_LCM_URL, i + 1) for i in range(_STREAM_BUS_COUNT)) +) + + +def lcm_url_for_channel(channel: str) -> str: + """Map a stream channel onto one bus of the stream pool. + + LCM filters channels in userspace, after the kernel has already copied + every datagram on a bus into every socket joined to it. Hashing each + channel onto its own (group, port) bus moves that filtering into the + kernel: a socket only receives the channels that share its bus, so + high-volume streams no longer fan out into every LCM handle in every + process. The hash must be identical across processes for publishers and + subscribers to meet, hence crc32 rather than the seeded builtin hash(). + + Set LCM_STREAM_BUSES=0 to disable sharding (single shared bus), or to a + different count to trade fewer sockets against more channel collisions. + """ + if not _STREAM_BUS_URLS: + return _DEFAULT_LCM_URL + index = zlib.crc32(channel.encode()) % len(_STREAM_BUS_URLS) + return _STREAM_BUS_URLS[index] + + +def lcm_bus_urls() -> tuple[str, ...]: + """Return each distinct DimOS LCM bus URL: the default bus plus the stream pool.""" + return tuple(dict.fromkeys((_DEFAULT_LCM_URL, *_STREAM_BUS_URLS))) + + def autoconf(check_only: bool = False) -> None: checks = lcm_configurators() if not checks: diff --git a/dimos/utils/cli/agentspy/demo_agentspy.py b/dimos/utils/cli/agentspy/demo_agentspy.py index 851229131b..95db2e4a41 100755 --- a/dimos/utils/cli/agentspy/demo_agentspy.py +++ b/dimos/utils/cli/agentspy/demo_agentspy.py @@ -26,14 +26,15 @@ import dimos.protocol.pubsub.impl.lcmpubsub as lcm from dimos.protocol.pubsub.impl.lcmpubsub import PickleLCM +from dimos.protocol.service.lcmservice import lcm_url_for_channel def test_publish_messages() -> None: """Publish test messages to verify agentspy is working.""" print("Starting agent message publisher demo...") - # Create transport - transport = PickleLCM() + # Create transport on the same sharded bus agentspy subscribes to + transport = PickleLCM(url=lcm_url_for_channel("/agent")) topic = lcm.Topic("/agent") print(f"Publishing to topic: {topic}") diff --git a/dimos/utils/cli/spy/core.py b/dimos/utils/cli/spy/core.py index 843cfb5032..5eabf6ed0d 100644 --- a/dimos/utils/cli/spy/core.py +++ b/dimos/utils/cli/spy/core.py @@ -193,17 +193,37 @@ def __init__(self, **lcm_kwargs: Any) -> None: # Inline import: an unavailable LCM backend must not break the other # spy sources (see default_sources). from dimos.protocol.pubsub.impl.lcmpubsub import LCMPubSubBase + from dimos.protocol.service.lcmservice import lcm_bus_urls - self._bus = LCMPubSubBase(**lcm_kwargs) + urls = (lcm_kwargs["url"],) if "url" in lcm_kwargs else lcm_bus_urls() + self._buses = [LCMPubSubBase(**{**lcm_kwargs, "url": url}) for url in urls] def start(self) -> None: - self._bus.start() + started: list[Any] = [] + try: + for bus in self._buses: + bus.start() + started.append(bus) + except BaseException: + for bus in reversed(started): + bus.stop() + raise def stop(self) -> None: - self._bus.stop() + for bus in self._buses: + bus.stop() def tap(self, callback: Callable[[str, int], None]) -> Callable[[], None]: - return self._bus.subscribe_all(lambda msg, topic: callback(str(topic), len(msg))) + unsubs = [ + bus.subscribe_all(lambda msg, topic: callback(str(topic), len(msg))) + for bus in self._buses + ] + + def unsubscribe() -> None: + for unsub in unsubs: + unsub() + + return unsubscribe def subscribe_decoded(self, topic: str, callback: Callable[[Any], None]) -> Callable[[], None]: """Opt-in per-topic decoded tap — OFF the spy hot path. Not implemented in v1.""" diff --git a/dimos/visualization/rerun/bridge.py b/dimos/visualization/rerun/bridge.py index f968216526..2253c40078 100644 --- a/dimos/visualization/rerun/bridge.py +++ b/dimos/visualization/rerun/bridge.py @@ -48,7 +48,7 @@ from dimos.protocol.pubsub.impl.zenohpubsub import Zenoh from dimos.protocol.pubsub.patterns import Glob, pattern_matches from dimos.protocol.pubsub.spec import SubscribeAllCapable -from dimos.protocol.service.lcmservice import autoconf +from dimos.protocol.service.lcmservice import autoconf, lcm_bus_urls from dimos.utils.generic import get_local_ips from dimos.utils.logging_config import setup_logger from dimos.visualization.rerun.constants import ( @@ -175,7 +175,7 @@ def _default_pubsubs(config: Any = None) -> list[SubscribeAllCapable[Any, Any]]: transport = getattr(config, "transport", None) or global_config.transport if transport == "zenoh": return [Zenoh()] - return [LCM()] + return [LCM(url=url) for url in lcm_bus_urls()] def _resolve_pubsubs(config: Any) -> list[SubscribeAllCapable[Any, Any]]: From 4991f2f927c13c25873b9212c3d204c251aacce0 Mon Sep 17 00:00:00 2001 From: bogwi Date: Mon, 27 Jul 2026 16:06:42 +0900 Subject: [PATCH 2/3] fix: make shard on the path --- dimos/core/transport.py | 4 ++-- dimos/protocol/service/lcmservice.py | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/dimos/core/transport.py b/dimos/core/transport.py index 25b561433f..bdf374ff45 100644 --- a/dimos/core/transport.py +++ b/dimos/core/transport.py @@ -140,7 +140,7 @@ class LCMTransport(PubSubTransport[T]): def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no-untyped-def] super().__init__(LCMTopic(topic, type)) if not hasattr(self, "lcm"): - kwargs.setdefault("url", lcm_url_for_channel(str(self.topic))) + kwargs.setdefault("url", lcm_url_for_channel(topic)) self.lcm = LCM(**kwargs) def start(self) -> None: @@ -171,7 +171,7 @@ def __init__(self, topic: str, type: type, **kwargs) -> None: # type: ignore[no JpegLCM, ) # ~330ms: deferred to avoid pulling in Image/cv2/rerun - kwargs.setdefault("url", lcm_url_for_channel(str(LCMTopic(topic, type)))) + kwargs.setdefault("url", lcm_url_for_channel(topic)) self.lcm = JpegLCM(**kwargs) # type: ignore[assignment] super().__init__(topic, type) diff --git a/dimos/protocol/service/lcmservice.py b/dimos/protocol/service/lcmservice.py index ab3c570b79..2f65afb61b 100644 --- a/dimos/protocol/service/lcmservice.py +++ b/dimos/protocol/service/lcmservice.py @@ -91,7 +91,10 @@ def lcm_url_for_channel(channel: str) -> str: """ if not _STREAM_BUS_URLS: return _DEFAULT_LCM_URL - index = zlib.crc32(channel.encode()) % len(_STREAM_BUS_URLS) + # Typed LCM channels are "/path#pkg.Msg"; shard on the path so transports + # that hash the topic string and spies that see the wire name meet. + path = channel.split("#", 1)[0] + index = zlib.crc32(path.encode()) % len(_STREAM_BUS_URLS) return _STREAM_BUS_URLS[index] From e01328e96f9434caf96b27b5eedcfcd1909fb790 Mon Sep 17 00:00:00 2001 From: bogwi Date: Mon, 27 Jul 2026 16:39:55 +0900 Subject: [PATCH 3/3] update tests --- dimos/agents/conftest.py | 8 +++++--- dimos/agents/mcp/conftest.py | 8 +++++--- dimos/robot/cli/test_topic.py | 6 ++++-- dimos/visualization/rerun/test_viewer_integration.py | 7 +++++-- 4 files changed, 19 insertions(+), 10 deletions(-) diff --git a/dimos/agents/conftest.py b/dimos/agents/conftest.py index 7fe9fecee1..0ea4215b34 100644 --- a/dimos/agents/conftest.py +++ b/dimos/agents/conftest.py @@ -34,7 +34,7 @@ @pytest.fixture -def agent_setup(request, mcp_url: str, lcm_url: str): +def agent_setup(request, mcp_url: str): coordinator = None transports: list[pLCMTransport] = [] unsubs: list = [] @@ -50,8 +50,10 @@ def fn( history: list[BaseMessage] = [] finished_event = Event() - agent_transport: pLCMTransport = pLCMTransport("/agent", url=lcm_url) - finished_transport: pLCMTransport = pLCMTransport("/finished", url=lcm_url) + # No explicit url: each channel shards onto its own bus and the module + # side does the same, so publisher and subscriber meet. + agent_transport: pLCMTransport = pLCMTransport("/agent") + finished_transport: pLCMTransport = pLCMTransport("/finished") transports.extend([agent_transport, finished_transport]) def on_message(msg: BaseMessage) -> None: diff --git a/dimos/agents/mcp/conftest.py b/dimos/agents/mcp/conftest.py index a06224bd8b..bc70ee7365 100644 --- a/dimos/agents/mcp/conftest.py +++ b/dimos/agents/mcp/conftest.py @@ -34,7 +34,7 @@ @pytest.fixture -def agent_setup(request, mcp_url: str, lcm_url: str): +def agent_setup(request, mcp_url: str): coordinator = None transports: list[pLCMTransport] = [] unsubs: list = [] @@ -50,8 +50,10 @@ def fn( history: list[BaseMessage] = [] finished_event = Event() - agent_transport: pLCMTransport = pLCMTransport("/agent", url=lcm_url) - finished_transport: pLCMTransport = pLCMTransport("/finished", url=lcm_url) + # No explicit url: each channel shards onto its own bus and the module + # side does the same, so publisher and subscriber meet. + agent_transport: pLCMTransport = pLCMTransport("/agent") + finished_transport: pLCMTransport = pLCMTransport("/finished") transports.extend([agent_transport, finished_transport]) def on_message(msg: BaseMessage) -> None: diff --git a/dimos/robot/cli/test_topic.py b/dimos/robot/cli/test_topic.py index c2245cd64f..f8af3867d5 100644 --- a/dimos/robot/cli/test_topic.py +++ b/dimos/robot/cli/test_topic.py @@ -21,6 +21,7 @@ from dimos.msgs.geometry_msgs.Vector3 import Vector3 from dimos.msgs.sensor_msgs.CameraInfo import CameraInfo from dimos.protocol.pubsub.impl.lcmpubsub import LCM, Topic +from dimos.protocol.service.lcmservice import lcm_url_for_channel from dimos.robot.cli.topic import _build_eval_context, _decode_typed_lcm_message, topic_send from dimos.utils.testing.collector import CallbackCollector @@ -54,7 +55,7 @@ def test_build_eval_context_maps_message_names_to_classes() -> None: assert cls.__name__ == name -def test_topic_send_delivers_over_lcm(monkeypatch: pytest.MonkeyPatch, lcm_url: str) -> None: +def test_topic_send_delivers_over_lcm(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(global_config, "transport", "lcm") # topic_send never stops the transport it creates; capture it so its LCM @@ -68,7 +69,8 @@ def capturing_make_transport(name: str, msg_type: type) -> PubSubTransport[objec monkeypatch.setattr("dimos.robot.cli.topic.make_transport", capturing_make_transport) - bus = LCM(url=lcm_url) + # Subscribe on the bus this channel shards onto, where topic_send publishes. + bus = LCM(url=lcm_url_for_channel("/test_topic_send")) bus.start() collector = CallbackCollector(1) bus.subscribe(Topic(topic="/test_topic_send", lcm_type=Twist), collector) diff --git a/dimos/visualization/rerun/test_viewer_integration.py b/dimos/visualization/rerun/test_viewer_integration.py index d6a0ba1e95..0c6795eda3 100644 --- a/dimos/visualization/rerun/test_viewer_integration.py +++ b/dimos/visualization/rerun/test_viewer_integration.py @@ -29,6 +29,7 @@ from dimos.core.global_config import GlobalConfig from dimos.protocol.pubsub.impl.lcmpubsub import LCM +from dimos.protocol.service.lcmservice import lcm_bus_urls from dimos.visualization.rerun.bridge import Config, _resolve_pubsubs @@ -132,8 +133,10 @@ def test_legacy_lcm_pubsubs_defers_to_transport_default(self): config = Config(pubsubs=[LCM()], g=GlobalConfig(transport="lcm")) pubsubs = _resolve_pubsubs(config) - assert len(pubsubs) == 1 - assert isinstance(pubsubs[0], LCM) + # The transport default taps every bus: the control bus plus the + # sharded stream pool. + assert all(isinstance(p, LCM) for p in pubsubs) + assert {p.config.url for p in pubsubs} == set(lcm_bus_urls()) def test_explicit_custom_pubsubs_override_is_honored(self): custom = ExplicitPubSubOverride()