[releases/2026/0][Legacy] Remove disconnected requests from execution queue (#3968) (#3973)

Author: mzegla

src/llm/language_model/legacy/legacy_executor.cpp

@@ -32,16 +32,22 @@ size_t LegacyExecutor::requestsQueueSize() {
 
 void LegacyExecutor::processRequest() {
     OVMS_PROFILE_FUNCTION();
-    SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation started");
-    try {
-        requests.front()->results = pipe->generate(requests.front()->inputIds, requests.front()->generationConfigBuilder->getConfig(), requests.front()->textStreamer);
-    } catch (std::exception& e) {
-        requests.front()->success = false;
-        SPDLOG_LOGGER_ERROR(llm_executor_logger, "LLM pipeline generation failed: {}.", e.what());
+    auto& requestExecutionContext = requests.front();
+    if (requestExecutionContext->clientDisconnected) {
+        requestExecutionContext->success = false;
+        SPDLOG_LOGGER_DEBUG(llm_executor_logger, "Client disconnected, skipping request processing.");
+    } else {
+        SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation started");
+        try {
+            requestExecutionContext->results = pipe->generate(requestExecutionContext->inputIds, requestExecutionContext->generationConfigBuilder->getConfig(), requestExecutionContext->textStreamer);
+        } catch (std::exception& e) {
+            requestExecutionContext->success = false;
+            SPDLOG_LOGGER_ERROR(llm_executor_logger, "LLM pipeline generation failed: {}.", e.what());
+        }
+        SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation ended");
     }
-    SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation ended");
-    requests.front()->readySignal.set_value();
-    requests.front()->executionInProgress.notify_one();
+    requestExecutionContext->readySignal.set_value();
+    requestExecutionContext->executionInProgress.notify_one();
     std::unique_lock<std::mutex> lock(queueMutex);
     requests.pop();
 }

src/llm/language_model/legacy/servable.cpp

@@ -127,6 +127,12 @@ absl::Status LegacyServable::prepareInputs(std::shared_ptr<GenAiServableExecutio
 
 absl::Status LegacyServable::scheduleExecution(std::shared_ptr<GenAiServableExecutionContext>& executionContext) {
     auto legacyExecutionContext = std::static_pointer_cast<LegacyServableExecutionContext>(executionContext);
+    std::weak_ptr<LegacyServableExecutionContext> weakContext = legacyExecutionContext;
+    legacyExecutionContext->payload.client->registerDisconnectionCallback([weakContext]() {
+        if (auto context = weakContext.lock()) {
+            context->clientDisconnected = true;
+        }
+    });
     if (legacyExecutionContext->payload.client->isDisconnected()) {
         return absl::CancelledError();
     }

src/llm/language_model/legacy/servable.hpp

@@ -33,7 +33,10 @@ struct LegacyServableExecutionContext : public GenAiServableExecutionContext {
     std::condition_variable executionInProgress;
     // Workaround needed to pass generation config to the executor that requires it
     ov::genai::GenerationConfig baseGenerationConfig;
-    bool success = true;
+    bool success{true};
+
+    // Disconnection handling
+    std::atomic<bool> clientDisconnected{false};
 };
 
 struct LegacyServableProperties : public GenAiServableProperties {

src/llm/visual_language_model/legacy/legacy_executor.cpp

@@ -33,14 +33,22 @@ size_t VisualLanguageModelLegacyExecutor::requestsQueueSize() {
 
 void VisualLanguageModelLegacyExecutor::processRequest() {
     OVMS_PROFILE_FUNCTION();
-    try {
-        requests.front()->results = pipe->generate(requests.front()->inputText, requests.front()->inputImages, requests.front()->generationConfigBuilder->getConfig(), requests.front()->textStreamer);
-    } catch (std::exception& e) {
-        requests.front()->success = false;
-        SPDLOG_LOGGER_ERROR(llm_executor_logger, "VLM pipeline generation failed: {}.", e.what());
+    auto& requestExecutionContext = requests.front();
+    if (requestExecutionContext->clientDisconnected) {
+        requestExecutionContext->success = false;
+        SPDLOG_LOGGER_DEBUG(llm_executor_logger, "Client disconnected, skipping request processing.");
+    } else {
+        SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation started");
+        try {
+            requestExecutionContext->results = pipe->generate(requestExecutionContext->inputText, requestExecutionContext->inputImages, requestExecutionContext->generationConfigBuilder->getConfig(), requestExecutionContext->textStreamer);
+        } catch (std::exception& e) {
+            requestExecutionContext->success = false;
+            SPDLOG_LOGGER_ERROR(llm_executor_logger, "VLM pipeline generation failed: {}.", e.what());
+        }
+        SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation ended");
     }
-    requests.front()->readySignal.set_value();
-    requests.front()->executionInProgress.notify_one();
+    requestExecutionContext->readySignal.set_value();
+    requestExecutionContext->executionInProgress.notify_one();
     std::unique_lock<std::mutex> lock(queueMutex);
     requests.pop();
 }

src/llm/visual_language_model/legacy/servable.cpp

@@ -120,6 +120,12 @@ absl::Status VisualLanguageModelLegacyServable::parseRequest(std::shared_ptr<Gen
 
 absl::Status VisualLanguageModelLegacyServable::scheduleExecution(std::shared_ptr<GenAiServableExecutionContext>& executionContext) {
     auto legacyExecutionContext = std::static_pointer_cast<VisualLanguageModelLegacyServableExecutionContext>(executionContext);
+    std::weak_ptr<VisualLanguageModelLegacyServableExecutionContext> weakContext = legacyExecutionContext;
+    legacyExecutionContext->payload.client->registerDisconnectionCallback([weakContext]() {
+        if (auto context = weakContext.lock()) {
+            context->clientDisconnected = true;
+        }
+    });
     if (legacyExecutionContext->payload.client->isDisconnected()) {
         return absl::CancelledError();
     }

src/llm/visual_language_model/legacy/servable.hpp

@@ -36,7 +36,10 @@ struct VisualLanguageModelLegacyServableExecutionContext : public GenAiServableE
     std::string inputText;
     // Workaround needed to pass generation config to the executor that requires it
     ov::genai::GenerationConfig baseGenerationConfig;
-    bool success = true;
+    bool success{true};
+
+    // Disconnection handling
+    std::atomic<bool> clientDisconnected{false};
 };
 
 struct VisualLanguageModelLegacyServableProperties : public GenAiServableProperties {