Skip to content

Send only to opened queues #1871

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
asahtik wants to merge 1 commit into
base: develop
Choose a base branch
from
Draft

Send only to opened queues #1871

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/pipeline/Node.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,7 +246,7 @@ void Node::Output::send(const std::shared_ptr<ADatatype>& msg) {
// }
auto sendToInputs = [this, &msg]() {
for(auto& messageQueue : connectedInputs) {
messageQueue->send(msg);
if(!messageQueue->isClosed()) messageQueue->send(msg);
}
Comment on lines 248 to 250

@aljazkonec1 aljazkonec1 Jul 6, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@asahtik asahtik Jul 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

};
Comment on lines 247 to 251

@coderabbitai coderabbitai Bot Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 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:

🏁 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'
done

Repository: 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:

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.

if(pipelineEventDispatcher && pipelineEventDispatcher->sendEvents) {
Expand Down
Loading