Improve depthai-core's responsiveness upon a device crash or a node error, attempt 2

README.md

@@ -215,7 +215,7 @@ The following environment variables can be set to alter default behavior of the
 | DEPTHAI_ALLOW_FACTORY_FLASHING | Internal use only |
 | DEPTHAI_LIBUSB_ANDROID_JAVAVM | JavaVM pointer that is passed to libusb for rootless Android interaction with devices. Interpreted as decimal value of uintptr_t |
 | DEPTHAI_CRASHDUMP | Directory in which to save the crash dump. |
-| DEPTHAI_CRASHDUMP_TIMEOUT | Specifies the duration in seconds to wait for device reboot when obtaining a crash dump. Crash dump retrieval disabled if 0. |
+| DEPTHAI_CRASHDUMP_TIMEOUT | Specifies the duration in milliseconds to wait for device reboot when obtaining a crash dump. Crash dump retrieval disabled if 0. |
 | DEPTHAI_ENABLE_ANALYTICS_COLLECTION | Enables automatic analytics collection (pipeline schemas) used to improve the library |
 | DEPTHAI_DISABLE_CRASHDUMP_COLLECTION | Disables automatic crash dump collection used to improve the library |
 | DEPTHAI_HUB_API_KEY | API key for the Luxonis Hub |

bindings/python/src/DeviceBindings.cpp

@@ -25,30 +25,6 @@ PYBIND11_MAKE_OPAQUE(std::unordered_map<std::int8_t, dai::BoardConfig::UART>);
 // Searches for available devices (as Device constructor)
 // but pooling, to check for python interrupts, and releases GIL in between
 
-template <typename DEVICE, class... Args>
-static auto deviceSearchHelper(Args&&... args) {
-    bool found;
-    dai::DeviceInfo deviceInfo;
-    // releases python GIL
-    py::gil_scoped_release release;
-    std::tie(found, deviceInfo) = DEVICE::getAnyAvailableDevice(DEVICE::getDefaultSearchTime(), []() {
-        py::gil_scoped_acquire acquire;
-        if(PyErr_CheckSignals() != 0) throw py::error_already_set();
-    });
-
-    // if no devices found, then throw
-    if(!found) {
-        auto numConnected = DEVICE::getAllAvailableDevices().size();
-        if(numConnected > 0) {
-            throw std::runtime_error("No available devices (" + std::to_string(numConnected) + " connected, but in use)");
-        } else {
-            throw std::runtime_error("No available devices");
-        }
-    }
-
-    return deviceInfo;
-}
-
 // static std::vector<std::string> deviceGetQueueEventsHelper(dai::Device& d, const std::vector<std::string>& queueNames, std::size_t maxNumEvents,
 // std::chrono::microseconds timeout){
 //     using namespace std::chrono;

bindings/python/src/DeviceBindings.hpp

@@ -1,8 +1,33 @@
 #pragma once
 
 // pybind
+#include "depthai/device/Device.hpp"
 #include "pybind11_common.hpp"
 
+template <typename DEVICE, class... Args>
+static auto deviceSearchHelper(Args&&... args) {
+    bool found;
+    dai::DeviceInfo deviceInfo;
+    // releases python GIL
+    py::gil_scoped_release release;
+    std::tie(found, deviceInfo) = DEVICE::getAnyAvailableDevice(DEVICE::getDefaultSearchTime(), []() {
+        py::gil_scoped_acquire acquire;
+        if(PyErr_CheckSignals() != 0) throw py::error_already_set();
+    });
+
+    // if no devices found, then throw
+    if(!found) {
+        auto numConnected = DEVICE::getAllAvailableDevices().size();
+        if(numConnected > 0) {
+            throw std::runtime_error("No available devices (" + std::to_string(numConnected) + " connected, but in use)");
+        } else {
+            throw std::runtime_error("No available devices");
+        }
+    }
+
+    return deviceInfo;
+}
+
 struct DeviceBindings {
     static void bind(pybind11::module& m, void* pCallstack);
 };

bindings/python/src/pipeline/PipelineBindings.cpp

@@ -4,6 +4,7 @@
 #include <pybind11/attr.h>
 #include <pybind11/gil.h>
 
+#include "DeviceBindings.hpp"
 #include "node/NodeBindings.hpp"
 
 // depthai
@@ -63,7 +64,7 @@ void PipelineBindings::bind(pybind11::module& m, void* pCallstack) {
     py::class_<GlobalProperties> globalProperties(m, "GlobalProperties", DOC(dai, GlobalProperties));
     py::class_<RecordConfig> recordConfig(m, "RecordConfig", DOC(dai, RecordConfig));
     py::class_<RecordConfig::VideoEncoding> recordVideoConfig(recordConfig, "VideoEncoding", DOC(dai, RecordConfig, VideoEncoding));
-    py::class_<Pipeline> pipeline(m, "Pipeline", DOC(dai, Pipeline, 2));
+    py::class_<Pipeline, std::shared_ptr<Pipeline>> pipeline(m, "Pipeline", DOC(dai, Pipeline, 2));
 
     ///////////////////////////////////////////////////////////////////////
     ///////////////////////////////////////////////////////////////////////
@@ -102,20 +103,70 @@ void PipelineBindings::bind(pybind11::module& m, void* pCallstack) {
         .def_readwrite("compressionLevel", &RecordConfig::compressionLevel, DOC(dai, RecordConfig, compressionLevel));
 
     // bind pipeline
-    pipeline.def(py::init<bool>(), py::arg("createImplicitDevice") = true, DOC(dai, Pipeline, Pipeline))
+    pipeline
+        .def(py::init([](bool createImplicitDevice) {
+                 if(createImplicitDevice) {
+                     auto deviceInfo = deviceSearchHelper<Device>();
+                     auto device = std::make_shared<Device>(deviceInfo);
+                     auto pipeline = std::make_shared<Pipeline>(device);
+                     return pipeline;
+                 }
+
+                 assert(createImplicitDevice == false);
+                 return std::make_shared<Pipeline>(createImplicitDevice);
+             }),
+             py::arg("createImplicitDevice") = true,
+             DOC(dai, Pipeline, Pipeline))
         .def(py::init<std::shared_ptr<Device>>(), py::arg("defaultDevice"), DOC(dai, Pipeline, Pipeline))
         // Python only methods
         .def("__enter__",
-             [](Pipeline& p) -> Pipeline& {
-                 setImplicitPipeline(&p);
+             [](std::shared_ptr<Pipeline> p) -> std::shared_ptr<Pipeline> {
+                 setImplicitPipeline(p.get());
                  return p;
              })
         .def("__exit__",
-             [](Pipeline& d, py::object type, py::object value, py::object traceback) {
-                 py::gil_scoped_release release;
+             [](std::shared_ptr<Pipeline> pipeline, py::object type, py::object value, py::object traceback) {
                  delImplicitPipeline();
-                 d.stop();
-                 d.wait();
+
+                 // Check how we got here (whether through an exception or the end of the with block was simply reached)
+                 bool isException = !type.is_none();
+                 bool isKeyboardInterrupt = type.is(py::handle(PyExc_KeyboardInterrupt));
+
+                 // Spawn a thread to clean up the pipeline
+                 std::atomic<bool> threadRunning{true};
+                 auto cleanupThread = std::thread([pipeline, &threadRunning, isException]() {
+                     // In case of an exception, we want the device to close as quickly as possible
+                     if(isException) {
+                         pipeline->closeQuick();
+                     }
+                     pipeline->stop();
+                     pipeline->wait();
+                     threadRunning = false;
+                 });
+
+                 // While the thread above is cleaning up, check for interrupts
+                 // https://docs.python.org/3/c-api/exceptions.html#c.PyErr_CheckSignals
+                 py::gil_scoped_release release;
+                 while(threadRunning) {
+                     std::this_thread::sleep_for(std::chrono::milliseconds(200));
+                     py::gil_scoped_acquire acquire;
+                     if(PyErr_CheckSignals() != 0) {
+                         // If we are handling a KeyboardInterrupt and another KeyboardInterrupt is thrown, continue
+                         // Otherwise, rethrow the exception
+                         if(PyErr_ExceptionMatches(PyExc_KeyboardInterrupt) && isKeyboardInterrupt) {
+                             try {
+                                 throw py::error_already_set();
+                             } catch(const py::error_already_set& e) {
+                                 // Expect this to be thrown, but do nothing
+                             }
+                             continue;
+                         } else {
+                             throw py::error_already_set();
+                         }
+                     }
+                 }
+
+                 cleanupThread.join();
              })
         //.def(py::init<const Pipeline&>())
         .def("getDefaultDevice", &Pipeline::getDefaultDevice, DOC(dai, Pipeline, getDefaultDevice))
@@ -211,7 +262,7 @@ void PipelineBindings::bind(pybind11::module& m, void* pCallstack) {
                 return node;
             },
             py::keep_alive<1, 0>())
-        .def("start", &Pipeline::start)
+        .def("start", &Pipeline::start, py::call_guard<py::gil_scoped_release>(), DOC(dai, Pipeline, start))
         .def("wait",
              [](Pipeline& p) {
                  py::gil_scoped_release release;

include/depthai/device/DeviceBase.hpp

@@ -754,6 +754,11 @@ class DeviceBase {
      */
     void close();
 
+    /**
+     * @brief Close the device as quickly as possible without performing any cleanup tasks or other checks. Users should preferablly use close() instead.
+     */
+    void closeQuick();
+
     /**
      * Is the device already closed (or disconnected)
      *
@@ -816,6 +821,13 @@ class DeviceBase {
      */
     virtual void closeImpl();
 
+    /**
+     * @brief Whether the device should be closed as quickly as possible. Long-running methods ought to check this flag every so often and return if true, all
+     * long-running operations should be aborted.
+     * @note It is a responsibility of the developer to ensure that this flag is checked frequently.
+     */
+    std::atomic<bool> shouldCloseQuickly{false};
+
    protected:
     // protected functions
     void init();

include/depthai/pipeline/Pipeline.hpp

@@ -228,8 +228,17 @@ class PipelineImpl : public std::enable_shared_from_this<PipelineImpl> {
     void resetConnections();
     void disconnectXLinkHosts();
 
+    inline void closeQuick() {
+        shouldCloseQuickly = true;
+        if(!isHostOnly()) {
+            defaultDevice->closeQuick();
+        }
+    }
+
    private:
     // Resource
+    std::atomic<bool> shouldCloseQuickly{false};
+
     std::vector<uint8_t> loadResource(fs::path uri);
     std::vector<uint8_t> loadResourceCwd(fs::path uri, fs::path cwd, bool moveAsset = false);
 };
@@ -503,6 +512,10 @@ class Pipeline {
         impl()->addTask(std::move(task));
     }
 
+    void closeQuick() {
+        impl()->closeQuick();
+    }
+
     /// Record and Replay
     void enableHolisticRecord(const RecordConfig& config);
     void enableHolisticReplay(const std::string& pathToRecording);