Shard LCM stream channels across dedicated multicast buses. DIM - 1337

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:

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:

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(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(topic))
         self.lcm = JpegLCM(**kwargs)  # type: ignore[assignment]
         super().__init__(topic, type)
 

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:

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,67 @@
 )
 
 
+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
+    # 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]
+
+
+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:

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)

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}")

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."""

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]]:

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()