Create DDS Transport Protocol

[Kaweees]

This merge request implements DDS (Data Distribution Service) transport layer using CycloneDDS, providing a new high-performance pub/sub transport option.

Quick Start

Note

We currently use Eclipse Cyclone DDS as our DDS implementation. Its IdlStruct feature lets you define DDS topic types in pure Python, eliminating the need for separate IDL files, with automatic serialization support.

from dataclasses import dataclass
from cyclonedds.idl import IdlStruct
from dimos.protocol.pubsub.ddspubsub import DDS, Topic

@dataclass
class SensorReading(IdlStruct):
    value: float

dds = DDS()
dds.start()

received = []
sensor_topic = Topic(name="sensors/temperature", data_type=SensorReading)

dds.subscribe(sensor_topic, lambda msg, t: received.append(msg))
dds.publish(sensor_topic, SensorReading(value=22.5))

import time
time.sleep(0.1)

print(f"Received: {received}")
dds.stop()

Received: [SensorReading(value=22.5)]

Unit Tests/Benchmarks

uv run pytest -svm tool dimos/protocol/pubsub/benchmark/test_benchmark.py --override-ini="addopts=" -k "DDS"
uv run pytest -svm tool dimos/protocol/pubsub/benchmark/test_benchmark.py

This builds off of #1036, which was closed because the branch was renamed to miguel/dds_transport

[greptile-apps]

Greptile Overview

Greptile Summary

This PR implements DDS (Data Distribution Service) transport using Eclipse Cyclone DDS, providing a new pub/sub transport option alongside existing LCM, ROS, and shared memory transports.

Key additions:

• DDSService - Manages singleton DomainParticipant for DDS domain coordination
• DDS class - PubSub implementation with thread-safe DataWriter/DataReader management
• DDSTransport - Transport layer integration with auto-start on first use
• Benchmark test cases comparing DDS performance against other transports
• Documentation and examples for usage

Critical issues found:

• DDSService.stop() doesn't properly clean up the DomainParticipant resource (line 62-63)
• DDSTransport has race conditions in broadcast() and subscribe() where locks are released before calling DDS methods (lines 280-290)
• Missing idempotency checks in start() and stop() methods could lead to double-initialization

Note: Several issues from previous review threads remain unaddressed, including the string formatting error in ddspubsub.py and duplicate imports in transport.py.

Confidence Score: 3/5

• This PR has multiple race conditions and resource cleanup issues that could cause instability in multi-threaded environments
• Score reflects several critical concurrency bugs in DDSTransport (lock release before method calls) and missing resource cleanup in DDSService, plus unresolved issues from previous review threads
• Pay close attention to dimos/core/transport.py (race conditions in broadcast/subscribe) and dimos/protocol/service/ddsservice.py (missing resource cleanup)

Important Files Changed

Filename	Overview
dimos/protocol/pubsub/ddspubsub.py	New DDS pubsub implementation with thread-safe listener pattern
dimos/protocol/service/ddsservice.py	DDS service with global singleton participant, missing cleanup logic
dimos/core/transport.py	Added DDSTransport class with thread-safe initialization

Sequence Diagram

sequenceDiagram
    participant App
    participant DDSTransport
    participant DDS
    participant DDSService
    participant DomainParticipant
    participant DataWriter
    participant DataReader
    participant Listener

    Note over App,Listener: Initialization and Publishing Flow
    App->>DDSTransport: __init__(topic, type)
    DDSTransport->>DDS: __init__(qos)
    DDS->>DDSService: __init__()
    
    App->>DDSTransport: broadcast(msg)
    DDSTransport->>DDSTransport: Check _started flag
    DDSTransport->>DDS: start()
    DDS->>DDSService: start()
    DDSService->>DomainParticipant: Create singleton
    
    DDSTransport->>DDS: publish(topic, msg)
    DDS->>DDS: _get_writer(topic)
    DDS->>DataWriter: Create if not exists
    DDS->>DataWriter: write(message)

    Note over App,Listener: Subscription Flow
    App->>DDSTransport: subscribe(callback)
    DDSTransport->>DDS: subscribe(topic, callback)
    DDS->>DDS: _get_listener(topic)
    DDS->>Listener: Create listener if not exists
    DDS->>DataReader: Create with listener
    DDS->>Listener: add_callback(callback)
    
    Note over App,Listener: Message Reception
    DataWriter->>DataReader: DDS message arrives
    DataReader->>Listener: on_data_available(reader)
    Listener->>DataReader: take()
    loop For each sample
        Listener->>App: callback(sample, topic)
    end

Loading

[greptile-apps]

_{5 files reviewed, 3 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{5 files reviewed, no comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{5 files reviewed, 6 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{3 files reviewed, 5 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

Setting _participant to None doesn't actually clean up the DomainParticipant resource. Should call .close() or similar cleanup method before setting to None.

Suggested change

	if _participant is not None:
	_participant = None
	if _participant is not None:
	_participant.close()
	_participant = None

[greptile-apps]

Lock is released before calling publish(), allowing race condition where start() could be called again by another thread before publish executes. Should keep lock held or check _started again inside the lock.

Suggested change

	def broadcast(self, _, msg) -> None: # type: ignore[no-untyped-def]
	with self._start_lock:
	if not self._started:
	self.start()
	self.dds.publish(self.topic, msg)
	def broadcast(self, _, msg) -> None: # type: ignore[no-untyped-def]
	with self._start_lock:
	if not self._started:
	self.start()
	self.dds.publish(self.topic, msg)

[greptile-apps]

Same lock release issue - subscribe() releases lock before calling dds.subscribe(), allowing race condition.

Suggested change

	def subscribe(self, callback: Callable[[T], None], selfstream: In[T] = None) -> None: # type: ignore[assignment, override]
	with self._start_lock:
	if not self._started:
	self.start()
	return self.dds.subscribe(self.topic, lambda msg, topic: callback(msg)) # type: ignore[return-value]
	def subscribe(self, callback: Callable[[T], None], selfstream: In[T] = None) -> None: # type: ignore[assignment, override]
	with self._start_lock:
	if not self._started:
	self.start()
	return self.dds.subscribe(self.topic, lambda msg, topic: callback(msg)) # type: ignore[return-value]

[greptile-apps]

Calling start() recursively while holding _start_lock (an RLock) could lead to nested start calls if start() triggers any code path that calls broadcast() or subscribe(). Check _started flag before calling dds.start().

Suggested change

	def start(self) -> None:
	with self._start_lock:
	self.dds.start()
	self._started = True
	def start(self) -> None:
	with self._start_lock:
	if not self._started:
	self.dds.start()
	self._started = True

[greptile-apps]

Check _started flag to avoid calling stop() multiple times

Suggested change

	def stop(self) -> None:
	with self._start_lock:
	self.dds.stop()
	self._started = False
	def stop(self) -> None:
	with self._start_lock:
	if self._started:
	self.dds.stop()
	self._started = False