Fix infer-request slot leak and deadlock on async start_async() error path (relates to #2871)

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<OutputKeeper>& outKeeper) {
+    if (outKeeper) {
+        outKeeper->cancel();
+    }
+    inferRequest.set_callback([](std::exception_ptr) {});
+}
+
 template <typename RequestType, typename ResponseType>
 Status modelInferAsync(ModelInstance& instance, const RequestType* request,
     std::unique_ptr<ModelInstanceUnloadGuard>& 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<ModelInstanceUnloadGuard>(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<ModelInstanceUnloadGuard>(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<void*>(&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;

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()) {

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
          *

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<void*>(&request), name, reinterpret_cast<void*>(&v));
         request.set_tensor(name, v);

src/outputkeeper.hpp

@@ -27,7 +27,12 @@ namespace ovms {
 struct OutputKeeper {
     std::unordered_map<std::string, ov::Tensor> 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

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<ModelInstanceUnloadGuard> 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<float> 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<ovms::InferenceRequest, ovms::InferenceResponse>(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();
+}