Add Transport benchmarking capabilities to CycloneDDS

[Kaweees]

This merge request brings the work-in-progress changes from (#1038) so that DDSTransport (#1036) can be benchmarked against the transport protocol benchmark suite before merging

[greptile-apps]

Greptile Summary

Adds comprehensive transport benchmarking suite comparing LCM, SharedMemory, Redis, and ROS 2 pub/sub implementations, plus new RawROS transport for direct ROS messaging.

Major changes:

• Implemented RawROS class for ROS 2 pub/sub with configurable QoS profiles (best-effort and reliable modes)
• Added benchmark framework with throughput, latency, and message loss metrics across 12 message sizes (64B to 10MB)
• Enhanced PubSubEncoderMixin to support generic encoding types beyond bytes using new EncodingT type parameter
• Improved logging by reducing SharedMemory start/stop messages from info to debug level
• Added comprehensive type annotations across test files for better type safety

Implementation notes:

• Benchmark uses threading with proper synchronization to measure concurrent pub/sub performance
• Results displayed via rich tables and color-coded heatmaps for throughput, bandwidth, and latency
• ROS implementation spawns background thread for executor spinning with non-blocking spin_once(timeout_sec=0)

Confidence Score: 4/5

• Safe to merge with minor race condition fixes needed in ROS implementation
• Well-structured benchmark suite with proper abstractions and comprehensive testing, but ROS pub/sub has race conditions in publish/subscribe methods where _running and _node are accessed without lock protection while stop() can modify them concurrently
• Pay close attention to dimos/protocol/pubsub/rospubsub.py - fix race conditions in publish() and subscribe() methods before benchmarking under high load

Important Files Changed

Filename	Overview
dimos/protocol/pubsub/rospubsub.py	Adds RawROS and DimosROS classes for ROS 2 pub/sub support with configurable QoS profiles
dimos/protocol/pubsub/benchmark/test_benchmark.py	Implements throughput benchmarking test suite for various transport protocols with parametrized message sizes
dimos/protocol/pubsub/benchmark/testdata.py	Configures test cases for LCM, SHM, Redis, and ROS transports with message generators
dimos/protocol/pubsub/spec.py	Adds EncodingT type parameter to PubSubEncoderMixin for flexible encoding types beyond bytes

Sequence Diagram

sequenceDiagram
    participant Test as test_throughput
    participant PS as PubSub Implementation
    participant Sub as Subscriber Thread
    participant Pub as Publisher Loop
    
    Test->>PS: subscribe(topic, callback)
    PS-->>Test: unsubscribe_fn
    Test->>Test: sleep(0.1) warmup
    
    Note over Test,Pub: Benchmarking Phase
    Test->>Pub: Start publishing loop
    
    loop Until time_limit OR max_messages OR all_received
        Pub->>PS: publish(topic, msg)
        PS->>Sub: callback(msg, topic)
        Sub->>Sub: received_count++
        Sub->>Sub: Check if target reached
        opt All messages received
            Sub->>Test: all_received.set()
        end
    end
    
    Pub->>Test: Record publish_end time
    Test->>Test: Update target_count to actual sent
    
    opt Not all received yet
        Test->>Test: wait(RECEIVE_TIMEOUT)
    end
    
    Test->>Test: Record latency_end time
    Test->>Test: Calculate metrics (throughput, latency, loss)
    Test->>Test: Add result to benchmark_results

Loading

[greptile-apps]

_{11 files reviewed, 4 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

logic: Race condition - _running and _node are checked without lock protection, while stop() modifies them from another thread

Suggested change

	if not self._running or not self._node:
	return
	with self._lock:
	if not self._running or not self._node:
	return

[greptile-apps]

logic: Race condition - check happens outside lock, then _node is used inside lock. stop() could set _node = None between the check and use

Suggested change

	if not self._running or not self._node:
	raise RuntimeError("ROS pubsub not started")
	with self._lock:
	if not self._running or not self._node:
	raise RuntimeError("ROS pubsub not started")

[greptile-apps]

style: nonlocal received_count is redundant since received_count is already modified within lock context

_{Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!}

[greptile-apps]

style: Duplicate transport name extraction logic - same code exists in pubsub_id() function at lines 68-73

_{Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!}