Refactor model management loading to use queue

src/BUILD

@@ -611,6 +611,7 @@ ovms_cc_library(
         "//src/graph_export:graph_export",
         "//src/metrics:libovms_metric_provider",
         "//src/metrics:libovmsmetrics",
+        "//src/model_management:servable_loading_queue",
         "@com_github_tencent_rapidjson//:rapidjson",
         "//src/port:rapidjson_stringbuffer",
         "//src/port:rapidjson_writer",
@@ -2258,6 +2259,7 @@ cc_test(
             ":test_test_models_configs",
             ":test_cmd_exec",
             ":test_modelinstance_test",
+            ":servable_loading_queue_test",
             ] + select({
                 "//conditions:default": [
                     ":openvino_remote_tensors_tests",
@@ -2370,6 +2372,15 @@ cc_library(
     linkopts = COMMON_STATIC_LIBS_LINKOPTS,
 )
 
+ovms_cc_test_library(
+    name = "servable_loading_queue_test",
+    srcs = ["test/servable_loading_queue_test.cpp"],
+    deps = [
+        "//src/model_management:servable_loading_queue",
+        "@com_google_googletest//:gtest",
+    ],
+)
+
 cc_library(
     name = "test_utils",
     linkstatic = 1,

src/grpc_utils.cpp

@@ -102,6 +102,7 @@ const grpc::Status grpc(const Status& status) {
         {StatusCode::MODEL_VERSION_NOT_LOADED_YET, grpc::StatusCode::UNAVAILABLE},
         {StatusCode::PIPELINE_DEFINITION_NOT_LOADED_YET, grpc::StatusCode::UNAVAILABLE},
         {StatusCode::MEDIAPIPE_DEFINITION_NOT_LOADED_YET, grpc::StatusCode::UNAVAILABLE},
+        {StatusCode::SERVER_SHUTTING_DOWN, grpc::StatusCode::UNAVAILABLE},
         // UNKNOWN
     };
     auto it = grpcStatusMap.find(status.getCode());

src/http_server.cpp

@@ -82,6 +82,7 @@ static const ovms::HTTPStatusCode http(const ovms::Status& status) {
         {StatusCode::NO_MODEL_VERSION_AVAILABLE, ovms::HTTPStatusCode::ERROR},
         {StatusCode::MODEL_NOT_LOADED, ovms::HTTPStatusCode::ERROR},
         {StatusCode::SERVER_NOT_READY, ovms::HTTPStatusCode::SERVICE_UNAV},
+        {StatusCode::SERVER_SHUTTING_DOWN, ovms::HTTPStatusCode::SERVICE_UNAV},
         {StatusCode::JSON_INVALID, ovms::HTTPStatusCode::PRECOND_FAILED},
         {StatusCode::MODELINSTANCE_NOT_FOUND, ovms::HTTPStatusCode::ERROR},
         {StatusCode::SHAPE_WRONG_FORMAT, ovms::HTTPStatusCode::ERROR},

src/model_management/BUILD

@@ -0,0 +1,41 @@
+#
+# Copyright (c) 2026 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.
+#
+load("//:common_settings.bzl", "ovms_cc_library")
+
+ovms_cc_library(
+    name = "servable_loading_task",
+    hdrs = ["servable_loading_task.hpp"],
+    deps = select({
+        "//conditions:default": [],
+        "//:not_disable_mediapipe": [
+            "//src/mediapipe_internal:mediapipegraphconfig",
+        ],
+    }) + [
+        "//src:modelconfig",
+    ],
+    visibility = ["//visibility:public"],
+)
+
+ovms_cc_library(
+    name = "servable_loading_queue",
+    hdrs = ["servable_loading_queue.hpp"],
+    srcs = ["servable_loading_queue.cpp"],
+    deps = [
+        "servable_loading_task",
+        "//src:libovmslogging",
+    ],
+    visibility = ["//visibility:public"],
+)

src/model_management/servable_loading_queue.cpp

@@ -0,0 +1,97 @@
+//*****************************************************************************
+// Copyright 2026 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.
+//*****************************************************************************
+#include "servable_loading_queue.hpp"
+
+#include <utility>
+
+#include "src/logging.hpp"
+
+namespace ovms {
+
+ServableLoadingQueue::~ServableLoadingQueue() {
+    stop();
+}
+
+void ServableLoadingQueue::start(TaskProcessor processor) {
+    std::lock_guard<std::mutex> lock(this->mutex);
+    if (this->running) {
+        return;
+    }
+    this->processor = std::move(processor);
+    this->running = true;
+    this->worker = std::thread(&ServableLoadingQueue::workerLoop, this);
+}
+
+void ServableLoadingQueue::requestStop() {
+    {
+        std::lock_guard<std::mutex> lock(this->mutex);
+        if (!this->running) {
+            return;
+        }
+        this->running = false;
+    }
+    this->cv.notify_one();
+}
+
+void ServableLoadingQueue::stop() {
+    requestStop();
+    if (this->worker.joinable()) {
+        this->worker.join();
+    }
+    std::lock_guard<std::mutex> lock(this->mutex);
+    while (!this->queue.empty()) {
+        auto& task = this->queue.front();
+        task.completion.set_value(StatusCode::SERVER_SHUTTING_DOWN);
+        this->queue.pop_front();
+    }
+}
+
+std::future<Status> ServableLoadingQueue::scheduleTask(ServableLoadingTask task, bool urgent) {
+    auto future = task.completion.get_future();
+    {
+        std::lock_guard<std::mutex> lock(this->mutex);
+        if (urgent) {
+            this->queue.push_front(std::move(task));
+        } else {
+            this->queue.push_back(std::move(task));
+        }
+    }
+    this->cv.notify_one();
+    return future;
+}
+
+void ServableLoadingQueue::workerLoop() {
+    SPDLOG_LOGGER_INFO(modelmanager_logger, "Started servable loading queue thread");
+    while (true) {
+        ServableLoadingTask task{ServableLoadingTaskType::LoadModel, ""};
+        {
+            std::unique_lock<std::mutex> lock(this->mutex);
+            this->cv.wait(lock, [this] { return !this->queue.empty() || !this->running; });
+            if (!this->running) {
+                break;
+            }
+            task = std::move(this->queue.front());
+            this->queue.pop_front();
+        }
+        SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Processing {} task for: {}",
+            static_cast<int>(task.type), task.name);
+        Status status = this->processor(task);
+        task.completion.set_value(status);
+    }
+    SPDLOG_LOGGER_INFO(modelmanager_logger, "Stopped servable loading queue thread");
+}
+
+}  // namespace ovms

src/model_management/servable_loading_queue.hpp

@@ -0,0 +1,60 @@
+//*****************************************************************************
+// Copyright 2026 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 <condition_variable>
+#include <deque>
+#include <functional>
+#include <future>
+#include <mutex>
+#include <thread>
+
+#include "servable_loading_task.hpp"
+
+namespace ovms {
+
+using TaskProcessor = std::function<Status(ServableLoadingTask&)>;
+
+class ServableLoadingQueue {
+public:
+    ServableLoadingQueue() = default;
+    ~ServableLoadingQueue();
+
+    ServableLoadingQueue(const ServableLoadingQueue&) = delete;
+    ServableLoadingQueue& operator=(const ServableLoadingQueue&) = delete;
+
+    void start(TaskProcessor processor);
+    // Blocks until worker joins, then drains pending tasks. Must always be called.
+    void stop();
+    // Signals worker to stop without blocking. stop() must still be called after.
+    void requestStop();
+
+    // Returns future that resolves when the task completes.
+    // urgent=true inserts at front (for inference-triggered loads).
+    std::future<Status> scheduleTask(ServableLoadingTask task, bool urgent = false);
+
+private:
+    void workerLoop();
+
+    TaskProcessor processor;
+    std::thread worker;
+    std::deque<ServableLoadingTask> queue;
+    std::mutex mutex;
+    std::condition_variable cv;
+    bool running = false;
+};
+
+}  // namespace ovms

src/model_management/servable_loading_task.hpp

@@ -0,0 +1,67 @@
+//*****************************************************************************
+// Copyright 2026 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 <future>
+#include <optional>
+#include <string>
+
+#include "src/modelconfig.hpp"
+#if (MEDIAPIPE_DISABLE == 0)
+#include "src/mediapipe_internal/mediapipegraphconfig.hpp"
+#endif
+
+namespace ovms {
+
+enum class ServableLoadingTaskType {
+    LoadModel,
+    RetireModel,
+    LoadMediapipe,
+    UnloadMediapipe
+};
+
+struct ServableLoadingTask {
+    ServableLoadingTaskType type;
+    std::string name;
+    std::optional<ModelConfig> modelConfig;
+#if (MEDIAPIPE_DISABLE == 0)
+    std::optional<MediapipeGraphConfig> graphConfig;
+#endif
+    std::promise<Status> completion;
+
+    ServableLoadingTask(ServableLoadingTaskType type, const std::string& name, const ModelConfig& config) :
+        type(type),
+        name(name),
+        modelConfig(config) {}
+
+#if (MEDIAPIPE_DISABLE == 0)
+    ServableLoadingTask(ServableLoadingTaskType type, const std::string& name, const MediapipeGraphConfig& config) :
+        type(type),
+        name(name),
+        graphConfig(config) {}
+#endif
+
+    ServableLoadingTask(ServableLoadingTaskType type, const std::string& name) :
+        type(type),
+        name(name) {}
+
+    ServableLoadingTask(ServableLoadingTask&&) = default;
+    ServableLoadingTask& operator=(ServableLoadingTask&&) = default;
+    ServableLoadingTask(const ServableLoadingTask&) = delete;
+    ServableLoadingTask& operator=(const ServableLoadingTask&) = delete;
+};
+
+}  // namespace ovms