Python finalize implementation (#2159)

rasapala · web-flow · commit 1bf04dbec1bc · 2023-11-28T17:43:38.000+01:00
* Python finalize implementation
diff --git a/src/mediapipe_internal/pythonnoderesource.cpp b/src/mediapipe_internal/pythonnoderesource.cpp
@@ -36,16 +36,40 @@ namespace ovms {
 PythonNodeResource::PythonNodeResource(PythonBackend* pythonBackend) {
     this->nodeResourceObject = nullptr;
     this->pythonBackend = pythonBackend;
+    this->pythonNodeFilePath = "";
 }
 
-Status PythonNodeResource::createPythonNodeResource(std::shared_ptr<PythonNodeResource>& nodeResource, const google::protobuf::Any& nodeOptions, PythonBackend* pythonBackend) {
-    mediapipe::PythonExecutorCalculatorOptions options;
-    nodeOptions.UnpackTo(&options);
-    if (!std::filesystem::exists(options.handler_path())) {
-        SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Python node file: {} does not exist. ", options.handler_path());
+void PythonNodeResource::finalize() {
+    if (this->nodeResourceObject) {
+        py::gil_scoped_acquire acquire;
+        try {
+            if (!py::hasattr(*nodeResourceObject.get(), "finalize")) {
+                SPDLOG_DEBUG("Python node resource does not have a finalize method. Python node path {} ", this->pythonNodeFilePath);
+                return;
+            }
+
+            nodeResourceObject.get()->attr("finalize")();
+        } catch (const pybind11::error_already_set& e) {
+            SPDLOG_ERROR("Failed to process python node finalize method. {}  Python node path {} ", e.what(), this->pythonNodeFilePath);
+            return;
+        } catch (...) {
+            SPDLOG_ERROR("Failed to process python node finalize method. Python node path {} ", this->pythonNodeFilePath);
+            return;
+        }
+    } else {
+        SPDLOG_ERROR("nodeResourceObject is not initialized. Python node path {} ", this->pythonNodeFilePath);
+        throw std::exception();
+    }
+}
+
+Status PythonNodeResource::createPythonNodeResource(std::shared_ptr<PythonNodeResource>& nodeResource, const google::protobuf::Any& nodeConfig, PythonBackend* pythonBackend) {
+    mediapipe::PythonExecutorCalculatorOptions nodeOptions;
+    nodeConfig.UnpackTo(&nodeOptions);
+    if (!std::filesystem::exists(nodeOptions.handler_path())) {
+        SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Python node file: {} does not exist. ", nodeOptions.handler_path());
         return StatusCode::PYTHON_NODE_FILE_DOES_NOT_EXIST;
     }
-    auto fsHandlerPath = std::filesystem::path(options.handler_path());
+    auto fsHandlerPath = std::filesystem::path(nodeOptions.handler_path());
     fsHandlerPath.replace_extension();
 
     std::string parentPath = fsHandlerPath.parent_path();
@@ -62,17 +86,18 @@ Status PythonNodeResource::createPythonNodeResource(std::shared_ptr<PythonNodeRe
         py::bool_ success = pythonModel.attr("initialize")(kwargsParam);
 
         if (!success) {
-            SPDLOG_ERROR("Python node initialize script call returned false for: {}", options.handler_path());
+            SPDLOG_ERROR("Python node initialize script call returned false for: {}", nodeOptions.handler_path());
             return StatusCode::PYTHON_NODE_FILE_STATE_INITIALIZATION_FAILED;
         }
 
         nodeResource = std::make_shared<PythonNodeResource>(pythonBackend);
         nodeResource->nodeResourceObject = std::make_unique<py::object>(pythonModel);
+        nodeResource->pythonNodeFilePath = nodeOptions.handler_path();
     } catch (const pybind11::error_already_set& e) {
-        SPDLOG_ERROR("Failed to process python node file {} : {}", options.handler_path(), e.what());
+        SPDLOG_ERROR("Failed to process python node file {} : {}", nodeOptions.handler_path(), e.what());
         return StatusCode::PYTHON_NODE_FILE_STATE_INITIALIZATION_FAILED;
     } catch (...) {
-        SPDLOG_ERROR("Failed to process python node file {}", options.handler_path());
+        SPDLOG_ERROR("Failed to process python node file {}", nodeOptions.handler_path());
         return StatusCode::PYTHON_NODE_FILE_STATE_INITIALIZATION_FAILED;
     }
 
@@ -81,6 +106,7 @@ Status PythonNodeResource::createPythonNodeResource(std::shared_ptr<PythonNodeRe
 
 PythonNodeResource::~PythonNodeResource() {
     SPDLOG_DEBUG("Calling Python node resource destructor");
+    this->finalize();
     py::gil_scoped_acquire acquire;
     this->nodeResourceObject.reset();
 }
diff --git a/src/mediapipe_internal/pythonnoderesource.hpp b/src/mediapipe_internal/pythonnoderesource.hpp
@@ -38,10 +38,12 @@ struct PythonNodeResource {
 #if (PYTHON_DISABLE == 0)
     std::unique_ptr<py::object> nodeResourceObject;
     PythonBackend* pythonBackend;
+    std::string pythonNodeFilePath;
 
     PythonNodeResource(PythonBackend* pythonBackend);
     ~PythonNodeResource();
     static Status createPythonNodeResource(std::shared_ptr<PythonNodeResource>& nodeResource, const google::protobuf::Any& nodeOptions, PythonBackend* pythonBackend);
+    void finalize();
 #endif
 };
 
diff --git a/src/test/mediapipe/python/scripts/bad_finalize_exception.py b/src/test/mediapipe/python/scripts/bad_finalize_exception.py
@@ -0,0 +1,25 @@
+#*****************************************************************************
+# Copyright 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#*****************************************************************************
+
+class OvmsPythonModel:
+    def initialize(self, kwargs: dict):
+        return True
+
+    def execute(self, inputs: list, kwargs: dict) -> list:
+        return None
+    
+    def finalize(self):
+        return NotExistingException()
diff --git a/src/test/mediapipe/python/scripts/good_finalize_missing_definition.py b/src/test/mediapipe/python/scripts/good_finalize_missing_definition.py
@@ -0,0 +1,22 @@
+#*****************************************************************************
+# Copyright 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#*****************************************************************************
+
+class OvmsPythonModel:
+    def initialize(self, kwargs: dict):
+        return True
+
+    def execute(self, inputs: list, kwargs: dict) -> list:
+        return None
diff --git a/src/test/mediapipe/python/scripts/good_finalize_pass.py b/src/test/mediapipe/python/scripts/good_finalize_pass.py
@@ -0,0 +1,25 @@
+#*****************************************************************************
+# Copyright 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#*****************************************************************************
+
+class OvmsPythonModel:
+    def initialize(self, kwargs: dict):
+        return True
+
+    def execute(self, inputs: list, kwargs: dict) -> list:
+        return None
+
+    def finalize(self):
+        pass
diff --git a/src/test/mediapipe/python/scripts/good_finalize_remove_file.py b/src/test/mediapipe/python/scripts/good_finalize_remove_file.py
@@ -0,0 +1,30 @@
+#*****************************************************************************
+# Copyright 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#*****************************************************************************
+
+import os
+
+class OvmsPythonModel:
+    def initialize(self, kwargs: dict):
+        with open("/tmp/pythonNodeTestRemoveFile.txt", 'wt') as file:
+            file.write("1111")
+        return True
+
+    def execute(self, inputs: list, kwargs: dict) -> list:
+        return None
+
+    def finalize(self):
+        os.remove("/tmp/pythonNodeTestRemoveFile.txt")
+        return True
diff --git a/src/test/mediapipe/python/two_python_outputs_graph.pbtxt b/src/test/mediapipe/python/two_python_outputs_graph.pbtxt
@@ -40,4 +40,16 @@ node {
       handler_path: "/ovms/src/test/mediapipe/python/scripts/symmetric_increment.py"
     }
   }
+}
+node {
+  name: "pythonNode3"
+  calculator: "PythonExecutorCalculator"
+  input_side_packet: "PYTHON_NODE_RESOURCES:py"
+  input_stream: "INPUT:in"
+  output_stream: "OUTPUT:out3"
+  node_options: {
+    [type.googleapis.com / mediapipe.PythonExecutorCalculatorOptions]: {
+      handler_path: "/ovms/src/test/mediapipe/python/scripts/good_finalize_remove_file.py"
+    }
+  }
 }
diff --git a/src/test/pythonnode_test.cpp b/src/test/pythonnode_test.cpp
@@ -72,7 +72,7 @@ It's launching along with the server and even most tests will not use the server
 std::unique_ptr<std::thread> serverThread;
 
 class PythonFlowTest : public ::testing::TestWithParam<std::pair<std::string, std::string>> {
-protected:
+public:
     static void SetUpTestSuite() {
         std::string configPath = "/ovms/src/test/mediapipe/python/mediapipe_add_python_node.json";
         ovms::Server::instance().setShutdownRequest(0);
@@ -97,6 +97,8 @@ class PythonFlowTest : public ::testing::TestWithParam<std::pair<std::string, st
         ovms::Server::instance().setShutdownRequest(1);
         serverThread->join();
         ovms::Server::instance().setShutdownRequest(0);
+        std::string path = std::string("/tmp/pythonNodeTestRemoveFile.txt");
+        ASSERT_TRUE(!std::filesystem::exists(path));
     }
 };
 
@@ -110,6 +112,16 @@ TEST_F(PythonFlowTest, InitializationPass) {
     EXPECT_TRUE(graphDefinition->getStatus().isAvailable());
 }
 
+TEST_F(PythonFlowTest, FinalizationPass) {
+    ModelManager* manager;
+    std::string path = std::string("/tmp/pythonNodeTestRemoveFile.txt");
+    manager = &(dynamic_cast<const ovms::ServableManagerModule*>(ovms::Server::instance().getModule(SERVABLE_MANAGER_MODULE_NAME))->getServableManager());
+    auto graphDefinition = manager->getMediapipeFactory().findDefinitionByName("mediapipePythonBackend");
+    ASSERT_NE(graphDefinition, nullptr);
+    EXPECT_TRUE(graphDefinition->getStatus().isAvailable());
+    ASSERT_TRUE(std::filesystem::exists(path));
+}
+
 class DummyMediapipeGraphDefinition : public MediapipeGraphDefinition {
 public:
     std::string inputConfig;
@@ -844,3 +856,110 @@ TEST_F(PythonFlowTest, PythonCalculatorTestSingleInSingleOutMultiRunWithErrors)
     - bad input stream element (py::object that is not pyovms.Tensor)
     - bad output stream element (py::object that is not pyovms.Tensor)
 */
+
+TEST_F(PythonFlowTest, FinalizePassTest) {
+    const std::string pbTxt{R"(
+    input_stream: "in"
+    output_stream: "out"
+        node {
+            name: "pythonNode2"
+            calculator: "PythonBackendCalculator"
+            input_side_packet: "PYOBJECT:pyobject"
+            input_stream: "in"
+            output_stream: "out2"
+            node_options: {
+                [type.googleapis.com / mediapipe.PythonExecutorCalculatorOptions]: {
+                    handler_path: "/ovms/src/test/mediapipe/python/scripts/good_finalize_pass.py"
+                }
+            }
+        }
+    )"};
+    ::mediapipe::CalculatorGraphConfig config;
+    ASSERT_TRUE(::google::protobuf::TextFormat::ParseFromString(pbTxt, &config));
+
+    std::shared_ptr<PythonNodeResource> nodeResource = nullptr;
+    ASSERT_EQ(PythonNodeResource::createPythonNodeResource(nodeResource, config.node(0).node_options(0), getPythonBackend()), StatusCode::OK);
+    nodeResource->finalize();
+}
+
+TEST_F(PythonFlowTest, FinalizeMissingPassTest) {
+    const std::string pbTxt{R"(
+    input_stream: "in"
+    output_stream: "out"
+        node {
+            name: "pythonNode2"
+            calculator: "PythonBackendCalculator"
+            input_side_packet: "PYOBJECT:pyobject"
+            input_stream: "in"
+            output_stream: "out2"
+            node_options: {
+                [type.googleapis.com / mediapipe.PythonExecutorCalculatorOptions]: {
+                    handler_path: "/ovms/src/test/mediapipe/python/scripts/good_finalize_pass.py"
+                }
+            }
+        }
+    )"};
+    ::mediapipe::CalculatorGraphConfig config;
+    ASSERT_TRUE(::google::protobuf::TextFormat::ParseFromString(pbTxt, &config));
+
+    std::shared_ptr<PythonNodeResource> nodeResource = nullptr;
+    ASSERT_EQ(PythonNodeResource::createPythonNodeResource(nodeResource, config.node(0).node_options(0), getPythonBackend()), StatusCode::OK);
+    nodeResource->finalize();
+}
+
+TEST_F(PythonFlowTest, FinalizeDestructorRemoveFileTest) {
+    const std::string pbTxt{R"(
+    input_stream: "in"
+    output_stream: "out"
+        node {
+            name: "pythonNode2"
+            calculator: "PythonBackendCalculator"
+            input_side_packet: "PYOBJECT:pyobject"
+            input_stream: "in"
+            output_stream: "out2"
+            node_options: {
+                [type.googleapis.com / mediapipe.PythonExecutorCalculatorOptions]: {
+                    handler_path: "/ovms/src/test/mediapipe/python/scripts/good_finalize_remove_file.py"
+                }
+            }
+        }
+    )"};
+    ::mediapipe::CalculatorGraphConfig config;
+    ASSERT_TRUE(::google::protobuf::TextFormat::ParseFromString(pbTxt, &config));
+
+    std::string path = std::string("/tmp/pythonNodeTestRemoveFile.txt");
+    {
+        std::shared_ptr<PythonNodeResource> nodeResouce = nullptr;
+        ASSERT_EQ(PythonNodeResource::createPythonNodeResource(nodeResouce, config.node(0).node_options(0), getPythonBackend()), StatusCode::OK);
+
+        ASSERT_TRUE(std::filesystem::exists(path));
+        // nodeResource destructor calls finalize and removes the file
+    }
+
+    ASSERT_TRUE(!std::filesystem::exists(path));
+}
+
+TEST_F(PythonFlowTest, FinalizeException) {
+    const std::string pbTxt{R"(
+    input_stream: "in"
+    output_stream: "out"
+        node {
+            name: "pythonNode2"
+            calculator: "PythonBackendCalculator"
+            input_side_packet: "PYOBJECT:pyobject"
+            input_stream: "in"
+            output_stream: "out2"
+            node_options: {
+                [type.googleapis.com / mediapipe.PythonExecutorCalculatorOptions]: {
+                    handler_path: "/ovms/src/test/mediapipe/python/scripts/bad_finalize_exception.py"
+                }
+            }
+        }
+    )"};
+    ::mediapipe::CalculatorGraphConfig config;
+    ASSERT_TRUE(::google::protobuf::TextFormat::ParseFromString(pbTxt, &config));
+
+    std::shared_ptr<PythonNodeResource> nodeResource = nullptr;
+    ASSERT_EQ(PythonNodeResource::createPythonNodeResource(nodeResource, config.node(0).node_options(0), getPythonBackend()), StatusCode::OK);
+    nodeResource->finalize();
+}

Original file line number	Diff line number	Diff line change
`@@ -40,4 +40,16 @@ node {`
`40`	`40`	`handler_path: "/ovms/src/test/mediapipe/python/scripts/symmetric_increment.py"`
`41`	`41`	`}`
`42`	`42`	`}`
	`43`	`+}`
	`44`	`+node {`
	`45`	`+ name: "pythonNode3"`
	`46`	`+ calculator: "PythonExecutorCalculator"`
	`47`	`+ input_side_packet: "PYTHON_NODE_RESOURCES:py"`
	`48`	`+ input_stream: "INPUT:in"`
	`49`	`+ output_stream: "OUTPUT:out3"`
	`50`	`+ node_options: {`
	`51`	`+ [type.googleapis.com / mediapipe.PythonExecutorCalculatorOptions]: {`
	`52`	`+ handler_path: "/ovms/src/test/mediapipe/python/scripts/good_finalize_remove_file.py"`
	`53`	`+ }`
	`54`	`+ }`
`43`	`55`	`}`