Send only to opened queues - #1871
Conversation
📝 WalkthroughWalkthroughThe change modifies ChangesClosed Queue Check in Message Dispatch
Estimated code review effort: 1 (Trivial) | ~3 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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 |
There was a problem hiding this comment.
Pull request overview
This PR updates host-side Node::Output message dispatch to avoid sending to queues that have been closed (notably when a topic is removed and later re-added), preventing the output path from getting stuck on defunct queue connections.
Changes:
- Guard
Node::Output::send()so it only sends to non-closed connected queues.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| for(auto& messageQueue : connectedInputs) { | ||
| messageQueue->send(msg); | ||
| if(!messageQueue->isClosed()) messageQueue->send(msg); | ||
| } |
There was a problem hiding this comment.
Out of curiosity, where did you hit this? Seems like something that would be hit quite often so I am surprised we haven't hit it yet
There was a problem hiding this comment.
This PR fixes the issue Removal of RemoteConnection before pipeline is started doesn't work. In most cases if a queue closes it means that the pipeline is shutting down. I don't believe closing a queue during runtime is handled very well in some parts of Depthai.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/pipeline/Node.cpp`:
- Around line 247-251: `Node::Output::trySend()` still attempts
`MessageQueue::trySend()` on every connected input, which can throw if a queue
is already closed. Update the send loop in `Node::Output::trySend()` to mirror
the `sendToInputs` logic by checking `isClosed()` on each `messageQueue` before
calling `trySend()`, so closed inputs are skipped instead of causing
non-blocking sends to fail.
- Around line 247-251: The `sendToInputs` lambda in `Node::send()` still needs
the same closed-queue guard for `MessageQueue::trySend()` as it already has for
`send()`. Update the loop over `connectedInputs` so it checks
`messageQueue->isClosed()` before invoking either send path, and ensure any
`trySend()` call is skipped for closed queues to keep behavior consistent with
the guarded `send()` logic.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 397a52ab-8d4d-4ac7-a52e-a5d25a9b2b0f
📒 Files selected for processing (1)
src/pipeline/Node.cpp
📜 Review details
⏰ Context from checks skipped due to timeout. (1)
- GitHub Check: copilot-pull-request-reviewer
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2026-03-24T22:39:04.364Z
Learnt from: MaticTonin
Repo: luxonis/depthai-core PR: 1732
File: src/pipeline/Pipeline.cpp:705-705
Timestamp: 2026-03-24T22:39:04.364Z
Learning: Do not flag the `!= ""` part of the auto-calibration condition as redundant when it appears in `PipelineImpl::build()` (or closely related pipeline build logic). If the code uses `utility::getEnvAs<std::string>(..., default)` with a default such as `"ON_START"`, the explicit empty-string guard may still be intentional to treat an explicitly empty env var as “OFF/disabled” (or to avoid special-casing elsewhere). Only consider removing `!= ""` if the codebase has an explicit, enforceable guarantee that `DEPTHAI_AUTOCALIBRATION` can never be set to an empty string (e.g., via validated parsing/CI checks); otherwise, keep the guard.
Applied to files:
src/pipeline/Node.cpp
| auto sendToInputs = [this, &msg]() { | ||
| for(auto& messageQueue : connectedInputs) { | ||
| messageQueue->send(msg); | ||
| if(!messageQueue->isClosed()) messageQueue->send(msg); | ||
| } | ||
| }; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
depthai-core MessageQueue trySend behavior on closed queue
💡 Result:
In the depthai-core library, the trySend method of the MessageQueue class attempts to place a message into the queue [1][2]. When a queue has been closed (or destroyed), attempting to interact with it—such as through has, tryGet, or similar accessors—typically triggers a QueueException with the message CLOSED_QUEUE_MESSAGE [3][4]. While explicit public documentation on the specific return value of trySend on a closed queue is limited, the internal implementation pattern for queue operations in depthai-core consistently checks for the closed/destroyed state [3][4]. If a queue is in a closed state, callers should anticipate that the operation will either throw a QueueException or return a status indicating failure, depending on the specific version and internal checks [3][4]. It is best practice to verify the state of the queue using isClosed before attempting to send messages to avoid exceptions or unexpected behavior in the pipeline [1][2].
Citations:
- 1: https://docs.ros.org/en/humble/p/depthai_v3/generated/classdai_1_1MessageQueue.html
- 2: https://docs.ros.org/en/kilted/p/depthai/generated/classdai_1_1MessageQueue.html
- 3: https://docs.ros.org/en/humble/p/depthai_v3/generated/program_listing_file_include_depthai_pipeline_MessageQueue.hpp.html
- 4: https://docs.ros.org/en/kilted/p/depthai/generated/program_listing_file_include_depthai_pipeline_MessageQueue.hpp.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the relevant source and implementation details.
git ls-files | rg '(^|/)Node\.cpp$|(^|/)MessageQueue(\.h|\.hpp|\.cpp)$|(^|/)MessageQueue'
echo '--- Node.cpp around trySend/send ---'
sed -n '220,310p' src/pipeline/Node.cpp
echo '--- MessageQueue declarations/definitions ---'
for f in $(git ls-files | rg 'MessageQueue(\.h|\.hpp|\.cpp)$'); do
echo "### $f"
sed -n '1,260p' "$f" | rg -n 'trySend|send|isClosed|closed|CLOSED_QUEUE_MESSAGE|QueueException|destroy'
doneRepository: luxonis/depthai-core
Length of output: 6048
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# If the API docs are available in the repo, inspect the exact contract for trySend.
git ls-files | rg 'docs|MessageQueue'Repository: luxonis/depthai-core
Length of output: 535
🌐 Web query:
depthai MessageQueue trySend closed queue QueueException CLOSED_QUEUE_MESSAGE source
💡 Result:
In the DepthAI (dai) library, the error dai.MessageQueue.QueueException: MessageQueue was closed occurs when you attempt to interact with a queue (such as calling trySend, get, or tryGet) after the queue has been closed [1][2][3]. The string "MessageQueue was closed" corresponds to the static constant CLOSED_QUEUE_MESSAGE used within the MessageQueue class to signal this state [1][2]. Key causes for this exception include: - Pipeline Termination: The pipeline has stopped running, rendering its associated queues invalid [3][4]. - Device/Connection Issues: A hardware disconnect, PoE/IP connection loss, or device failure has occurred, leading to the closure of communication channels [3][4]. - Upstream Node Failure: A node providing data to the queue may have crashed or stopped, causing the output queue to close [3]. When this error is encountered, it generally indicates that the queue is no longer functional and should not be used further. To debug or resolve this, consider checking the following: - Verify that the pipeline is still running (e.g., check pipeline.isRunning) [3]. - Check device connectivity and power stability [3]. - Examine logs or perform testing to determine if specific nodes are failing under load [3]. - Ensure that your code is designed to handle this exception gracefully in scenarios where pipeline restarts or device reconnects might happen [3][5].
Citations:
- 1: https://docs.ros.org/en/humble/p/depthai_v3/generated/program_listing_file_include_depthai_pipeline_MessageQueue.hpp.html
- 2: https://docs.ros.org/en/kilted/p/depthai/generated/program_listing_file_include_depthai_pipeline_MessageQueue.hpp.html
- 3: https://discuss.luxonis.com/d/6761-apriltag-queue-closed-unexpectedly-messagequeue-was-closed-on-oak4s
- 4: https://discuss.luxonis.com/d/6765-messagequeue-was-closed-error
- 5: https://docs.ros.org/en/api/depthai/html/classdai_1_1DataInputQueue.html
Guard trySend() against closed queues too Node::Output::trySend() still calls MessageQueue::trySend() for every connected input, and that API throws when the queue is closed. Add the same isClosed() check here so non-blocking sends don’t fail as soon as one input closes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/pipeline/Node.cpp` around lines 247 - 251, `Node::Output::trySend()`
still attempts `MessageQueue::trySend()` on every connected input, which can
throw if a queue is already closed. Update the send loop in
`Node::Output::trySend()` to mirror the `sendToInputs` logic by checking
`isClosed()` on each `messageQueue` before calling `trySend()`, so closed inputs
are skipped instead of causing non-blocking sends to fail.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Guard trySend() the same way as send()
MessageQueue::trySend() still follows the closed-queue path here, so this loop should skip closed inputs before calling it to avoid inconsistent behavior with the guarded send() path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/pipeline/Node.cpp` around lines 247 - 251, The `sendToInputs` lambda in
`Node::send()` still needs the same closed-queue guard for
`MessageQueue::trySend()` as it already has for `send()`. Update the loop over
`connectedInputs` so it checks `messageQueue->isClosed()` before invoking either
send path, and ensure any `trySend()` call is skipped for closed queues to keep
behavior consistent with the guarded `send()` logic.
Purpose
This fixes topics not sending data when removing and re-adding them.
Specification
Removing a topic closes the created queue but does not unlink it from the output (this object is not available at that point). The output then tried to send to the closed queue (before sending to the new queue) which blocked the output.
Dependencies & Potential Impact
None / not applicable
Deployment Plan
None / not applicable
Testing & Validation
None / not applicable
Summary by CodeRabbit