From 16b513c75d28d707b77eba9dabc87cbf5cd551f3 Mon Sep 17 00:00:00 2001 From: Adrian Tobiszewski Date: Mon, 24 Aug 2026 16:39:00 +0200 Subject: [PATCH 1/3] Initial refactor commit --- src/BUILD | 1 + src/mediapipe_internal/mediapipefactory.cpp | 1 + src/model_management/BUILD | 26 ++++ .../servable_loading_queue.cpp | 87 +++++++++++++ .../servable_loading_queue.hpp | 57 +++++++++ .../servable_loading_task.hpp | 67 ++++++++++ src/modelmanager.cpp | 119 +++++++++++++----- src/modelmanager.hpp | 6 +- src/test/metrics_flow_test.cpp | 2 + 9 files changed, 335 insertions(+), 31 deletions(-) create mode 100644 src/model_management/BUILD create mode 100644 src/model_management/servable_loading_queue.cpp create mode 100644 src/model_management/servable_loading_queue.hpp create mode 100644 src/model_management/servable_loading_task.hpp diff --git a/src/BUILD b/src/BUILD index 4fa62be9e2..02f46725d1 100644 --- a/src/BUILD +++ b/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", diff --git a/src/mediapipe_internal/mediapipefactory.cpp b/src/mediapipe_internal/mediapipefactory.cpp index 5f8deebd6b..4d3b0d803a 100644 --- a/src/mediapipe_internal/mediapipefactory.cpp +++ b/src/mediapipe_internal/mediapipefactory.cpp @@ -141,6 +141,7 @@ Status MediapipeFactory::create(std::unique_ptr& pipelin return StatusCode::MEDIAPIPE_DEFINITION_NAME_MISSING; } auto& definition = *it->second; + lock.unlock(); return definition.create(pipeline); } diff --git a/src/model_management/BUILD b/src/model_management/BUILD new file mode 100644 index 0000000000..b0100c1797 --- /dev/null +++ b/src/model_management/BUILD @@ -0,0 +1,26 @@ +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"], +) diff --git a/src/model_management/servable_loading_queue.cpp b/src/model_management/servable_loading_queue.cpp new file mode 100644 index 0000000000..5e46647dc0 --- /dev/null +++ b/src/model_management/servable_loading_queue.cpp @@ -0,0 +1,87 @@ +//***************************************************************************** +// 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 + +#include "src/logging.hpp" + +namespace ovms { + +ServableLoadingQueue::~ServableLoadingQueue() { + stop(); +} + +void ServableLoadingQueue::start(TaskProcessor processor) { + std::lock_guard lock(this->mutex); + if (this->running) { + return; + } + this->processor = std::move(processor); + this->running = true; + this->worker = std::thread(&ServableLoadingQueue::workerLoop, this); +} + +void ServableLoadingQueue::stop() { + { + std::lock_guard lock(this->mutex); + if (!this->running) { + return; + } + this->running = false; + } + this->cv.notify_one(); + if (this->worker.joinable()) { + this->worker.join(); + } +} + +std::future ServableLoadingQueue::scheduleTask(ServableLoadingTask task, bool urgent) { + auto future = task.completion.get_future(); + { + std::lock_guard 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 lock(this->mutex); + this->cv.wait(lock, [this] { return !this->queue.empty() || !this->running; }); + if (!this->running && this->queue.empty()) { + break; + } + task = std::move(this->queue.front()); + this->queue.pop_front(); + } + SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Processing {} task for: {}", + static_cast(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 diff --git a/src/model_management/servable_loading_queue.hpp b/src/model_management/servable_loading_queue.hpp new file mode 100644 index 0000000000..88509b2ae2 --- /dev/null +++ b/src/model_management/servable_loading_queue.hpp @@ -0,0 +1,57 @@ +//***************************************************************************** +// 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 +#include +#include +#include +#include +#include + +#include "servable_loading_task.hpp" + +namespace ovms { + +using TaskProcessor = std::function; + +class ServableLoadingQueue { +public: + ServableLoadingQueue() = default; + ~ServableLoadingQueue(); + + ServableLoadingQueue(const ServableLoadingQueue&) = delete; + ServableLoadingQueue& operator=(const ServableLoadingQueue&) = delete; + + void start(TaskProcessor processor); + void stop(); + + // Returns future that resolves when the task completes. + // urgent=true inserts at front (for inference-triggered loads). + std::future scheduleTask(ServableLoadingTask task, bool urgent = false); + +private: + void workerLoop(); + + TaskProcessor processor; + std::thread worker; + std::deque queue; + std::mutex mutex; + std::condition_variable cv; + bool running = false; +}; + +} // namespace ovms diff --git a/src/model_management/servable_loading_task.hpp b/src/model_management/servable_loading_task.hpp new file mode 100644 index 0000000000..797398a735 --- /dev/null +++ b/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 +#include +#include + +#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; +#if (MEDIAPIPE_DISABLE == 0) + std::optional graphConfig; +#endif + std::promise 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 diff --git a/src/modelmanager.cpp b/src/modelmanager.cpp index c14c1efe4e..78e1cacb82 100644 --- a/src/modelmanager.cpp +++ b/src/modelmanager.cpp @@ -58,6 +58,7 @@ #include "filesystem/filesystemfactory.hpp" #include "graph_export/graph_export.hpp" #include "logging.hpp" +#include "model_management/servable_loading_queue.hpp" #if (MEDIAPIPE_DISABLE == 0) #include "mediapipe_internal/mediapipefactory.hpp" #include "mediapipe_internal/mediapipegraphdefinition.hpp" @@ -82,6 +83,7 @@ const std::string DEFAULT_MODEL_CACHE_DIRECTORY = "c:\\Intel\\openvino_cache"; const std::string DEFAULT_MODEL_CACHE_DIRECTORY = "/opt/cache"; #endif ModelManager::ModelManager(const std::string& modelCacheDirectory, MetricRegistry* registry, PythonBackend* pythonBackend) : + loadingQueue(std::make_unique()), pipelineFactory(std::make_unique()), #if (MEDIAPIPE_DISABLE == 0) mediapipeFactory(std::make_unique(pythonBackend)), @@ -92,6 +94,60 @@ ModelManager::ModelManager(const std::string& modelCacheDirectory, MetricRegistr metricRegistry(registry), pythonBackend(pythonBackend) { this->ieCore = std::make_unique(); + loadingQueue->start([this](ServableLoadingTask& task) -> Status { + switch (task.type) { + case ServableLoadingTaskType::LoadModel: { + if (!task.modelConfig.has_value()) { + auto it = servedModelConfigs.find(task.name); + if (it == servedModelConfigs.end()) + return StatusCode::MODEL_NAME_MISSING; + task.modelConfig = it->second; + } + return reloadModelWithVersions(task.modelConfig.value()); + } + case ServableLoadingTaskType::RetireModel: { + auto model = findModelByName(task.name); + if (!model) { + return StatusCode::MODEL_NAME_MISSING; + } + model->retireAllVersions(); + return StatusCode::OK; + } +#if (MEDIAPIPE_DISABLE == 0) + case ServableLoadingTaskType::LoadMediapipe: { + auto* def = mediapipeFactory->findDefinitionByName(task.name); + if (task.graphConfig.has_value()) { + const auto& config = task.graphConfig.value(); + if (!def) { + return mediapipeFactory->createDefinition(task.name, config, *this, *this); + } + if (def->isReloadRequired(config)) { + return mediapipeFactory->reloadDefinition(task.name, config, *this); + } + } else { + // Urgent reload with existing config (inference-triggered) + if (!def) + return StatusCode::MEDIAPIPE_DEFINITION_NOT_LOADED_ANYMORE; + return def->reload(*this, def->getMediapipeGraphConfig()); + } + return StatusCode::OK; + } + case ServableLoadingTaskType::UnloadMediapipe: { + auto* def = mediapipeFactory->findDefinitionByName(task.name); + if (!def) { + return StatusCode::MEDIAPIPE_DEFINITION_NOT_LOADED_ANYMORE; + } + def->retire(); + return StatusCode::OK; + } +#else + case ServableLoadingTaskType::LoadMediapipe: + case ServableLoadingTaskType::UnloadMediapipe: + return StatusCode::INTERNAL_ERROR; +#endif + } + return StatusCode::INTERNAL_ERROR; + }); OV_LOGGER("ov::Core(): {}", reinterpret_cast(this->ieCore.get())); // Take --cache_dir from CLI @@ -449,28 +505,6 @@ Status ModelManager::validateUserSettingsInSingleModelCliGraphStart(const Models return StatusCode::OK; } -Status ModelManager::processMediapipeConfig(const MediapipeGraphConfig& config, std::set& mediapipesInConfigFile, MediapipeFactory& factory) { - if (mediapipesInConfigFile.find(config.getGraphName()) != mediapipesInConfigFile.end()) { - SPDLOG_LOGGER_WARN(modelmanager_logger, "Duplicated mediapipe names: {} defined in config file. Only first graph will be loaded.", config.getGraphName()); - return StatusCode::OK; - } - mediapipesInConfigFile.insert(config.getGraphName()); - MediapipeGraphDefinition* mediapipeGraphDefinition = factory.findDefinitionByName(config.getGraphName()); - if (mediapipeGraphDefinition == nullptr) { - SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Mediapipe graph:{} was not loaded so far. Triggering load", config.getGraphName()); - auto status = factory.createDefinition(config.getGraphName(), config, *this, *this); - return status; - } - if (mediapipeGraphDefinition->isReloadRequired(config)) { - SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Mediapipe graph:{} triggering reload", config.getGraphName()); - auto status = factory.reloadDefinition(config.getGraphName(), - config, - *this); - return status; - } - SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Mediapipe graph:{} already loaded and reload is not required", config.getGraphName()); - return StatusCode::OK; -} #endif #if (MEDIAPIPE_DISABLE == 0) @@ -545,12 +579,18 @@ Status ModelManager::loadMediapipeGraphsConfig(std::vector mediapipesInConfigFileNames.insert(mediapipeGraphConfig.getGraphName()); } mediapipeFactory->retireOtherThan(std::move(mediapipesInConfigFileNames)); - std::set mediapipesAlreadyLoaded; + std::set alreadyScheduled; for (const auto& mediapipeGraphConfig : mediapipesInConfigFile) { + if (!alreadyScheduled.insert(mediapipeGraphConfig.getGraphName()).second) { + SPDLOG_LOGGER_WARN(modelmanager_logger, "Duplicated mediapipe names: {} defined in config file. Only first graph will be loaded.", mediapipeGraphConfig.getGraphName()); + continue; + } if (spdlog::default_logger_raw()->level() <= spdlog::level::debug) { mediapipeGraphConfig.logGraphConfigContent(); } - auto status = processMediapipeConfig(mediapipeGraphConfig, mediapipesAlreadyLoaded, *mediapipeFactory); + ServableLoadingTask task{ServableLoadingTaskType::LoadMediapipe, mediapipeGraphConfig.getGraphName(), mediapipeGraphConfig}; + auto future = loadingQueue->scheduleTask(std::move(task)); + auto status = future.get(); if (status != StatusCode::OK) { IF_ERROR_NOT_OCCURRED_EARLIER_THEN_SET_FIRST_ERROR(status); } @@ -751,7 +791,9 @@ Status ModelManager::ConfigLoader::loadModels(ModelManager& modelManager, const continue; } - status = modelManager.reloadModelWithVersions(modelConfig); + ServableLoadingTask task{ServableLoadingTaskType::LoadModel, modelName, modelConfig}; + auto future = modelManager.loadingQueue->scheduleTask(std::move(task)); + status = future.get(); IF_ERROR_NOT_OCCURRED_EARLIER_THEN_SET_FIRST_ERROR(status); modelsInConfigFile.emplace(modelName); @@ -867,7 +909,9 @@ Status ModelManager::tryReloadGatedModelConfigs(std::vector& gatedM Status firstErrorStatus = StatusCode::OK; for (auto& modelConfig : gatedModelConfigs) { SPDLOG_LOGGER_DEBUG(modelmanager_logger, "Trying to reload model({}) configuration", modelConfig.getName()); - auto status = reloadModelWithVersions(modelConfig); + ServableLoadingTask task{ServableLoadingTaskType::LoadModel, modelConfig.getName(), modelConfig}; + auto future = loadingQueue->scheduleTask(std::move(task)); + auto status = future.get(); if (!status.ok()) { IF_ERROR_NOT_OCCURRED_EARLIER_THEN_SET_FIRST_ERROR(status); continue; @@ -885,7 +929,7 @@ Status ModelManager::tryReloadGatedModelConfigs(std::vector& gatedM Status ModelManager::loadConfig() { rapidjson::Document configJson; - std::lock_guard loadingLock(configMtx); + std::lock_guard loadingLock(configMtx); // TODO(idle-unload): @atobiszei narrow scope to parsing-only after queue refactoring Status status = parseConfig(this->configFilename, configJson, this->lastConfigFileMD5, WRONG_CONFIG_FILE_RETRY_DELAY_MS, MAX_CONFIG_JSON_READ_RETRY_COUNT); if (!status.ok()) { this->lastLoadConfigStatus = status; @@ -991,13 +1035,15 @@ void ModelManager::retireModelsRemovedFromConfigFile(const std::set } Status ModelManager::updateConfigurationWithoutConfigFile() { - std::lock_guard loadingLock(configMtx); + std::lock_guard loadingLock(configMtx); // TODO(idle-unload): @atobiszei narrow scope to parsing-only after queue refactoring SPDLOG_LOGGER_TRACE(modelmanager_logger, "Checking if something changed with model versions"); bool reloadNeeded = false; Status firstErrorStatus = StatusCode::OK; Status status; for (auto& [name, config] : servedModelConfigs) { - status = reloadModelWithVersions(config); + ServableLoadingTask task{ServableLoadingTaskType::LoadModel, name, config}; + auto future = loadingQueue->scheduleTask(std::move(task)); + status = future.get(); if (!status.ok()) { IF_ERROR_NOT_OCCURRED_EARLIER_THEN_SET_FIRST_ERROR(status); } else if (status == StatusCode::OK_RELOADED) { @@ -1048,7 +1094,7 @@ void ModelManager::watcher(std::future exitSignal, bool watchConfigFile) { SPDLOG_LOGGER_INFO(modelmanager_logger, "Started model manager thread"); while (exitSignal.wait_for(std::chrono::milliseconds(this->watcherIntervalMillisec)) == std::future_status::timeout) { SPDLOG_LOGGER_TRACE(modelmanager_logger, "Models configuration and filesystem check cycle begin"); - std::unique_lock loadingLock(configMtx); + std::unique_lock loadingLock(configMtx); // TODO(idle-unload): @atobiszei narrow scope to parsing-only after queue refactoring if (watchConfigFile) { bool isNeeded; configFileReloadNeeded(isNeeded); @@ -1107,6 +1153,7 @@ void ModelManager::join() { if (cleanerStarted) { cleanerExitTrigger.set_value(); } + loadingQueue->stop(); if (watcherStarted) { if (monitor.joinable()) { @@ -1459,6 +1506,18 @@ Status ModelManager::reloadModelWithVersions(ModelConfig& config) { return blocking_status; } +std::future ModelManager::requestServableLoad(const std::string& name) { +// TODO @atobiszei check at which point the requestLoad is happening - shoudl be possible only on existing servable +#if (MEDIAPIPE_DISABLE == 0) + if (mediapipeFactory->findDefinitionByName(name)) { + ServableLoadingTask task{ServableLoadingTaskType::LoadMediapipe, name}; + return loadingQueue->scheduleTask(std::move(task), true); + } +#endif + ServableLoadingTask task{ServableLoadingTaskType::LoadModel, name}; + return loadingQueue->scheduleTask(std::move(task), true); +} + const std::shared_ptr ModelManager::findModelInstance(const std::string& name, model_version_t version) const { auto model = findModelByName(name); if (!model) { diff --git a/src/modelmanager.hpp b/src/modelmanager.hpp index 21f8dbad0e..429f4c59f8 100644 --- a/src/modelmanager.hpp +++ b/src/modelmanager.hpp @@ -60,6 +60,7 @@ class ModelInstance; class ServableDefinition; class ModelInstanceUnloadGuard; class Pipeline; +class ServableLoadingQueue; class PipelineFactory; struct FunctorResourcesCleaner; class PythonBackend; @@ -85,6 +86,7 @@ class ModelManager : public ServableNameChecker, public MetricProvider, public M std::map> models; std::unique_ptr ieCore; + std::unique_ptr loadingQueue; std::unique_ptr pipelineFactory; #if (MEDIAPIPE_DISABLE == 0) std::unique_ptr mediapipeFactory; @@ -109,7 +111,6 @@ class ModelManager : public ServableNameChecker, public MetricProvider, public M Status addModelVersions(std::shared_ptr& model, std::shared_ptr& fs, ModelConfig& config, std::shared_ptr& versionsToStart, std::shared_ptr& versionsFailed); #if (MEDIAPIPE_DISABLE == 0) - Status processMediapipeConfig(const MediapipeGraphConfig& config, std::set& mediapipesInConfigFile, MediapipeFactory& factory); Status loadMediapipeGraphsConfig(std::vector& mediapipesInConfigFile); Status loadMediapipeSubConfigModels(std::vector& gatedModelConfigs, std::set& modelsInConfigFile, std::set& modelsWithInvalidConfig, std::unordered_map& newModelConfigs, std::vector& mediapipesInConfigFile); @@ -394,6 +395,9 @@ class ModelManager : public ServableNameChecker, public MetricProvider, public M */ Status reloadModelWithVersions(ModelConfig& config); + // Enqueue an urgent servable load request (for inference threads). + std::future requestServableLoad(const std::string& name); + /** * @brief Starts model manager using ovms::Config * diff --git a/src/test/metrics_flow_test.cpp b/src/test/metrics_flow_test.cpp index e50e72cf90..c02db2889a 100644 --- a/src/test/metrics_flow_test.cpp +++ b/src/test/metrics_flow_test.cpp @@ -135,6 +135,8 @@ class ServerWithMockedManagerModule : public Server { module = this->createModule(GRPC_SERVER_MODULE_NAME); this->modules.emplace(GRPC_SERVER_MODULE_NAME, std::move(module)); } + // Modules hold a reference to manager; shut them down before manager is destroyed + ~ServerWithMockedManagerModule() override { shutdownModules(); } ConstructorEnabledModelManager& getManager() { return this->manager; From 1f0a42660d7eecf936e5346f1fce98a5a9ed8b47 Mon Sep 17 00:00:00 2001 From: Adrian Tobiszewski Date: Tue, 25 Aug 2026 16:33:12 +0200 Subject: [PATCH 2/3] fix style --- src/model_management/BUILD | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/model_management/BUILD b/src/model_management/BUILD index b0100c1797..83f4dc70cc 100644 --- a/src/model_management/BUILD +++ b/src/model_management/BUILD @@ -1,3 +1,18 @@ +# +# 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( From ea0ad0b8ddcbc1336215ac1b88a22f4e2f733532 Mon Sep 17 00:00:00 2001 From: Adrian Tobiszewski Date: Fri, 28 Aug 2026 11:41:25 +0200 Subject: [PATCH 3/3] Add scheduler queue tests --- src/BUILD | 15 +++++++++++++++ src/grpc_utils.cpp | 1 + src/http_server.cpp | 1 + src/model_management/servable_loading_queue.cpp | 14 ++++++++++++-- src/model_management/servable_loading_queue.hpp | 3 +++ src/modelmanager.cpp | 9 +++++---- src/status.cpp | 1 + src/status.hpp | 1 + 8 files changed, 39 insertions(+), 6 deletions(-) diff --git a/src/BUILD b/src/BUILD index 02f46725d1..3faa03918b 100644 --- a/src/BUILD +++ b/src/BUILD @@ -2259,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 +2371,20 @@ cc_library( linkopts = COMMON_STATIC_LIBS_LINKOPTS, ) +cc_library( + name = "servable_loading_queue_test", + srcs = ["test/servable_loading_queue_test.cpp"], + deps = [ + "//src/model_management:servable_loading_queue", + "@com_google_googletest//:gtest", + ], + linkstatic = 1, + alwayslink = 1, + local_defines = COMMON_LOCAL_DEFINES, + copts = COPTS_TESTS, + linkopts = COMMON_STATIC_LIBS_LINKOPTS, +) + cc_library( name = "test_utils", linkstatic = 1, diff --git a/src/grpc_utils.cpp b/src/grpc_utils.cpp index a9e41eb4eb..c75a8ca838 100644 --- a/src/grpc_utils.cpp +++ b/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()); diff --git a/src/http_server.cpp b/src/http_server.cpp index 747187d873..0e2c5635f7 100644 --- a/src/http_server.cpp +++ b/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}, diff --git a/src/model_management/servable_loading_queue.cpp b/src/model_management/servable_loading_queue.cpp index 5e46647dc0..7ac77d5b87 100644 --- a/src/model_management/servable_loading_queue.cpp +++ b/src/model_management/servable_loading_queue.cpp @@ -35,7 +35,7 @@ void ServableLoadingQueue::start(TaskProcessor processor) { this->worker = std::thread(&ServableLoadingQueue::workerLoop, this); } -void ServableLoadingQueue::stop() { +void ServableLoadingQueue::requestStop() { { std::lock_guard lock(this->mutex); if (!this->running) { @@ -44,9 +44,19 @@ void ServableLoadingQueue::stop() { this->running = false; } this->cv.notify_one(); +} + +void ServableLoadingQueue::stop() { + requestStop(); if (this->worker.joinable()) { this->worker.join(); } + std::lock_guard 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 ServableLoadingQueue::scheduleTask(ServableLoadingTask task, bool urgent) { @@ -70,7 +80,7 @@ void ServableLoadingQueue::workerLoop() { { std::unique_lock lock(this->mutex); this->cv.wait(lock, [this] { return !this->queue.empty() || !this->running; }); - if (!this->running && this->queue.empty()) { + if (!this->running) { break; } task = std::move(this->queue.front()); diff --git a/src/model_management/servable_loading_queue.hpp b/src/model_management/servable_loading_queue.hpp index 88509b2ae2..b7712b1c8f 100644 --- a/src/model_management/servable_loading_queue.hpp +++ b/src/model_management/servable_loading_queue.hpp @@ -37,7 +37,10 @@ class ServableLoadingQueue { 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). diff --git a/src/modelmanager.cpp b/src/modelmanager.cpp index 78e1cacb82..340092ea56 100644 --- a/src/modelmanager.cpp +++ b/src/modelmanager.cpp @@ -1153,7 +1153,7 @@ void ModelManager::join() { if (cleanerStarted) { cleanerExitTrigger.set_value(); } - loadingQueue->stop(); + loadingQueue->requestStop(); if (watcherStarted) { if (monitor.joinable()) { @@ -1170,6 +1170,7 @@ void ModelManager::join() { SPDLOG_INFO("Shutdown cleaner thread"); } } + loadingQueue->stop(); } void ModelManager::getVersionsToChange( @@ -1507,15 +1508,15 @@ Status ModelManager::reloadModelWithVersions(ModelConfig& config) { } std::future ModelManager::requestServableLoad(const std::string& name) { -// TODO @atobiszei check at which point the requestLoad is happening - shoudl be possible only on existing servable + const bool isPriorityRequest = true; #if (MEDIAPIPE_DISABLE == 0) if (mediapipeFactory->findDefinitionByName(name)) { ServableLoadingTask task{ServableLoadingTaskType::LoadMediapipe, name}; - return loadingQueue->scheduleTask(std::move(task), true); + return loadingQueue->scheduleTask(std::move(task), isPriorityRequest); } #endif ServableLoadingTask task{ServableLoadingTaskType::LoadModel, name}; - return loadingQueue->scheduleTask(std::move(task), true); + return loadingQueue->scheduleTask(std::move(task), isPriorityRequest); } const std::shared_ptr ModelManager::findModelInstance(const std::string& name, model_version_t version) const { diff --git a/src/status.cpp b/src/status.cpp index 0394192e86..6aca0fa0fe 100644 --- a/src/status.cpp +++ b/src/status.cpp @@ -318,6 +318,7 @@ const std::unordered_map Status::statusMessageMap = { {StatusCode::NONEXISTENT_LOG_LEVEL, "Tried to use nonexisting log level"}, {StatusCode::NONEXISTENT_PTR, "Tried to use nonexisting pointer"}, {StatusCode::SERVER_NOT_READY, "Server is not ready"}, + {StatusCode::SERVER_SHUTTING_DOWN, "Server is shutting down"}, // Server Start errors {StatusCode::OPTIONS_USAGE_ERROR, "options validation error"}, diff --git a/src/status.hpp b/src/status.hpp index 94be7948cb..cb81ece841 100644 --- a/src/status.hpp +++ b/src/status.hpp @@ -330,6 +330,7 @@ enum class StatusCode { NONEXISTENT_LOG_LEVEL, NONEXISTENT_PTR, SERVER_NOT_READY, + SERVER_SHUTTING_DOWN, // Server Start errors OPTIONS_USAGE_ERROR,