Force removal of Pyo3 Object to avoid memory leak

apis/python/node/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
+version = "0.1.2"
 name = "dora-node-api-python"
-version.workspace = true
 edition = "2021"
 license = "Apache-2.0"
 

binaries/runtime/src/operator/python.rs

@@ -10,12 +10,12 @@ use dora_message::uhlc;
 use dora_node_api::communication::Publisher;
 use dora_operator_api_python::metadata_to_pydict;
 use dora_operator_api_types::DoraStatus;
-use eyre::{bail, eyre, Context};
+use eyre::{bail, eyre, Context, Result};
 use pyo3::{
-    pyclass,
+    ffi, pyclass,
     types::IntoPyDict,
     types::{PyBytes, PyDict},
-    Py, Python,
+    AsPyPointer, Py, Python,
 };
 use std::{
     borrow::Cow,
@@ -137,21 +137,27 @@ pub fn spawn(
             };
             input.metadata.parameters.open_telemetry_context = Cow::Owned(string_cx);
 
-            let status_enum = Python::with_gil(|py| {
+            let status = Python::with_gil(|py| -> Result<i32> {
+                let pool = unsafe { py.new_pool() };
+                let py = pool.python();
                 let input_dict = PyDict::new(py);
+                let bytes = PyBytes::new(py, &input.data());
 
                 input_dict.set_item("id", input.id.as_str())?;
-                input_dict.set_item("data", PyBytes::new(py, &input.data()))?;
+                input_dict.set_item("data", bytes)?;
                 input_dict.set_item("metadata", metadata_to_pydict(input.metadata(), py))?;
 
-                operator
+                let status_enum = operator
                     .call_method1(py, "on_input", (input_dict, send_output.clone()))
-                    .map_err(traceback)
+                    .map_err(traceback)?;
+
+                let status_val = status_enum
+                    .getattr(py, "value")
+                    .wrap_err("on_input must have enum return value")?;
+                status_val
+                    .extract(py)
+                    .wrap_err("on_input has invalid return value")
             })?;
-            let status_val = Python::with_gil(|py| status_enum.getattr(py, "value"))
-                .wrap_err("on_input must have enum return value")?;
-            let status: i32 = Python::with_gil(|py| status_val.extract(py))
-                .wrap_err("on_input has invalid return value")?;
             match status {
                 s if s == DoraStatus::Continue as i32 => {} // ok
                 s if s == DoraStatus::Stop as i32 => break StopReason::ExplicitStop,

examples/python-dataflow/plot.py

@@ -81,6 +81,3 @@ def on_input(
                 return DoraStatus.STOP
 
         return DoraStatus.CONTINUE
-
-    def __del__(self):
-        cv2.destroyAllWindows()

examples/python-dataflow/requirements.txt

@@ -2,17 +2,20 @@
 # Usage: pip install -r requirements.txt
 
 # Base ----------------------------------------
+gitpython
+ipython  # interactive notebook
 matplotlib>=3.2.2
 numpy>=1.18.5
 opencv-python>=4.1.1
 Pillow>=7.1.2
+psutil  # system resources
 PyYAML>=5.3.1
 requests>=2.23.0
 scipy>=1.4.1
-torch>=1.7.0
+thop>=0.1.1  # FLOPs computation
+torch>=1.7.0  # see https://pytorch.org/get-started/locally (recommended)
 torchvision>=0.8.1
 tqdm>=4.64.0
-protobuf<=3.20.1  # https://github.com/ultralytics/yolov5/issues/8012
 
 # Logging -------------------------------------
 tensorboard>=2.4.1
@@ -35,9 +38,6 @@ seaborn>=0.11.0
 # openvino-dev  # OpenVINO export
 
 # Extras --------------------------------------
-ipython  # interactive notebook
-psutil  # system utilization
-thop>=0.1.1  # FLOPs computation
 # albumentations>=1.0.3
 # pycocotools>=2.0  # COCO mAP
 # roboflow