From f418629c5dc971aa685f967dc8aaf16bfdf3851e Mon Sep 17 00:00:00 2001 From: Ryan Doupe Date: Thu, 15 Oct 2026 03:25:30 -0400 Subject: [PATCH] requesthandler: Keep Parallel batch results in request order A RequestBatch with executionType Parallel runs each request as its own thread-pool task, and each task appended its result when it finished, so the results came back in completion order. The protocol layer labels results[i] with the requestType and requestId of requests[i], so with a slow request first (a full-size GetSourceScreenshot) and fast ones after it every result was attached to another request (upstream obsproject/obs-websocket#1317). The results vector now has one slot per request and each task writes to its request's index; the batch waits for a finished count instead of the vector's size. Parallel results now come back in request order, as serial batches already do. Co-Authored-By: Claude Opus 5.5 --- src/requesthandler/RequestBatchHandler.cpp | 26 +++++++++++++++------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/src/requesthandler/RequestBatchHandler.cpp b/src/requesthandler/RequestBatchHandler.cpp index 15302aa..bbd3ff1 100644 --- a/src/requesthandler/RequestBatchHandler.cpp +++ b/src/requesthandler/RequestBatchHandler.cpp @@ -46,12 +46,18 @@ struct SerialFrameBatch { struct ParallelBatchResults { RequestHandler &requestHandler; + // One slot per request, in request order std::vector results; + size_t finishedCount = 0; std::mutex conditionMutex; std::condition_variable condition; - ParallelBatchResults(RequestHandler &requestHandler) : requestHandler(requestHandler) {} + ParallelBatchResults(RequestHandler &requestHandler, size_t requestCount) + : requestHandler(requestHandler), + results(requestCount) + { + } }; // `{"inputName": "inputNameVariable"}` is essentially `inputName = inputNameVariable` @@ -203,27 +209,31 @@ RequestBatchHandler::ProcessRequestBatch(QThreadPool &threadPool, SessionPtr ses return serialFrameBatch.results; } else if (executionType == RequestBatchExecutionType::Parallel) { - ParallelBatchResults parallelResults(requestHandler); + size_t requestCount = requests.size(); + ParallelBatchResults parallelResults(requestHandler, requestCount); // Acquire the lock early to prevent the batch from finishing before we're ready std::unique_lock lock(parallelResults.conditionMutex); - // Submit each request as a task to the thread pool to be processed ASAP - for (auto &request : requests) { - threadPool.start(Utils::Compat::CreateFunctionRunnable([¶llelResults, &request]() { + // Submit each request as a task to the thread pool to be processed ASAP. + // Each task stores its result at its request's index, so the results stay + // in request order whatever order the requests finish in. + for (size_t i = 0; i < requestCount; i++) { + RequestBatchRequest &request = requests[i]; + threadPool.start(Utils::Compat::CreateFunctionRunnable([¶llelResults, &request, i]() { RequestResult requestResult = parallelResults.requestHandler.ProcessRequest(request); std::unique_lock lock(parallelResults.conditionMutex); - parallelResults.results.push_back(requestResult); + parallelResults.results[i] = requestResult; + parallelResults.finishedCount++; lock.unlock(); parallelResults.condition.notify_one(); })); } // Wait for the last request to finish processing - size_t requestCount = requests.size(); parallelResults.condition.wait(lock, [¶llelResults, requestCount] { - return parallelResults.results.size() == requestCount; + return parallelResults.finishedCount == requestCount; }); return parallelResults.results;