MP executor wrapper for streaming callbacks (#3822)

Author: mzegla

src/BUILD

@@ -477,6 +477,7 @@ ovms_cc_library(
         "capi_frontend/capi.cpp",
         "cleaner_utils.cpp",
         "cleaner_utils.hpp",
+        "copyable_object_wrapper.hpp",
         "customloaderconfig.hpp",
         "customloaders.hpp",
         "customloaders.cpp",

src/copyable_object_wrapper.hpp

@@ -0,0 +1,53 @@
+//*****************************************************************************
+// Copyright 2025 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.
+//*****************************************************************************
+#pragma once
+
+#include <memory>
+#include <utility>
+
+template <typename T>
+class UniqueObjectHolder {
+    std::unique_ptr<T> object = nullptr;
+
+public:
+    std::unique_ptr<T>& get() {
+        return object;
+    }
+
+    void reset() {
+        object.reset();
+    }
+
+    bool valid() const {
+        return object != nullptr;
+    }
+};
+
+template <typename T>
+class CopyableObjectWrapper {
+    std::shared_ptr<UniqueObjectHolder<T>> objectHolder = nullptr;
+
+public:
+    explicit CopyableObjectWrapper() :
+        objectHolder(std::make_shared<UniqueObjectHolder<T>>()) {}
+
+    explicit CopyableObjectWrapper(std::shared_ptr<UniqueObjectHolder<T>> objectHolder) :
+        objectHolder(std::move(objectHolder)) {}
+
+    std::shared_ptr<UniqueObjectHolder<T>>& getObjectHolder() {
+        return objectHolder;
+    }
+};

src/http_rest_api_handler.cpp

@@ -62,6 +62,7 @@
 #include "status.hpp"
 #include "stringutils.hpp"
 #include "timer.hpp"
+#include "copyable_object_wrapper.hpp"
 
 #if (MEDIAPIPE_DISABLE == 0)
 #include "http_frontend/http_client_connection.hpp"
@@ -703,21 +704,28 @@ Status HttpRestApiHandler::processV3(const std::string_view uri, const HttpReque
         return status;
     }
 
-    std::shared_ptr<MediapipeGraphExecutor> executor;
+    CopyableObjectWrapper<MediapipeGraphExecutor> executorWrapper;
+    auto& executor = executorWrapper.getObjectHolder()->get();
     status = this->modelManager.createPipeline(executor, modelName);
     if (!status.ok()) {
         return status;
     }
 
+    if (!executorWrapper.getObjectHolder()->valid()) {
+        SPDLOG_ERROR("Failed to acquire MediaPipe graph executor for model: {}", modelName);
+        return StatusCode::INTERNAL_ERROR;
+    }
+
     if (streamFieldVal == false) {
         ExecutionContext executionContext{ExecutionContext::Interface::REST, ExecutionContext::Method::V3Unary};
         return executor->infer(&request, &response, executionContext);
     } else {
         serverReaderWriter->OverwriteResponseHeader("Content-Type", "text/event-stream");
         serverReaderWriter->OverwriteResponseHeader("Cache-Control", "no-cache");
         serverReaderWriter->OverwriteResponseHeader("Connection", "keep-alive");
-        serverReaderWriter->PartialReplyBegin([executor = std::move(executor), serverReaderWriter, request = std::move(request)] {
+        serverReaderWriter->PartialReplyBegin([executorWrapper = executorWrapper, serverReaderWriter, request = std::move(request)]() mutable {
             ExecutionContext executionContext{ExecutionContext::Interface::REST, ExecutionContext::Method::V3Stream};
+            auto& executor = executorWrapper.getObjectHolder()->get();
             auto status = executor->inferStream(request, *serverReaderWriter, executionContext);
             if (!status.ok()) {
                 rapidjson::StringBuffer buffer;
@@ -728,6 +736,7 @@ Status HttpRestApiHandler::processV3(const std::string_view uri, const HttpReque
                 writer.EndObject();
                 serverReaderWriter->PartialReplyWithStatus(buffer.GetString(), HTTPStatusCode::BAD_REQUEST);
             }
+            executor.reset();
             serverReaderWriter->PartialReplyEnd();
         });
         return StatusCode::PARTIAL_END;

src/kfs_frontend/kfs_grpc_inference_service.cpp

@@ -303,7 +303,7 @@ Status KFSInferenceServiceImpl::ModelInferImpl(::grpc::ServerContext* context, c
         if (status == StatusCode::PIPELINE_DEFINITION_NAME_MISSING) {
             SPDLOG_DEBUG("Requested DAG: {} does not exist. Searching for mediapipe graph with that name...", request->model_name());
 #if (MEDIAPIPE_DISABLE == 0)
-            std::shared_ptr<MediapipeGraphExecutor> executor;
+            std::unique_ptr<MediapipeGraphExecutor> executor;
             status = this->modelManager.createPipeline(executor, request->model_name());
             if (!status.ok()) {
                 return status;
@@ -346,7 +346,7 @@ Status KFSInferenceServiceImpl::ModelStreamInferImpl(::grpc::ServerContext* cont
         SPDLOG_DEBUG(status.string());
         return status;
     }
-    std::shared_ptr<MediapipeGraphExecutor> executor;
+    std::unique_ptr<MediapipeGraphExecutor> executor;
     auto status = this->modelManager.createPipeline(executor, firstRequest.model_name());
     if (!status.ok()) {
         return status;

src/mediapipe_internal/mediapipefactory.cpp

@@ -104,7 +104,7 @@ Status MediapipeFactory::reloadDefinition(const std::string& name,
     return mgd->reload(manager, config);
 }
 
-Status MediapipeFactory::create(std::shared_ptr<MediapipeGraphExecutor>& pipeline,
+Status MediapipeFactory::create(std::unique_ptr<MediapipeGraphExecutor>& pipeline,
     const std::string& name,
     ModelManager& manager) const {
     std::shared_lock lock(definitionsMtx);

src/mediapipe_internal/mediapipefactory.hpp

@@ -66,7 +66,7 @@ class MediapipeFactory {
         ModelManager& manager) const;
 
 public:
-    Status create(std::shared_ptr<MediapipeGraphExecutor>& pipeline,
+    Status create(std::unique_ptr<MediapipeGraphExecutor>& pipeline,
         const std::string& name,
         ModelManager& manager) const;
 

src/mediapipe_internal/mediapipegraphdefinition.cpp

@@ -253,7 +253,7 @@ Status MediapipeGraphDefinition::createOutputsInfo() {
     return StatusCode::OK;
 }
 
-Status MediapipeGraphDefinition::create(std::shared_ptr<MediapipeGraphExecutor>& pipeline) {
+Status MediapipeGraphDefinition::create(std::unique_ptr<MediapipeGraphExecutor>& pipeline) {
     std::unique_ptr<MediapipeGraphDefinitionUnloadGuard> unloadGuard;
     Status status = waitForLoaded(unloadGuard);
     if (!status.ok()) {
@@ -262,7 +262,7 @@ Status MediapipeGraphDefinition::create(std::shared_ptr<MediapipeGraphExecutor>&
     }
     SPDLOG_DEBUG("Creating Mediapipe graph executor: {}", getName());
 
-    pipeline = std::make_shared<MediapipeGraphExecutor>(getName(), std::to_string(getVersion()),
+    pipeline = std::make_unique<MediapipeGraphExecutor>(getName(), std::to_string(getVersion()),
         this->config, this->inputTypes, this->outputTypes, this->inputNames, this->outputNames,
         this->sidePacketMaps,
         this->pythonBackend, this->reporter.get());

src/mediapipe_internal/mediapipegraphdefinition.hpp

@@ -119,7 +119,7 @@ class MediapipeGraphDefinition {
     const tensor_map_t getOutputsInfo() const;
     const MediapipeGraphConfig& getMediapipeGraphConfig() const { return this->mgconfig; }
     MediapipeServableMetricReporter& getMetricReporter() const { return *this->reporter; }
-    Status create(std::shared_ptr<MediapipeGraphExecutor>& pipeline);
+    Status create(std::unique_ptr<MediapipeGraphExecutor>& pipeline);
 
     Status reload(ModelManager& manager, const MediapipeGraphConfig& config);
     Status validate(ModelManager& manager);

src/mediapipe_internal/mediapipegraphexecutor.hpp

@@ -393,4 +393,5 @@ class MediapipeGraphExecutor {
         }
     }
 };
+
 }  // namespace ovms

src/modelmanager.cpp

@@ -1697,7 +1697,7 @@ const std::vector<std::string> ModelManager::getNamesOfAvailableModels() const {
     return names;
 }
 
-Status ModelManager::createPipeline(std::shared_ptr<MediapipeGraphExecutor>& graph,
+Status ModelManager::createPipeline(std::unique_ptr<MediapipeGraphExecutor>& graph,
     const std::string& name) {
 #if (MEDIAPIPE_DISABLE == 0)
     return this->mediapipeFactory.create(graph, name, *this);