Pipeline debugging #1470

asahtik · 2025-09-24T07:11:37Z

Purpose

Implements pipeline events which should simplify debugging stuck pipelines and help profile the pipeline.

Specification

None / not applicable

Dependencies & Potential Impact

None / not applicable

Deployment Plan

None / not applicable

Testing & Validation

None / not applicable

…e/pipeline_debugging

asahtik · 2025-09-24T07:15:28Z

bindings/python/CMakeLists.txt

    src/pipeline/node/UVCBindings.cpp
    src/pipeline/node/ToFBindings.cpp
    src/pipeline/node/PointCloudBindings.cpp
-    src/pipeline/node/SyncBindings.cpp


SyncBindings were duplicated (line 96)

…e/pipeline_debugging

moratom

Will play around more with it, so this just an initial review;

The general design looks good
The example seems to produce strange results (shared locally)
I think we should create a design document to doucment this as it's quite involved PipelineEvent agregations both in device context and in host context, the merger between the two, ...
Maybe good to add an example of how to subscribe to events for a particular node

moratom · 2025-11-10T12:28:58Z

include/depthai/pipeline/NodeObjInfo.hpp


    std::string name;
    std::string alias;
+    std::string device;


Will this be the deviceId?

I think we should add bool deviceNode as well, to be explicit.

moratom · 2025-11-10T12:42:12Z

src/pipeline/node/host/Display.cpp

+    while(mainLoop()) {
+        std::shared_ptr<dai::ImgFrame> imgFrame = nullptr;
+        {
+            auto blockEvent = this->inputBlockEvent();


What does this do?

mainLoop sends a main loop event and returns isRunning(), blockEvent sends the start event in the constructor and the end event in the destructor.

moratom · 2025-11-10T16:35:35Z

bindings/python/src/pipeline/datatype/PipelineEventBindings.cpp

+        .def_readwrite("queueSize", &PipelineEvent::queueSize, DOC(dai, PipelineEvent, queueSize))
+        .def_readwrite("interval", &PipelineEvent::interval, DOC(dai, PipelineEvent, interval))
+        .def_readwrite("type", &PipelineEvent::type, DOC(dai, PipelineEvent, type))
+        .def_readwrite("source", &PipelineEvent::source, DOC(dai, PipelineEvent, source))


Doesn't need bindings, should work "automagically"

moratom · 2025-11-10T16:35:57Z

bindings/python/src/pipeline/datatype/PipelineStateBindings.cpp

+        .def("getTimestamp", &PipelineState::Buffer::getTimestamp, DOC(dai, Buffer, getTimestamp))
+        .def("getTimestampDevice", &PipelineState::Buffer::getTimestampDevice, DOC(dai, Buffer, getTimestampDevice))
+        .def("getSequenceNum", &PipelineState::Buffer::getSequenceNum, DOC(dai, Buffer, getSequenceNum))
+        .def("setTimestamp", &PipelineState::setTimestamp, DOC(dai, Buffer, setTimestamp))
+        .def("setTimestampDevice", &PipelineState::setTimestampDevice, DOC(dai, Buffer, setTimestampDevice))
+        .def("setSequenceNum", &PipelineState::setSequenceNum, DOC(dai, Buffer, setSequenceNum));


Same as for PipelineEvent bindings

moratom · 2025-11-11T12:37:55Z

include/depthai/pipeline/Pipeline.hpp

    std::unordered_map<std::string, std::filesystem::path> recordReplayFilenames;
    bool removeRecordReplayFiles = true;
    std::string defaultDeviceId;
+    bool pipelineOnHost = true;


What does this signify? If the pipeline has any host nodes?

Pipeline debugging and record & replay sections should be skipped when the pipeline is being built on device. This flag specifies if it's building on host. I added a comment and changed the name to make it more clear.

moratom · 2025-11-11T13:01:46Z

include/depthai/pipeline/Pipeline.hpp

+    void buildDevice() {
+        impl()->pipelineOnHost = false;
+        impl()->build();
+    }


It's used on RVC4

moratom · 2025-11-11T13:02:43Z

examples/python/Misc/PipelineDebugging/get_pipeline_state.py

+                    # If pipeline debugging is disabled, this will raise an exception
+                    pipelineState = pipeline.getPipelineState().nodes().detailed()
+                    for nodeId, nodeState in pipelineState.nodeStates.items():
+                        nodeName = pipeline.getNode(nodeId).getName()
+                        print(f"\n# State for node {pipeline.getNode(nodeId).getName()} ({nodeId}):")
+                        print(f"## State: {nodeState.state}")
+                        print(f"## mainLoopTiming: {'invalid' if not nodeState.mainLoopTiming.isValid() else ''}")
+                        if(nodeState.mainLoopTiming.isValid()):
+                            print("-----")
+                            print(nodeState.mainLoopTiming)
+                            print("-----")
+                        print(f"## inputsGetTiming: {'invalid' if not nodeState.inputsGetTiming.isValid() else ''}")
+                        if(nodeState.inputsGetTiming.isValid()):
+                            print("-----")
+                            print(nodeState.inputsGetTiming)
+                            print("-----")
+                        print(f"## outputsSendTiming: {'invalid' if not nodeState.outputsSendTiming.isValid() else ''}")
+                        if(nodeState.outputsSendTiming.isValid()):
+                            print("-----")
+                            print(nodeState.outputsSendTiming)
+                            print("-----")
+                        print(f"## inputStates: {'empty' if not nodeState.inputStates else ''}")
+                        for inputName, inputState in nodeState.inputStates.items():
+                            if inputState.isValid():
+                                print(f"### {inputName}:\n-----{inputState}\n-----")
+                            else:
+                                print(f"### {inputName}: invalid")
+                        print(f"## outputStates: {'empty' if not nodeState.outputStates else ''}")


Can we print json version of the object instead? Likely easier to digest

I think this makes it more clear what data is in the pipeline state and how the user can interact with it. Also just printing the serialized json doesn't handle invalid timings so there'd be a lot more junk.

moratom · 2025-11-11T13:05:49Z

src/pipeline/Pipeline.cpp


-    if(defaultDevice) {
-        auto recordPath = std::filesystem::path(utility::getEnvAs<std::string>("DEPTHAI_RECORD", ""));
-        auto replayPath = std::filesystem::path(utility::getEnvAs<std::string>("DEPTHAI_REPLAY", ""));
+    if(pipelineOnHost) {
+        if(defaultDevice) {
+            auto recordPath = std::filesystem::path(utility::getEnvAs<std::string>("DEPTHAI_RECORD", ""));
+            auto replayPath = std::filesystem::path(utility::getEnvAs<std::string>("DEPTHAI_REPLAY", ""));

-        if(defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_2
-           || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_X
-           || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_RVC4) {
-            try {
+            if(defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_2
+               || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_X
+               || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_RVC4) {
+                try {
 #ifdef DEPTHAI_MERGED_TARGET
-                if(enableHolisticRecordReplay) {
-                    switch(recordConfig.state) {
-                        case RecordConfig::RecordReplayState::RECORD:
-                            recordPath = recordConfig.outputDir;
-                            replayPath = "";
-                            break;
-                        case RecordConfig::RecordReplayState::REPLAY:
-                            recordPath = "";
-                            replayPath = recordConfig.outputDir;
-                            break;
-                        case RecordConfig::RecordReplayState::NONE:
-                            enableHolisticRecordReplay = false;
-                            break;
+                    if(enableHolisticRecordReplay) {
+                        switch(recordConfig.state) {
+                            case RecordConfig::RecordReplayState::RECORD:
+                                recordPath = recordConfig.outputDir;
+                                replayPath = "";
+                                break;
+                            case RecordConfig::RecordReplayState::REPLAY:
+                                recordPath = "";
+                                replayPath = recordConfig.outputDir;
+                                break;
+                            case RecordConfig::RecordReplayState::NONE:
+                                enableHolisticRecordReplay = false;
+                                break;
+                        }
                    }
-                }

-                defaultDeviceId = defaultDevice->getDeviceId();
-
-                if(!recordPath.empty() && !replayPath.empty()) {
-                    Logging::getInstance().logger.warn("Both DEPTHAI_RECORD and DEPTHAI_REPLAY are set. Record and replay disabled.");
-                } else if(!recordPath.empty()) {
-                    if(enableHolisticRecordReplay || utility::checkRecordConfig(recordPath, recordConfig)) {
-                        if(platform::checkWritePermissions(recordPath)) {
-                            if(utility::setupHolisticRecord(parent,
-                                                            defaultDeviceId,
-                                                            recordConfig,
-                                                            recordReplayFilenames,
-                                                            defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_2
-                                                                || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_X)) {
-                                recordConfig.state = RecordConfig::RecordReplayState::RECORD;
-                                Logging::getInstance().logger.info("Record enabled.");
+                    defaultDeviceId = defaultDevice->getDeviceId();
+
+                    if(!recordPath.empty() && !replayPath.empty()) {
+                        Logging::getInstance().logger.warn("Both DEPTHAI_RECORD and DEPTHAI_REPLAY are set. Record and replay disabled.");
+                    } else if(!recordPath.empty()) {
+                        if(enableHolisticRecordReplay || utility::checkRecordConfig(recordPath, recordConfig)) {
+                            if(platform::checkWritePermissions(recordPath)) {
+                                if(utility::setupHolisticRecord(parent,
+                                                                defaultDeviceId,
+                                                                recordConfig,
+                                                                recordReplayFilenames,
+                                                                defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_2
+                                                                    || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_X)) {
+                                    recordConfig.state = RecordConfig::RecordReplayState::RECORD;
+                                    Logging::getInstance().logger.info("Record enabled.");
+                                } else {
+                                    Logging::getInstance().logger.warn("Could not set up holistic record. Record and replay disabled.");
+                                }
                            } else {
-                                Logging::getInstance().logger.warn("Could not set up holistic record. Record and replay disabled.");
+                                Logging::getInstance().logger.warn("DEPTHAI_RECORD path does not have write permissions. Record disabled.");
                            }
                        } else {
-                            Logging::getInstance().logger.warn("DEPTHAI_RECORD path does not have write permissions. Record disabled.");
+                            Logging::getInstance().logger.warn("Could not successfully parse DEPTHAI_RECORD. Record disabled.");
                        }
-                    } else {
-                        Logging::getInstance().logger.warn("Could not successfully parse DEPTHAI_RECORD. Record disabled.");
-                    }
-                } else if(!replayPath.empty()) {
-                    if(platform::checkPathExists(replayPath)) {
-                        if(platform::checkWritePermissions(replayPath)) {
-                            if(utility::setupHolisticReplay(parent,
-                                                            replayPath,
-                                                            defaultDeviceId,
-                                                            recordConfig,
-                                                            recordReplayFilenames,
-                                                            defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_2
-                                                                || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_X)) {
-                                recordConfig.state = RecordConfig::RecordReplayState::REPLAY;
-                                if(platform::checkPathExists(replayPath, true)) {
-                                    removeRecordReplayFiles = false;
+                    } else if(!replayPath.empty()) {
+                        if(platform::checkPathExists(replayPath)) {
+                            if(platform::checkWritePermissions(replayPath)) {
+                                if(utility::setupHolisticReplay(parent,
+                                                                replayPath,
+                                                                defaultDeviceId,
+                                                                recordConfig,
+                                                                recordReplayFilenames,
+                                                                defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_2
+                                                                    || defaultDevice->getDeviceInfo().platform == XLinkPlatform_t::X_LINK_MYRIAD_X)) {
+                                    recordConfig.state = RecordConfig::RecordReplayState::REPLAY;
+                                    if(platform::checkPathExists(replayPath, true)) {
+                                        removeRecordReplayFiles = false;
+                                    }
+                                    Logging::getInstance().logger.info("Replay enabled.");
+                                } else {
+                                    Logging::getInstance().logger.warn("Could not set up holistic replay. Record and replay disabled.");
                                }
-                                Logging::getInstance().logger.info("Replay enabled.");
                            } else {
-                                Logging::getInstance().logger.warn("Could not set up holistic replay. Record and replay disabled.");
+                                Logging::getInstance().logger.warn("DEPTHAI_REPLAY path does not have write permissions. Replay disabled.");
                            }
                        } else {
-                            Logging::getInstance().logger.warn("DEPTHAI_REPLAY path does not have write permissions. Replay disabled.");
+                            Logging::getInstance().logger.warn("DEPTHAI_REPLAY path does not exist or is invalid. Replay disabled.");
                        }
-                    } else {
-                        Logging::getInstance().logger.warn("DEPTHAI_REPLAY path does not exist or is invalid. Replay disabled.");
                    }
-                }
 #else
-                recordConfig.state = RecordConfig::RecordReplayState::NONE;
-                if(!recordPath.empty() || !replayPath.empty()) {
-                    Logging::getInstance().logger.warn("Merged target is required to use holistic record/replay.");
-                }
+                    recordConfig.state = RecordConfig::RecordReplayState::NONE;
+                    if(!recordPath.empty() || !replayPath.empty()) {
+                        Logging::getInstance().logger.warn("Merged target is required to use holistic record/replay.");
+                    }
 #endif
-            } catch(std::runtime_error& e) {
-                Logging::getInstance().logger.warn("Could not set up record / replay: {}", e.what());
+                } catch(std::runtime_error& e) {
+                    Logging::getInstance().logger.warn("Could not set up record / replay: {}", e.what());
+                }
+            } else if(enableHolisticRecordReplay || !recordPath.empty() || !replayPath.empty()) {
+                throw std::runtime_error("Holistic record/replay is only supported on RVC2 devices for now.");
            }
-        } else if(enableHolisticRecordReplay || !recordPath.empty() || !replayPath.empty()) {
-            throw std::runtime_error("Holistic record/replay is only supported on RVC2 devices for now.");
        }
    }

-    // Go through the build stages sequentially
+    // Run first build stage for all nodes
    for(const auto& node : getAllNodes()) {
        node->buildStage1();
    }

+    if(pipelineOnHost) {
+        // Create pipeline event aggregator node and link
+        enablePipelineDebugging = enablePipelineDebugging || utility::getEnvAs<bool>("DEPTHAI_PIPELINE_DEBUGGING", false);
+        if(enablePipelineDebugging) {
+            // Check if any nodes are on host or device
+            bool hasHostNodes = false;
+            bool hasDeviceNodes = false;
+            for(const auto& node : getAllNodes()) {
+                if(std::string(node->getName()) == std::string("NodeGroup") || std::string(node->getName()) == std::string("DeviceNodeGroup")) continue;
+
+                if(node->runOnHost()) {
+                    hasHostNodes = true;
+                } else {
+                    hasDeviceNodes = true;
+                }
+            }
+            std::shared_ptr<node::internal::PipelineEventAggregation> hostEventAgg = nullptr;
+            std::shared_ptr<node::internal::PipelineEventAggregation> deviceEventAgg = nullptr;
+            if(hasHostNodes) {
+                hostEventAgg = parent.create<node::internal::PipelineEventAggregation>();
+                hostEventAgg->setRunOnHost(true);
+            }
+            if(hasDeviceNodes) {
+                deviceEventAgg = parent.create<node::internal::PipelineEventAggregation>();
+                deviceEventAgg->setRunOnHost(false);
+            }
+            for(auto& node : getAllNodes()) {
+                if(std::string(node->getName()) == std::string("NodeGroup") || std::string(node->getName()) == std::string("DeviceNodeGroup")) continue;
+
+                auto threadedNode = std::dynamic_pointer_cast<ThreadedNode>(node);
+                if(threadedNode) {
+                    if(node->runOnHost() && hostEventAgg && node->id != hostEventAgg->id) {
+                        threadedNode->pipelineEventOutput.link(hostEventAgg->inputs[fmt::format("{} - {}", node->getName(), node->id)]);
+                    } else if(!node->runOnHost() && deviceEventAgg && node->id != deviceEventAgg->id) {
+                        threadedNode->pipelineEventOutput.link(deviceEventAgg->inputs[fmt::format("{} - {}", node->getName(), node->id)]);
+                    }
+                }
+            }
+            auto stateMerge = parent.create<node::PipelineStateMerge>()->build(hasDeviceNodes, hasHostNodes);
+            if(deviceEventAgg) {
+                deviceEventAgg->out.link(stateMerge->inputDevice);
+                stateMerge->outRequest.link(deviceEventAgg->request);
+            }
+            if(hostEventAgg) {
+                hostEventAgg->out.link(stateMerge->inputHost);
+                stateMerge->outRequest.link(hostEventAgg->request);
+            }
+            pipelineStateOut = stateMerge->out.createOutputQueue(1, false);
+            pipelineStateRequest = stateMerge->request.createInputQueue();
+        }
+    }
+
+    {
+        auto allNodes = getAllNodes();
+        if(std::find_if(allNodes.begin(), allNodes.end(), [](const std::shared_ptr<Node>& n) { return strcmp(n->getName(), "PipelineEventAggregation") == 0; })
+           == allNodes.end()) {
+            for(auto& node : allNodes) node->pipelineEventDispatcher->sendEvents = false;
+        }
+    }
+
+    isBuild = true;
+
+    // Go through the build stages sequentially


I think we should split this in a separate functions as the build() is already hard to follow.

…ai-core into feature/pipeline_debugging

…e/pipeline_debugging

…e/pipeline_debugging [no ci]

…ai-core into feature/pipeline_debugging

…e/pipeline_debugging

asahtik added 6 commits September 16, 2025 15:46

Added pipeline events messages and node

5110bcc

Added the pipeline event dispatcher

e165806

Add Pipeline event bindings

fbd4137

Integrate pipeline events

9876101

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

0e00bff

…e/pipeline_debugging

RVC4 FW: initial pipeline debugging impl

b2024f0

asahtik marked this pull request as draft September 24, 2025 07:12

asahtik commented Sep 24, 2025

View reviewed changes

asahtik added 22 commits September 24, 2025 14:41

Pipeline events bugfixes, pipeline state merge node

75e9386

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

a1668e7

…e/pipeline_debugging

Some pipeline debugging bugfixes

99ce452

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

10753f0

…e/pipeline_debugging

Fix pipeline events on host

914cbc5

Fix RVC4 pipeline events crash

d30b5fd

RVC4 FW: Fix pipeline events crash

3a579ac

Add more stats, queue state WIP

2e9b26d

Event instances now have independent sequence numbers

a5e93d8

Do not update stats for every event

38481f0

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

978e9b9

…e/pipeline_debugging

Fix dropped pipeline events

962d4af

RVC4 FW: Fix dropped pipeline events

b79ded6

Added queue size stats

de20f54

WIP pipeline state api

4854124

Implement service like api for pipeline state

c34cc45

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

f9835ee

…e/pipeline_debugging

Remove static assert

fcbfe78

Bugfixes

28e5b93

Collect events in a different thread

215119d

WIP: dispatcher + state improvements

e8045e4

Do not require adding events, node / input states, other improvements

36e917a

moratom reviewed Nov 11, 2025

View reviewed changes

asahtik added 5 commits November 11, 2025 14:53

Change device to deviceId & deviceNode in objinfo [no ci]

b9cda65

RVC4 FW: update core [no ci]

d32da8a

RVC2 FW: update core

8589893

PR fixes

568ad64

Merge branch 'feature/pipeline_debugging' of github.com:luxonis/depth…

e6e28b5

…ai-core into feature/pipeline_debugging

asahtik removed the testable PR is ready to be tested label Nov 12, 2025

asahtik added 4 commits November 12, 2025 08:40

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

a79afee

…e/pipeline_debugging

Remove pipeline debugging env var from tests

d0087cf

Add example for pipeline event subscription

2d0fc90

Clangformat

560d764

asahtik added the testable PR is ready to be tested label Nov 12, 2025

Macos build fix v2?

2c80e92

asahtik removed the testable PR is ready to be tested label Nov 12, 2025

asahtik added 15 commits November 12, 2025 13:49

Fix windows build of example

c4ba9a2

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

921993f

…e/pipeline_debugging [no ci]

MacOs build fix v3 [no ci]

bcbf584

RVC4 FW: merge develop [no ci]

97897e1

RVC2 FW: merge develop

201bbd3

Fix include [no ci]

c04ed85

RVC2 FW: Update core [no ci]

7637f2e

Added pipeline debugging docs

f3eeb97

Merge branch 'feature/pipeline_debugging' of github.com:luxonis/depth…

7bd26b6

…ai-core into feature/pipeline_debugging

Move section in PD docs

3a14630

Add background to PD doc schema

b71c87f

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

e5bc755

…e/pipeline_debugging

RVC2 FW: Update core

cd4b338

Merge branch 'develop' of github.com:luxonis/depthai-core into featur…

c84b861

…e/pipeline_debugging

RVC4 FW: Update core

0caa3d3

asahtik added the testable PR is ready to be tested label Nov 14, 2025

Pipeline debugging #1470

Are you sure you want to change the base?

Pipeline debugging #1470

Uh oh!

Conversation

asahtik commented Sep 24, 2025

Purpose

Specification

Dependencies & Potential Impact

Deployment Plan

Testing & Validation

Uh oh!

Choose a reason for hiding this comment

Uh oh!

moratom left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants