diff --git a/src/inference_executor.hpp b/src/inference_executor.hpp index 1092c82f7c..8c16fa7357 100644 --- a/src/inference_executor.hpp +++ b/src/inference_executor.hpp @@ -48,6 +48,26 @@ enum : unsigned int { POSTPROCESS, TIMER_END }; + +// Releases the infer-request slot acquired by modelInferAsync when start_async() throws. +// +// The stream-id guard that owns the slot was moved into the completion-callback lambda; +// since start_async() failed, that callback will never fire to destroy the lambda, so the +// slot would leak (#2871). Clearing the callback destroys the lambda and returns the slot. +// +// The callback lambda also holds a (copied) OutputKeeper. Clearing the callback runs while +// OpenVINO holds the InferRequest's internal lock, so the lambda's OutputKeeper copy must not +// reset output tensors there - doing so would reenter this InferRequest and deadlock +// (std::terminate via an exception leaving the noexcept destructor). The caller retains the +// original OutputKeeper handle (capture-by-copy), so we cancel it first: inference never ran, +// so there is nothing to restore, and the retained handle's destructor then becomes a no-op. +inline void releaseInferRequestSlotAfterStartAsyncFailure(ov::InferRequest& inferRequest, const std::shared_ptr& outKeeper) { + if (outKeeper) { + outKeeper->cancel(); + } + inferRequest.set_callback([](std::exception_ptr) {}); +} + template Status modelInferAsync(ModelInstance& instance, const RequestType* request, std::unique_ptr& modelUnloadGuardPtr) { @@ -136,7 +156,7 @@ Status modelInferAsync(ModelInstance& instance, const RequestType* request, { // order is important here - destructors are called in order from right to left inferRequest.set_callback( - [&instance, request, &inferRequest, userCallback, userCallbackData, modelUnloadGuardPtrMoved = std::shared_ptr(std::move(modelUnloadGuardPtr)), streamIdGuardMoved = std::move(executingStreamIdGuard), movedOutputKeeper = std::move(outKeeper)](std::exception_ptr exception) mutable { + [&instance, request, &inferRequest, userCallback, userCallbackData, modelUnloadGuardPtrMoved = std::shared_ptr(std::move(modelUnloadGuardPtr)), streamIdGuardMoved = std::move(executingStreamIdGuard), movedOutputKeeper = outKeeper](std::exception_ptr exception) mutable { struct CallbackGuard { OVMS_InferenceRequestCompletionCallback_t userCallback{nullptr}; void* userCallbackData{nullptr}; @@ -204,12 +224,14 @@ Status modelInferAsync(ModelInstance& instance, const RequestType* request, try { SPDLOG_DEBUG("ov::InferRequest: {}, inferRequest.start_async()", reinterpret_cast(&inferRequest)); - inferRequest.start_async(); + instance.startAsyncInference(inferRequest); } catch (std::exception& e) { SPDLOG_DEBUG("caught exception in ov::InferRequest.start_async: {}", e.what()); + releaseInferRequestSlotAfterStartAsyncFailure(inferRequest, outKeeper); return StatusCode::OV_INTERNAL_INFERENCE_ERROR; } catch (...) { SPDLOG_DEBUG("caught exception in ov::InferRequest.start_async"); + releaseInferRequestSlotAfterStartAsyncFailure(inferRequest, outKeeper); return StatusCode::OV_INTERNAL_INFERENCE_ERROR; } return StatusCode::OK; diff --git a/src/modelinstance.cpp b/src/modelinstance.cpp index 438b902b5e..d747beedc2 100644 --- a/src/modelinstance.cpp +++ b/src/modelinstance.cpp @@ -1604,6 +1604,10 @@ OVInferRequestsQueue& ModelInstance::getInferRequestsQueue() { return *inferRequestsQueue; } +void ModelInstance::startAsyncInference(ov::InferRequest& inferRequest) { + inferRequest.start_async(); +} + const size_t ModelInstance::getBatchSizeIndex() const { const auto& inputItr = this->inputsInfo.cbegin(); if (inputItr == this->inputsInfo.cend()) { diff --git a/src/modelinstance.hpp b/src/modelinstance.hpp index 76941e480c..63bfe07cee 100644 --- a/src/modelinstance.hpp +++ b/src/modelinstance.hpp @@ -533,6 +533,16 @@ class ModelInstance : public Servable { */ OVInferRequestsQueue& getInferRequestsQueue(); + /** + * @brief Starts asynchronous inference on the given infer request. + * + * Thin virtual wrapper over ov::InferRequest::start_async. In production it + * forwards unconditionally; it exists as a seam so tests can simulate + * start_async() throwing and verify the infer-request slot is released on + * that error path (see modelInferAsync and #2871). + */ + virtual void startAsyncInference(ov::InferRequest& inferRequest); + /** * @brief Combines plugin config from user with default config calculated at runtime * diff --git a/src/outputkeeper.cpp b/src/outputkeeper.cpp index f81a1de19b..886878027b 100644 --- a/src/outputkeeper.cpp +++ b/src/outputkeeper.cpp @@ -35,6 +35,9 @@ OutputKeeper::OutputKeeper(ov::InferRequest& request, const tensor_map_t& output } } OutputKeeper::~OutputKeeper() { + if (cancelled) { + return; + } for (auto [name, v] : outputs) { OV_LOGGER("ov::InferRequest: {}, request.set_tensor({}, {})", reinterpret_cast(&request), name, reinterpret_cast(&v)); request.set_tensor(name, v); diff --git a/src/outputkeeper.hpp b/src/outputkeeper.hpp index c755f3a070..87d79bfadc 100644 --- a/src/outputkeeper.hpp +++ b/src/outputkeeper.hpp @@ -27,7 +27,12 @@ namespace ovms { struct OutputKeeper { std::unordered_map outputs; ov::InferRequest& request; + bool cancelled{false}; OutputKeeper(ov::InferRequest& request, const tensor_map_t& outputsInfo); ~OutputKeeper(); + // Disable the output-tensor restore performed by the destructor. Used on the async + // start_async() error path, where inference never ran (so there is nothing to restore) + // and reentering the InferRequest from the destructor would deadlock (#2871). + void cancel() { this->cancelled = true; } }; } // namespace ovms diff --git a/src/test/c_api_tests.cpp b/src/test/c_api_tests.cpp index e59f27cbf4..0573eeba73 100644 --- a/src/test/c_api_tests.cpp +++ b/src/test/c_api_tests.cpp @@ -29,6 +29,7 @@ #include "../capi_frontend/capi_request_utils.hpp" #include "../deserialization_main.hpp" #include "../inference_executor.hpp" +#include "../ovinferrequestsqueue.hpp" #include "../capi_frontend/capi_utils.hpp" #include "../capi_frontend/serialization.hpp" #include "../capi_frontend/deserialization.hpp" @@ -2046,3 +2047,59 @@ TEST_F(CAPIInference, AsyncErrorHandling) { unloadGuard.reset(); instance.retireModel(); } + +// Model instance whose async inference start deterministically throws, to exercise the +// modelInferAsync start_async() error path (see #2871). Production ModelInstance forwards +// startAsyncInference() to ov::InferRequest::start_async(); here we override it to throw. +class MockModelInstanceThrowingOnStartAsync : public ovms::ModelInstance { +public: + MockModelInstanceThrowingOnStartAsync(ov::Core& ieCore) : + ModelInstance(std::string("UNUSED_NAME"), 0, ieCore, nullptr, nullptr) { + status = ovms::ModelVersionStatus("UNUSED_NAME", UNUSED_MODEL_VERSION, ovms::ModelVersionState::START); + } + virtual ~MockModelInstanceThrowingOnStartAsync() {} + ovms::Status loadModel(const ovms::ModelConfig& config) override { + return ModelInstance::loadModel(config); + } + void startAsyncInference(ov::InferRequest& inferRequest) override { + throw std::runtime_error("simulated start_async failure"); + } +}; + +// Regression test for #2871: when start_async() throws, modelInferAsync must release the +// infer-request slot it acquired. With a single-stream pool, a leak makes the slot +// permanently unavailable; we assert it is returned by querying the queue afterwards. +TEST_F(CAPIInference, AsyncStartAsyncThrowReleasesInferRequestSlot) { + ov::Core core; + MockModelInstanceThrowingOnStartAsync instance(core); + ovms::ModelConfig config = DUMMY_MODEL_CONFIG; + config.setNireq(1); // single slot -> a leaked slot is observable + ASSERT_EQ(instance.loadModel(config), ovms::StatusCode::OK); + std::unique_ptr unloadGuard; + instance.waitForLoaded(0, unloadGuard); + + // Sanity: the single slot is idle before inference. + { + auto idle = instance.getInferRequestsQueue().tryToGetIdleStream(); + ASSERT_TRUE(idle.has_value()); + instance.getInferRequestsQueue().returnStream(idle.value()); + } + + ovms::InferenceRequest request("dummy", 0); + std::vector in(DUMMY_MODEL_INPUT_SIZE, INITIAL_VALUE); + request.addInput(DUMMY_MODEL_INPUT_NAME, OVMS_DATATYPE_FP32, DUMMY_MODEL_SHAPE.data(), DUMMY_MODEL_SHAPE.size()); + request.setInputBuffer(DUMMY_MODEL_INPUT_NAME, in.data(), DUMMY_MODEL_INPUT_SIZE * sizeof(float), OVMS_BUFFERTYPE_CPU, 0); + CallbackUnblockingAndCheckingStruct callbackStruct; + request.setCompletionCallback(callbackCheckingIfErrorReported, &callbackStruct); + + auto status = ovms::modelInferAsync(instance, &request, unloadGuard); + EXPECT_EQ(status, ovms::StatusCode::OV_INTERNAL_INFERENCE_ERROR) << status.string(); + + // The slot acquired by modelInferAsync must have been returned to the pool despite the + // start_async() failure. Without the fix it leaks and tryToGetIdleStream() returns nullopt. + auto idleAfter = instance.getInferRequestsQueue().tryToGetIdleStream(); + EXPECT_TRUE(idleAfter.has_value()) << "infer-request slot leaked after start_async() threw (#2871)"; + + unloadGuard.reset(); + instance.retireModel(); +}