Rpc fixes merge

[paul-nechifor]

• implements exception passing to RPC clients (if there is an exception on the server, it will be raised on the caller side)

• stress tests for all transport protocols (we expect 10k messages < sec, 100% reliability expected across all transports)

• stress tests for RPC (1k rpc calls < sec, 100% reliability expected) - SHM actually fails this - has 25% reliability on RPC tests due to how it handles message reception, this is an issue that means we can't use SHM across the board in dimos

• moves rpc call threadpool from LCMRPC to the rpc base so it's used cross-protocol

• shares a single cpp LCM instance/loop per process for all instantiated transports

• investigated kernel networking stats and LCM cpp implementation, optimized details around this to get LCM to perform much better

[chatgpt-codex-connector]

Codex usage limits have been reached for code reviews. Please check with the admins of this repo to increase the limits by adding credits.
Credits must be used to enable repository wide code reviews.

[greptile-apps]

Greptile Overview

Greptile Summary

This PR refactors the RPC subsystem to improve reliability, performance, and exception handling.

Key changes:

• Consolidated RPC implementation into pubsubrpc.py, removing separate lcmrpc.py file
• Added thread pool for RPC handler execution to prevent deadlock in nested RPC calls
• Implemented shared response subscriptions (one per RPC name instead of one per call) to reduce overhead
• Added exception serialization/deserialization to propagate server-side exceptions to clients
• Introduced RemoteError wrapper for non-builtin exceptions with preserved stack traces
• Increased UDP buffer sizes (64MB max, 16MB default) and LCM subscription queue capacity (10k) for high-throughput scenarios
• Added defensive hasattr checks in Service base class for multiple inheritance compatibility
• Comprehensive test coverage with new grid tests for all RPC implementations

Minor fixes:

• Fixed resource cleanup in AgentSpec.stop() and test_coordinator.py
• Removed redundant LCM instantiation in lcmspy.py
• Added pcre2 dependency to Nix flake

Confidence Score: 5/5

• This PR is safe to merge - it refactors RPC infrastructure with improved thread safety, exception handling, and comprehensive test coverage.
• The changes are well-structured with proper thread synchronization (locks for thread pool, response subscriptions, and message ID counter). The code handles edge cases like shutdown from within pool threads, proper cleanup of resources, and exception propagation. The new test suite covers sync/async calls, exception handling, timeouts, concurrent calls, and multiple services. No breaking API changes detected.
• No files require special attention - all changes follow consistent patterns and are well-tested.

Important Files Changed

File Analysis

Filename	Score	Overview
dimos/protocol/rpc/pubsubrpc.py	5/5	Major refactoring of RPC mixin: added thread pool for handler execution, shared response subscriptions to reduce overhead, exception serialization support, and consolidated LCMRPC/ShmRPC classes. All changes improve reliability and performance.
dimos/protocol/rpc/spec.py	5/5	Added exception handling in call_sync and call_async to propagate remote exceptions properly to callers.
dimos/protocol/rpc/rpc_utils.py	5/5	New utility module for serializing/deserializing exceptions across RPC boundaries with RemoteError wrapper class for non-builtin exceptions.
dimos/protocol/service/lcmservice.py	5/5	Increased UDP buffer size requirements from 2MB to 64MB (rmem_max) and 16MB (rmem_default) to handle higher message throughput.
dimos/protocol/pubsub/lcmpubsub.py	5/5	Simplified __init__ to rely on super() MRO; added subscription queue capacity of 10000 to handle high-volume message bursts.
dimos/protocol/rpc/test_spec.py	5/5	New comprehensive grid test suite for RPC implementations covering sync/async calls, exception handling, timeouts, concurrent calls, and multiple services.

Sequence Diagram

sequenceDiagram
    participant Client
    participant PubSubRPCMixin
    participant ThreadPool
    participant Server
    participant RPCHandler

    Note over Client,RPCHandler: RPC Call Flow with Exception Handling

    Client->>PubSubRPCMixin: call_sync(name, args)
    PubSubRPCMixin->>PubSubRPCMixin: Generate unique msg_id
    PubSubRPCMixin->>PubSubRPCMixin: Register callback in shared subscription
    PubSubRPCMixin-->>Server: Publish request to /rpc/{name}/req

    Server->>PubSubRPCMixin: Receive request
    PubSubRPCMixin->>ThreadPool: Submit handler execution
    ThreadPool->>RPCHandler: Execute in thread pool

    alt Success
        RPCHandler-->>ThreadPool: Return result
        ThreadPool->>PubSubRPCMixin: Publish response
        PubSubRPCMixin-->>Client: Publish to /rpc/{name}/res
        Client->>PubSubRPCMixin: Callback invoked with result
        PubSubRPCMixin->>Client: Return result
    else Exception
        RPCHandler--xThreadPool: Raise exception
        ThreadPool->>PubSubRPCMixin: serialize_exception()
        PubSubRPCMixin-->>Client: Publish exception data
        Client->>PubSubRPCMixin: Callback invoked with exception
        PubSubRPCMixin->>Client: deserialize_exception() and raise
    end

Loading

[greptile-apps]

_{23 files reviewed, no comments}

_{Edit Code Review Agent Settings | Greptile}