Streamline transport start/stop methods

[Kaweees]

Several transport classes (LCMTransport, JpegLcmTransport, pSHMTransport, SHMTransport, JpegShmTransport) had empty start() and stop() method stubs (...). These are now fully implemented to properly start/stop the underlying transport and manage the _started flag.

uv run pytest -svm tool dimos/protocol/pubsub/benchmark/test_benchmark.py

[greptile-apps]

Greptile Summary

This PR refactors transport lifecycle management by implementing previously empty start() and stop() method stubs across multiple transport classes (pLCMTransport, LCMTransport, JpegLcmTransport, pSHMTransport, SHMTransport, JpegShmTransport).

Key changes:

• Centralized transport initialization logic into dedicated start() methods instead of inlining it in broadcast() and subscribe()
• stop() methods now properly reset the _started flag to False, enabling proper restart capability
• Improved code reusability by replacing duplicate transport.start() + _started = True blocks with calls to self.start()
• All transport classes now properly fulfill the abstract Resource interface requirements

The refactoring maintains backward compatibility as the lazy initialization behavior (starting on first use) remains unchanged.

Confidence Score: 4/5

• Safe to merge with good refactoring that improves code organization
• The refactoring is well-structured and maintains backward compatibility. The changes centralize transport lifecycle management, making the code more maintainable. The PR doesn't introduce new bugs - it simply reorganizes existing logic into dedicated methods. The existing race condition (check-then-act on _started flag) was present before this change and is not introduced by this refactoring.
• No files require special attention

Important Files Changed

Filename	Overview
dimos/core/transport.py	Implements start/stop methods in transport classes, centralizes transport lifecycle management

Sequence Diagram

sequenceDiagram
    participant Client
    participant Transport as Transport (LCM/SHM)
    participant Backend as Backend (lcm/shm)
    
    Note over Client,Backend: Before: Inline start logic
    Client->>Transport: broadcast(msg) / subscribe(callback)
    Transport->>Transport: Check _started flag
    alt not started
        Transport->>Backend: lcm.start() / shm.start()
        Transport->>Transport: Set _started = True
    end
    Transport->>Backend: publish(msg) / subscribe(callback)
    
    Note over Client,Backend: After: Centralized start() method
    Client->>Transport: broadcast(msg) / subscribe(callback)
    Transport->>Transport: Check _started flag
    alt not started
        Transport->>Transport: start()
        Transport->>Backend: lcm.start() / shm.start()
        Transport->>Transport: Set _started = True
    end
    Transport->>Backend: publish(msg) / subscribe(callback)
    
    Note over Client,Backend: Lifecycle management
    Client->>Transport: stop()
    Transport->>Backend: lcm.stop() / shm.stop()
    Transport->>Transport: Set _started = False

Loading

[greptile-apps]

Additional Comments (6)

1. dimos/core/transport.py, line 63-66 (link)

   style: pLCMTransport wasn't refactored to use self.start() like the other transport classes. Should call self.start() instead for consistency.

   _{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!}

2. dimos/core/transport.py, line 73-75 (link)

   style: Same as in broadcast() - should use self.start() for consistency.

   _{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!}

3. dimos/core/transport.py, line 78 (link)

   style: pLCMTransport.start() should implement the method body like other transport classes.

   _{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!}

4. dimos/core/transport.py, line 80-81 (link)

   style: pLCMTransport.stop() should reset _started flag like other transport classes.

5. dimos/core/transport.py, line 142-146 (link)

   style: pSHMTransport.broadcast() wasn't refactored to use self.start() like the other transport classes. Should call self.start() for consistency.

   _{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!}

6. dimos/core/transport.py, line 149-152 (link)

   style: Same as in broadcast() - should use self.start() for consistency.

   _{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!}

_{1 file reviewed, 6 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{1 file reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

style: check-then-act race condition in concurrent scenarios - if two threads call broadcast() simultaneously when _started is False, both could call start() leading to multiple start calls on the underlying transport

Suggested change

	if not self._started:
	self.lcm.start()
	self._started = True
	self.start()
	with getattr(self, '_start_lock', threading.Lock()):
	if not self._started:
	self.start()

[leshy]

Actually greptile is pointing out a legitimate issue here, LCM start() is instant and almost cannot fail (udp cannot fail to connect) so this is why we are so chill, ignoring start() and autostarting in broadcast() to save some resources.

Other protocols are slower, have legitimate slower blocking Start() processes, that can fail with "connection failiure" and such, so we expect them to have fat start() functions (ROS, some TCP connection etc, even SSH)

It makes sense to remove autostart from LCM if we want to be extra nitpicky, that's ok.
but definitely other protocols are not supposed to autostart like this and should have legitimate start() implementations

since we want our broadcast() and subscribe() functions to be instant, we are assuming tge transport is ready at that point, these shouldn't trigger unanticipated slow connection attempts, (especially without locks as greptile mentions, for like camera you'll get actual 30 parallel connection attempts per second - if we run DDS over ssh for agibot, this will be 100reds of SSH connection attemps, or (with a lock) a weird invisible message buffer filling)

[leshy]

we are autoiniting SHM and LCM which are instant so this is fine, but we won't be doing the same for other protocols