Repository navigation
Make CAN receive interruptible and bounded - #6
Merged
Merged
Conversation
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars. ⚙️ Run configuration
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Reviewer's GuideThe PR makes SocketCAN receive shutdown-safe by polling the socket with an eventfd wake mechanism, removes the read-loop throttle, and makes initialization transactional. It adds ownership-safe dependency injection for socketpair testing, regression tests for interruption, shutdown, throughput, ordering, and retry behavior, plus build and usage documentation. Sequence diagram for interruptible CAN shutdownsequenceDiagram
participant Piper as PiperInterfaceV2
participant Reader as readLoop
participant CAN as StdCanInterface
participant Wake as eventfd
Piper->>Reader: stop_read_.store(true)
Piper->>CAN: interruptRead()
CAN->>Wake: eventfd_write()
Reader->>CAN: readCanMessage()
CAN->>Wake: poll()
Wake-->>CAN: wake event
CAN-->>Reader: false
Reader-->>Piper: thread exits
Piper->>Reader: join()
Sequence diagram for bounded CAN initialization retrysequenceDiagram
participant Client as Caller
participant CAN as StdCanInterface
participant Kernel as SocketCAN
Client->>CAN: init()
CAN->>Kernel: socket()
Kernel-->>CAN: fd
CAN->>Kernel: ioctl()
Kernel-->>CAN: failure
CAN->>Kernel: close(fd)
CAN-->>Client: false
Client->>CAN: init()
CAN->>Kernel: socket()
Kernel-->>CAN: new fd
CAN->>Kernel: bind()
Kernel-->>CAN: success
CAN-->>Client: true
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
There was a problem hiding this comment.
🟡 Changes recommended
Constructor failures leak eventfds, and test linking is incompatible with supported older CMake versions.
2 open findings
What changed in this PR
Makes CAN reception interruptible for prompt shutdown and removes the per-frame delay to improve throughput.
Changes:
- Adds eventfd-based wakeups and retry-safe socket initialization.
- Adds transport-adopting constructors for hardware-free tests.
- Introduces six regression tests and documents test setup.
| File | Description |
|---|---|
| piper_cpp/test/std_can_interface_test.cpp | Tests interrupts, ordering, and initialization retries. |
| piper_cpp/test/piper_interface_v2_test.cpp | Tests shutdown and burst processing. |
| piper_cpp/test/CMakeLists.txt | Registers GTest tests with timeouts. |
| piper_cpp/src/interface/piper_interface_v2.cpp | Interrupts shutdown reads and removes receive delays. |
| piper_cpp/README.md | Documents test dependencies and commands. |
| piper_cpp/include/piper_cpp/std_can_interface.h | Adds interruptible reads and safer socket ownership. |
| piper_cpp/include/piper_cpp/interface/piper_interface_v2.h | Declares transport adoption and documents shutdown behavior. |
| piper_cpp/CMakeLists.txt | Enables optional unit-test builds. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 10 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="piper_cpp/include/piper_cpp/std_can_interface.h" line_range="188" />
<code_context>
+ /// Wake a thread blocked in `readCanMessage()`, which then returns false. Safe to call from
+ /// any thread. If no read is in progress, the next `readCanMessage()` returns false once.
+ void interruptRead() { eventfd_write(wake_fd_, 1); }
+
+ /// Replace the receive callback. Not synchronised with `readCanMessage()`; set it before
</code_context>
<issue_to_address>
**Shutdown can hang indefinitely**
When a signal interrupts `eventfd_write()` during shutdown, `interruptRead()` ignores the failed write, so the blocked reader receives no wake signal and `disconnectPort()` waits indefinitely in `join()`.
Retry `eventfd_write()` when it fails with `EINTR`, and handle any other write failure so the reader is not left blocked.
</issue_to_address>
### Comment 2
<location path="piper_cpp/src/interface/piper_interface_v2.cpp" line_range="98" />
<code_context>
- // The callback will handle parsing and updating state
- }
- std::this_thread::sleep_for(std::chrono::milliseconds(1)); // TODO: Adjust sleep duration as needed
+ arm_can_->readCanMessage();
}
}
</code_context>
<issue_to_address>
**Persistent receive errors spin CPU**
When the CAN socket repeatedly reports a poll or receive error, or reports `POLLHUP`, `readCanMessage()` returns false on receive errors, but `readLoop()` immediately calls it again, consuming CPU while the failure persists. On EOF, `readCanMessage()` passes an uninitialized frame to the callback.
Handle failed reads and EOF before retrying: do not pass an invalid frame to the callback, and exit or pause the loop on persistent errors.
</issue_to_address>
### Comment 3
<location path="piper_cpp/include/piper_cpp/std_can_interface.h" line_range="36" />
<code_context>
- : channel_name_(std::move(channel_name)), expected_bitrate_(expected_bitrate), callback_(std::move(callback)),
- sock_(-1)
+ : channel_name_(std::move(channel_name)), expected_bitrate_(expected_bitrate), sock_(-1),
+ wake_fd_(makeWakeFd()), callback_(std::move(callback))
{
if (judge_flag)
</code_context>
<issue_to_address>
**Failed construction leaks descriptors**
When CAN validation throws during construction, or wake-FD creation fails in the adopting constructor, the normal constructor creates `wake_fd_` before `judgeCanInfo()` can throw, and the destructor does not run when construction fails, leaking the eventfd on each retry. In the adopting constructor, a failure to create the wake FD also leaves the adopted socket open.
Manage the wake FD and adopted socket with RAII so both are closed if construction fails.
Also at `piper_cpp/include/piper_cpp/std_can_interface.h:47`.
</issue_to_address>
### Comment 4
<location path="piper_cpp/include/piper_cpp/interface/piper_interface_v2.h" line_range="78" />
<code_context>
- /// Idempotent.
+ /// Stop the background read thread (if running) and close the CAN port. Wakes the read
+ /// thread out of its blocking receive and joins it, so this returns promptly even on a
+ /// silent bus. `timeout` is currently unused and kept for API compatibility. Idempotent.
void disconnectPort(std::chrono::milliseconds timeout = std::chrono::milliseconds{100});
</code_context>
<issue_to_address>
**Disconnect exceeds its timeout**
When the read thread does not finish within the caller’s requested timeout, `disconnectPort()` unconditionally joins the read thread, so the call blocks beyond `timeout`; callers such as ROS cleanup lose the requested time bound.
Make `disconnectPort()` honor the requested timeout when waiting for the read thread.
Also at `piper_cpp/include/piper_cpp/interface/piper_interface_v2.h:80`.
</issue_to_address>
### Comment 5
<location path="piper_cpp/src/interface/piper_interface_v2.cpp" line_range="51-52" />
<code_context>
+)
+ : PiperInterfaceV2("", /*judge_flag=*/false, /*auto_init=*/false, dh_offset, sdk_joint_limit, sdk_gripper_limit)
+{
+ arm_can_ = std::move(can);
+ arm_can_->setCallback([this](const struct can_frame &frame, double timestamp) { parseCANFrame(frame, timestamp); });
+}
+
</code_context>
<issue_to_address>
**Null transport crashes construction**
When a caller passes an empty `std::unique_ptr<StdCanInterface>`, `arm_can_` receives the moved `can` pointer and `setCallback()` dereferences it, so passing a null `unique_ptr` crashes instead of producing a clear construction error.
Reject a null transport with a documented precondition or exception before dereferencing it.
</issue_to_address>
### Comment 6
<location path="piper_cpp/include/piper_cpp/std_can_interface.h" line_range="47" />
<code_context>
+ /// Adopt an already-open socket that delivers one `can_frame` per datagram (a raw CAN socket,
+ /// or a socketpair in tests). Takes ownership of `socket_fd`; `init()` is a no-op.
+ explicit StdCanInterface(int socket_fd, CanCallback callback = nullptr)
+ : channel_name_(), expected_bitrate_(0), sock_(socket_fd), wake_fd_(makeWakeFd()),
+ callback_(std::move(callback))
+ {
</code_context>
<issue_to_address>
**Adopted transports cannot reconnect**
When an interface constructed with the fd-adopting overload is disconnected and then reconnected, after `disconnectPort()` closes the adopted fd, the next `connectPort()` calls `init()`. Since `sock_` is now `-1`, `init()` opens a new CAN socket and looks up the empty `channel_name_`, so it fails and callers cannot reconnect this interface.
Preserve enough information to reopen the adopted transport, or explicitly make the adopted interface one-shot and prevent reconnect attempts.
</issue_to_address>
### Comment 7
<location path="piper_cpp/include/piper_cpp/std_can_interface.h" line_range="47" />
<code_context>
+ /// Adopt an already-open socket that delivers one `can_frame` per datagram (a raw CAN socket,
+ /// or a socketpair in tests). Takes ownership of `socket_fd`; `init()` is a no-op.
+ explicit StdCanInterface(int socket_fd, CanCallback callback = nullptr)
+ : channel_name_(), expected_bitrate_(0), sock_(socket_fd), wake_fd_(makeWakeFd()),
+ callback_(std::move(callback))
+ {
</code_context>
<issue_to_address>
**Adopted sockets lose timestamps**
When a caller adopts an already-open raw CAN socket without `SO_TIMESTAMP` enabled, `readCanMessage()` leaves `timestamp` at 0.0 when an adopted raw CAN socket has not enabled `SO_TIMESTAMP`; the regular `init()` path enables that option, but the adopting constructor skips it. `PiperInterfaceV2` then publishes epoch-zero timestamps in its state snapshots.
Enable or verify `SO_TIMESTAMP` for adopted raw CAN sockets, or define a valid fallback timestamp.
</issue_to_address>
### Comment 8
<location path="piper_cpp/test/CMakeLists.txt" line_range="7" />
<code_context>
+# A regression in the shutdown path shows up as a hang, so bound every test.
+function(add_piper_test name)
+ add_executable(${name} ${name}.cpp)
+ target_link_libraries(${name} PRIVATE piper_cpp GTest::gtest_main)
+ gtest_discover_tests(${name} PROPERTIES TIMEOUT 30)
+endfunction()
</code_context>
<issue_to_address>
**Tests fail with older CMake**
When CMake 3.10–3.19 uses `FindGTest` and the installed package does not define `GTest::gtest_main`, `target_link_libraries()` names `GTest::gtest_main`, which the older module does not provide, so test configuration or generation fails and the tests cannot build.
Link against a GTest target available from the older module, or select the target based on what the installed GTest package provides.
</issue_to_address>
### Comment 9
<location path="piper_cpp/test/CMakeLists.txt" line_range="1" />
<code_context>
+find_package(GTest REQUIRED)
+include(GoogleTest)
+
</code_context>
<issue_to_address>
**Default build requires GTest**
When a normal build environment lacks GTest, cMake configuration fails at `find_package(GTest REQUIRED)` when GTest is absent, so building the library and executables now requires a test-only development dependency unless every such build explicitly disables testing.
Make tests opt-in or discover GTest quietly and skip adding test targets when it is unavailable.
</issue_to_address>
### Comment 10
<location path="piper_cpp/test/std_can_interface_test.cpp" line_range="124" />
<code_context>
+ // No such interface, so init() fails at the ifindex lookup. Before the fix the half-open
+ // socket was kept and a second init() reported success.
+ StdCanInterface can("piper_cpp_no_such_iface", 1000000, /*judge_flag=*/false, /*auto_init=*/true);
+ EXPECT_FALSE(can.init());
+}
</code_context>
<issue_to_address>
**Initialization regression goes undetected**
When a future change reintroduces the half-open-socket behavior, or the test runs without permission to create a raw CAN socket, the test makes only one explicit `init()` call and does not establish that it reached the invalid-interface lookup. It can therefore pass without detecting the regression, including when socket creation itself fails.
Ensure the test reaches the invalid-interface lookup, then call `init()` again and verify that the failed socket was not retained.
</issue_to_address>Sourcery assessment
Approval pending. 10 findings to address first.
Blocking findings: piper_cpp/include/piper_cpp/std_can_interface.h:188, piper_cpp/src/interface/piper_interface_v2.cpp:98, piper_cpp/include/piper_cpp/std_can_interface.h:36, piper_cpp/include/piper_cpp/interface/piper_interface_v2.h:78, piper_cpp/src/interface/piper_interface_v2.cpp:52, and 5 more
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes the transport-level issues from the review: shutdown could hang forever on a silent
bus, the read loop capped throughput at about one frame per millisecond, and a failed
init() left a half-open socket that a later init() reported as open.
interruptRead() unblocks it from any thread. disconnectPort() uses that before joining.
non-copyable.
let the tests drive everything through a socketpair with no hardware.
the silent-bus shutdown, interrupt semantics, a 3000-frame burst reaching the state
cache, ordering under buffer saturation, and init retry.
Nothing on the wire or in the decoders changes. On hardware the visible differences are
lower receive latency and CPU, and a prompt disconnect with the arm powered off.
disconnectPort(timeout) keeps its parameter for API compatibility; it is now unused.
Short-read validation of recvmsg is left for the frame-validation PR.
Summary by Sourcery
Make CAN reception interruptible, promptly stoppable, and capable of draining traffic without artificial delays while adding regression coverage.
New Features:
Bug Fixes:
Enhancements:
Build:
Documentation:
Tests:
Chores: