From fb16b8f76fb1d4a1902aba0eff6810677a60aa69 Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Sun, 6 Sep 2026 03:54:53 -0700 Subject: [PATCH 01/10] server: deliver a result to the one thread waiting for it server_response kept every pending result in one vector behind one condition variable. Each result was pushed after a linear walk of the waiting id set, then notify_all woke every waiting HTTP thread, and each of them took the same mutex and scanned the whole vector before going back to sleep. With N slots generating that is N wakeups and N vector scans per token, so N^2 per decode step, all of it contending for the mutex the decode thread needs to send the next token. Results are now queued on a per reader waiter. Ids registered together share one waiter, so a send is an O(1) lookup followed by a push and a wakeup of exactly the thread that asked for that task. Order is preserved: the waiter holds a FIFO and recv() takes the front, which is what scanning the shared vector from the start did. A reader whose ids have already been removed from the waiting list still waits out the poll interval it asked for rather than returning at once, so a caller that keeps polling does not spin, and the blocking recv() re-checks the running flag on a bounded wait so terminate() cannot leave it parked. Measured with llama-server at 32 slots, one request per slot, 128 prompt and 256 generated tokens: queue_results.send() 119.0 us to 3.6 us per call, and the whole result path per decode step 3.97 ms to 0.34 ms. --- tools/server/server-queue.cpp | 139 ++++++++++++++++++++++------------ tools/server/server-queue.h | 36 +++++++-- 2 files changed, 121 insertions(+), 54 deletions(-) diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 78169e9a5d86..de663ddabf59 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -387,79 +387,114 @@ void server_queue::cleanup_pending_task(int id_target) { // void server_response::add_waiting_task_id(int id_task) { - RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting_task_ids.size()); - std::unique_lock lock(mutex_results); - waiting_task_ids.insert(id_task); + + RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting.size()); + + waiting.emplace(id_task, std::make_shared()); } void server_response::add_waiting_task_ids(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); + // one waiter for the whole set: these ids belong to a single reader, which waits for any + // of them at a time + auto w = std::make_shared(); + for (const auto & id_task : id_tasks) { - RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting_task_ids.size()); - waiting_task_ids.insert(id_task); + RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting.size()); + waiting.emplace(id_task, w); } } void server_response::remove_waiting_task_id(int id_task) { - RES_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting_task_ids.size()); - std::unique_lock lock(mutex_results); - waiting_task_ids.erase(id_task); - // make sure to clean up all pending results - queue_results.erase( - std::remove_if(queue_results.begin(), queue_results.end(), [id_task](const server_task_result_ptr & res) { + + RES_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting.size()); + + auto it = waiting.find(id_task); + if (it == waiting.end()) { + return; + } + + // make sure to clean up all pending results of this task, the waiter may still be held by + // the other ids of the same reader + auto & results = it->second->results; + results.erase( + std::remove_if(results.begin(), results.end(), [id_task](const server_task_result_ptr & res) { return res->id == id_task; }), - queue_results.end()); + results.end()); + + waiting.erase(it); } void server_response::remove_waiting_task_ids(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); for (const auto & id_task : id_tasks) { - RES_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting_task_ids.size()); - waiting_task_ids.erase(id_task); + RES_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting.size()); + waiting.erase(id_task); + } +} + +server_response::waiter_ptr server_response::find_waiter(const std::unordered_set & id_tasks) const { + for (const auto & id_task : id_tasks) { + auto it = waiting.find(id_task); + if (it != waiting.end()) { + return it->second; + } } + + return nullptr; } server_task_result_ptr server_response::recv(const std::unordered_set & id_tasks) { + std::unique_lock lock(mutex_results); + + auto w = find_waiter(id_tasks); + GGML_ASSERT(w && "recv() called for task ids that are not in the waiting list"); + while (true) { - std::unique_lock lock(mutex_results); - condition_results.wait(lock, [&]{ - if (!running) { - RES_DBG("%s : queue result stop\n", "recv"); - std::terminate(); // we cannot return here since the caller is HTTP code - } - return !queue_results.empty(); - }); + if (!running) { + RES_DBG("%s : queue result stop\n", "recv"); + std::terminate(); // we cannot return here since the caller is HTTP code + } - for (size_t i = 0; i < queue_results.size(); i++) { - if (id_tasks.find(queue_results[i]->id) != id_tasks.end()) { - server_task_result_ptr res = std::move(queue_results[i]); - queue_results.erase(queue_results.begin() + i); - return res; - } + if (!w->results.empty()) { + server_task_result_ptr res = std::move(w->results.front()); + w->results.pop_front(); + return res; } + + // bounded, so a terminate() that lands after the id was removed from the map still + // gets noticed here + w->cv.wait_for(lock, std::chrono::seconds(1)); } // should never reach here } server_task_result_ptr server_response::recv_with_timeout(const std::unordered_set & id_tasks, int timeout) { - while (true) { - std::unique_lock lock(mutex_results); + std::unique_lock lock(mutex_results); - for (int i = 0; i < (int) queue_results.size(); i++) { - if (id_tasks.find(queue_results[i]->id) != id_tasks.end()) { - server_task_result_ptr res = std::move(queue_results[i]); - queue_results.erase(queue_results.begin() + i); - return res; - } + auto w = find_waiter(id_tasks); + if (!w) { + // the tasks are no longer in the waiting list, so no result can arrive for them. + // wait out the timeout anyway, so the caller sees the poll interval it asked for + // instead of a busy loop + condition_gone.wait_for(lock, std::chrono::seconds(timeout)); + return nullptr; + } + + while (true) { + if (!w->results.empty()) { + server_task_result_ptr res = std::move(w->results.front()); + w->results.pop_front(); + return res; } - std::cv_status cr_res = condition_results.wait_for(lock, std::chrono::seconds(timeout)); + std::cv_status cr_res = w->cv.wait_for(lock, std::chrono::seconds(timeout)); if (!running) { RES_DBG("%s : queue result stop\n", __func__); std::terminate(); // we cannot return here since the caller is HTTP code @@ -481,31 +516,39 @@ void server_response::send(server_task_result_ptr && result) { RES_DBG("sending result for task id = %d\n", result->id); std::unique_lock lock(mutex_results); - for (const auto & id_task : waiting_task_ids) { - if (result->id == id_task) { - RES_DBG("task id = %d pushed to result queue\n", result->id); - queue_results.emplace_back(std::move(result)); - condition_results.notify_all(); - return; - } + auto it = waiting.find(result->id); + if (it == waiting.end()) { + return; } + + RES_DBG("task id = %d pushed to result queue\n", result->id); + + auto & w = *it->second; + + w.results.emplace_back(std::move(result)); + w.cv.notify_one(); } void server_response::broadcast(server_task_result_ptr && result) { std::unique_lock lock(mutex_results); - for (const auto & id_task : waiting_task_ids) { + for (const auto & [id_task, w] : waiting) { RES_DBG("task id = %d pushed to result queue\n", id_task); server_task_result_ptr res_copy(result->clone()); res_copy->id = id_task; // override id with target task id - queue_results.emplace_back(std::move(res_copy)); + w->results.emplace_back(std::move(res_copy)); + w->cv.notify_one(); } - condition_results.notify_all(); } void server_response::terminate() { + std::unique_lock lock(mutex_results); running = false; - condition_results.notify_all(); + for (const auto & [id_task, w] : waiting) { + (void) id_task; + w->cv.notify_all(); + } + condition_gone.notify_all(); } // diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index e17733a743f6..10109d7c2585 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -5,10 +5,12 @@ #include #include #include +#include #include #include -#include +#include #include +#include // struct for managing server tasks // in most cases, use server_response_reader to post new tasks and retrieve results @@ -155,14 +157,36 @@ struct server_response { private: bool running = true; - // for keeping track of all tasks waiting for the result - std::unordered_set waiting_task_ids; + // One waiter per reader, shared by every task id that reader registered in one call. + // Results are queued on the waiter that owns the id, so sending a result wakes only the + // thread that is waiting for it, and that thread finds its result without searching. + // + // Previously there was a single result vector and a single condition variable: every + // result woke every waiting HTTP thread, and each of them re-took the mutex and scanned + // the whole vector before going back to sleep. With N slots generating that is N wakeups + // and N scans per token, i.e. N^2 per decode step, all of it contending with the decode + // thread for the same mutex. + struct waiter { + std::condition_variable cv; + + // FIFO, so results are handed out in the order they were sent, as before + std::deque results; + }; - // the main result queue (using ptr for polymorphism) - std::vector queue_results; + using waiter_ptr = std::shared_ptr; + + // task id --> the waiter that is expecting its results + std::unordered_map waiting; std::mutex mutex_results; - std::condition_variable condition_results; + + // only used to park a reader whose ids are no longer in the waiting list, so that it + // still returns after the timeout it asked for rather than spinning + std::condition_variable condition_gone; + + // all ids registered together share one waiter, so the first hit is the right one + // must be called with mutex_results held + waiter_ptr find_waiter(const std::unordered_set & id_tasks) const; public: // add the id_task to the list of tasks waiting for response From 755fb62336b72bfab73bb0597f1666a6276052e7 Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Sun, 6 Sep 2026 03:54:53 -0700 Subject: [PATCH 02/10] server: only keep per token probabilities when the request asked for them server_slot::generated_token_probs is read in exactly one place, send_final_response(), and only under n_probs > 0. Every other request still pushed a completion_token_output per token, each with a heap allocated string, into a list that grows for the whole generation and is then discarded. The output is unchanged: with n_probs <= 0 nothing ever reads the list. --- tools/server/server-context.cpp | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index a9edbd7be8b4..1b208631c796 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -431,6 +431,13 @@ struct server_slot { return; } + // the buffer is only ever read by send_final_response(), and only when the request + // asked for per-token probabilities. Without them every token still copied a string + // and a vector into a list that grows for the whole generation and is then dropped. + if (task->params.sampling.n_probs <= 0) { + return; + } + generated_token_probs.push_back(token); } From 29fd150c7c6807088ce42feb6da5054649b95c74 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Tue, 8 Sep 2026 18:41:33 -0700 Subject: [PATCH 03/10] server: trim comments in the result queue change --- tools/server/server-context.cpp | 5 ++--- tools/server/server-queue.cpp | 13 ++++--------- tools/server/server-queue.h | 19 ++++--------------- 3 files changed, 10 insertions(+), 27 deletions(-) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 1b208631c796..ef78971d8b1f 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -431,9 +431,8 @@ struct server_slot { return; } - // the buffer is only ever read by send_final_response(), and only when the request - // asked for per-token probabilities. Without them every token still copied a string - // and a vector into a list that grows for the whole generation and is then dropped. + // only send_final_response() reads this, and only with n_probs > 0; otherwise every token + // copied a string and a vector into a list grown for the whole generation, then dropped if (task->params.sampling.n_probs <= 0) { return; } diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index de663ddabf59..2be748cef543 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -397,8 +397,7 @@ void server_response::add_waiting_task_id(int id_task) { void server_response::add_waiting_task_ids(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); - // one waiter for the whole set: these ids belong to a single reader, which waits for any - // of them at a time + // one waiter for the whole set: these ids belong to one reader auto w = std::make_shared(); for (const auto & id_task : id_tasks) { @@ -417,8 +416,7 @@ void server_response::remove_waiting_task_id(int id_task) { return; } - // make sure to clean up all pending results of this task, the waiter may still be held by - // the other ids of the same reader + // the waiter is shared with the reader's other ids, so drop only this task's results auto & results = it->second->results; results.erase( std::remove_if(results.begin(), results.end(), [id_task](const server_task_result_ptr & res) { @@ -467,8 +465,7 @@ server_task_result_ptr server_response::recv(const std::unordered_set & id_ return res; } - // bounded, so a terminate() that lands after the id was removed from the map still - // gets noticed here + // bounded: a terminate() landing after the id left the map is still noticed here w->cv.wait_for(lock, std::chrono::seconds(1)); } @@ -480,9 +477,7 @@ server_task_result_ptr server_response::recv_with_timeout(const std::unordered_s auto w = find_waiter(id_tasks); if (!w) { - // the tasks are no longer in the waiting list, so no result can arrive for them. - // wait out the timeout anyway, so the caller sees the poll interval it asked for - // instead of a busy loop + // no result can arrive now; still honour the timeout so the caller does not busy loop condition_gone.wait_for(lock, std::chrono::seconds(timeout)); return nullptr; } diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index 10109d7c2585..47c40a6c244e 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -157,35 +157,24 @@ struct server_response { private: bool running = true; - // One waiter per reader, shared by every task id that reader registered in one call. - // Results are queued on the waiter that owns the id, so sending a result wakes only the - // thread that is waiting for it, and that thread finds its result without searching. - // - // Previously there was a single result vector and a single condition variable: every - // result woke every waiting HTTP thread, and each of them re-took the mutex and scanned - // the whole vector before going back to sleep. With N slots generating that is N wakeups - // and N scans per token, i.e. N^2 per decode step, all of it contending with the decode - // thread for the same mutex. + // One waiter per reader, shared by every id it registered in one call. A single shared vector + // plus one cv instead costs N wakeups and N scans per token, N^2 per decode step. struct waiter { std::condition_variable cv; - // FIFO, so results are handed out in the order they were sent, as before std::deque results; }; using waiter_ptr = std::shared_ptr; - // task id --> the waiter that is expecting its results std::unordered_map waiting; std::mutex mutex_results; - // only used to park a reader whose ids are no longer in the waiting list, so that it - // still returns after the timeout it asked for rather than spinning + // parks a reader whose ids left the waiting list, so it honours its timeout std::condition_variable condition_gone; - // all ids registered together share one waiter, so the first hit is the right one - // must be called with mutex_results held + // ids registered together share one waiter, so the first hit is the right one. mutex_results held. waiter_ptr find_waiter(const std::unordered_set & id_tasks) const; public: From 18abc68be5df0077a7505401f38310400d71b849 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Tue, 8 Sep 2026 21:08:00 -0700 Subject: [PATCH 04/10] server: do not abort the process when a request's ids have left the waiting list recv() asserted that the waiter exists. GGML_ASSERT is GGML_ABORT, and recv() runs on the HTTP thread, so a single request whose ids had been dropped by a cancel or a cleanup took the whole server down for every other client. Before the per-waiter queues this case was harmless: recv() waited on a condition that no longer fires for those ids, which parks that one connection and nothing else. The assert came in with the per-waiter lookup in this branch, so it is a regression this branch introduced rather than existing behaviour. Restore the old outcome. The lookup moves inside the loop so a waiter re-added while we wait is picked up rather than waited out, and shutdown is still noticed because the running check stays at the top. recv_with_timeout() already tolerated the missing waiter and is unchanged. tests/test-server-queue.cpp covers it: recv() on ids that were never registered must leave the process alive. On the parent commit the test aborts with SIGABRT at server-queue.cpp:454 while main is only sleeping, which is the defect exactly. The target is behind LLAMA_BUILD_TESTS and needs no model. --- tools/server/CMakeLists.txt | 12 +++++ tools/server/server-queue.cpp | 16 +++++-- tools/server/tests/test-server-queue.cpp | 56 ++++++++++++++++++++++++ 3 files changed, 81 insertions(+), 3 deletions(-) create mode 100644 tools/server/tests/test-server-queue.cpp diff --git a/tools/server/CMakeLists.txt b/tools/server/CMakeLists.txt index 280bd9e19dca..5ac5400f94fe 100644 --- a/tools/server/CMakeLists.txt +++ b/tools/server/CMakeLists.txt @@ -63,3 +63,15 @@ install(TARGETS ${TARGET} RUNTIME) target_link_libraries(${TARGET} PRIVATE llama-server-impl) target_compile_features(${TARGET} PRIVATE cxx_std_17) + +# server-queue unit test: needs no model, so it stays out of the default build + +if (LLAMA_BUILD_TESTS AND NOT CMAKE_CROSSCOMPILING) + set(TARGET test-server-queue) + + add_executable(${TARGET} tests/test-server-queue.cpp) + target_link_libraries(${TARGET} PRIVATE server-context ${CMAKE_THREAD_LIBS_INIT}) + target_compile_features(${TARGET} PRIVATE cxx_std_17) + + add_test(NAME ${TARGET} COMMAND ${TARGET}) +endif() diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 2be748cef543..0c9d37adb64a 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -450,15 +450,25 @@ server_response::waiter_ptr server_response::find_waiter(const std::unordered_se server_task_result_ptr server_response::recv(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); - auto w = find_waiter(id_tasks); - GGML_ASSERT(w && "recv() called for task ids that are not in the waiting list"); - while (true) { if (!running) { RES_DBG("%s : queue result stop\n", "recv"); std::terminate(); // we cannot return here since the caller is HTTP code } + // The waiter can be absent, so this cannot assert. A cancel or a cleanup drops the ids + // between the caller posting them and arriving here, and recv() runs on the HTTP + // thread: aborting there turns one stuck request into a dead server for every other + // client. Before the per-waiter queues this waited on a condition that no longer fires + // for these ids, which blocks this one connection and nothing else, so that is what it + // does here too. The lookup is inside the loop rather than above it because a waiter + // re-added while we wait should be picked up instead of waited out. + auto w = find_waiter(id_tasks); + if (w == nullptr) { + condition_gone.wait_for(lock, std::chrono::seconds(1)); + continue; + } + if (!w->results.empty()) { server_task_result_ptr res = std::move(w->results.front()); w->results.pop_front(); diff --git a/tools/server/tests/test-server-queue.cpp b/tools/server/tests/test-server-queue.cpp new file mode 100644 index 000000000000..74c7a1676cf2 --- /dev/null +++ b/tools/server/tests/test-server-queue.cpp @@ -0,0 +1,56 @@ +#include "server-queue.h" + +#include +#include +#include +#include +#include +#include + +// recv() can be called with ids that are not in the waiting list: a cancel or a cleanup drops +// them between the caller posting and the caller arriving. That has to park this one connection +// and nothing else, which is what the unbounded wait did before the per-waiter queues. +// +// It must not assert. GGML_ASSERT is GGML_ABORT, and recv() runs on the HTTP thread, so a +// single dropped request would take the whole server down for every other client. The parent +// commit dies inside recv() here, in well under the 2.5 s this waits. +int main() { + server_response res; + + std::atomic returned{false}; + + std::thread parked([&] { + server_task_result_ptr r = res.recv(std::unordered_set{4242}); + (void) r; + returned.store(true); + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(2500)); + + if (returned.load()) { + fprintf(stderr, "FAIL: recv() returned for ids that are not in the waiting list\n"); + return 1; + } + + // the timeout form already tolerated this, and must keep doing so promptly + const auto t0 = std::chrono::steady_clock::now(); + server_task_result_ptr none = res.recv_with_timeout(std::unordered_set{4243}, 1); + const auto waited = std::chrono::steady_clock::now() - t0; + + if (none != nullptr) { + fprintf(stderr, "FAIL: recv_with_timeout() invented a result\n"); + return 1; + } + if (waited > std::chrono::seconds(5)) { + fprintf(stderr, "FAIL: recv_with_timeout() did not honour its timeout\n"); + return 1; + } + + printf("OK: a dropped request parks its own caller and leaves the server up\n"); + + // parked is still inside recv() by design: terminate() would make it std::terminate(), + // which is the documented behaviour for an HTTP caller, so leave without joining it. + parked.detach(); + fflush(stdout); + _Exit(0); +} From 170a675fd160b3a9914e340c762f311c84f50b76 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 9 Sep 2026 01:41:47 -0700 Subject: [PATCH 05/10] server: widen the result queue test to isolation, teardown, broadcast and churn --- tools/server/CMakeLists.txt | 8 + tools/server/tests/test-server-response.cpp | 321 ++++++++++++++++++++ 2 files changed, 329 insertions(+) create mode 100644 tools/server/tests/test-server-response.cpp diff --git a/tools/server/CMakeLists.txt b/tools/server/CMakeLists.txt index 5ac5400f94fe..700f08fd2cd4 100644 --- a/tools/server/CMakeLists.txt +++ b/tools/server/CMakeLists.txt @@ -74,4 +74,12 @@ if (LLAMA_BUILD_TESTS AND NOT CMAKE_CROSSCOMPILING) target_compile_features(${TARGET} PRIVATE cxx_std_17) add_test(NAME ${TARGET} COMMAND ${TARGET}) + + set(TARGET test-server-response) + + add_executable(${TARGET} tests/test-server-response.cpp) + target_link_libraries(${TARGET} PRIVATE server-context ${CMAKE_THREAD_LIBS_INIT}) + target_compile_features(${TARGET} PRIVATE cxx_std_17) + + add_test(NAME ${TARGET} COMMAND ${TARGET}) endif() diff --git a/tools/server/tests/test-server-response.cpp b/tools/server/tests/test-server-response.cpp new file mode 100644 index 000000000000..40034590d3d4 --- /dev/null +++ b/tools/server/tests/test-server-response.cpp @@ -0,0 +1,321 @@ +// Unit test for server_response, the result queue between the decode loop and the HTTP threads. +// +// It exercises the public API directly, so the awkward cases are injected rather than waited +// for: a reader whose ids were dropped between posting and receiving, a send that races a +// cancel, per id and bulk teardown, broadcast, and concurrent registration and removal. +// +// Run with "leak " to measure what the queue retains over n reader lifecycles, each leaving +// one result queued at teardown, which is what a client disconnect during generation does. + +#include "server-queue.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static int g_fail = 0; + +static void check(bool ok, const char * name, const char * detail = "") { + printf("%-46s %s %s\n", name, ok ? "PASS" : "FAIL", detail); + if (!ok) { g_fail++; } +} + +// minimal concrete result so we can put things on the queue without a model +struct fake_result : server_task_result { + int payload = 0; + bool stop = false; + fake_result(int id_, int payload_, bool stop_) : payload(payload_), stop(stop_) { id = id_; } + bool is_stop() override { return stop; } + json to_json() override { return json{{"payload", payload}}; } + server_task_result * clone() const override { return new fake_result(*this); } +}; + +static server_task_result_ptr mk(int id, int payload, bool stop = true) { + return server_task_result_ptr(new fake_result(id, payload, stop)); +} + +static int payload_of(const server_task_result_ptr & p) { + return p ? static_cast(p.get())->payload : -1; +} + +using ms = std::chrono::milliseconds; + +// --------------------------------------------------------------------------- + +// a result for an id that has left the waiting list is dropped, silently, no crash +static void t_send_to_absent_id() { + server_response res; + res.send(mk(1, 100)); // never registered + res.add_waiting_task_id(2); + res.remove_waiting_task_id(2); + res.send(mk(2, 200)); // registered then removed + res.add_waiting_task_id(3); + res.send(mk(3, 300)); + auto got = res.recv_with_timeout({3}, 1); + check(got != nullptr && payload_of(got) == 300, "send to absent id is dropped, live id still delivered"); +} + +// FIFO order per reader is what scanning the shared vector from the front used to give +static void t_fifo_order() { + server_response res; + res.add_waiting_task_id(7); + for (int i = 0; i < 32; i++) { res.send(mk(7, i, i == 31)); } + bool ok = true; + for (int i = 0; i < 32; i++) { + auto r = res.recv_with_timeout({7}, 1); + if (payload_of(r) != i) { ok = false; break; } + } + check(ok, "FIFO order preserved for a single reader"); +} + +// two independent readers must not see each other's results +static void t_reader_isolation() { + server_response res; + res.add_waiting_task_ids({10, 11}); + res.add_waiting_task_ids({20, 21}); + res.send(mk(20, 2000)); + res.send(mk(10, 1000)); + res.send(mk(21, 2100)); + res.send(mk(11, 1100)); + + std::vector a, b; + for (int i = 0; i < 2; i++) { a.push_back(payload_of(res.recv_with_timeout({10, 11}, 1))); } + for (int i = 0; i < 2; i++) { b.push_back(payload_of(res.recv_with_timeout({20, 21}, 1))); } + const bool ok = a.size() == 2 && b.size() == 2 && + (a[0] == 1000 && a[1] == 1100) && (b[0] == 2000 && b[1] == 2100); + char d[128]; + snprintf(d, sizeof(d), "a=[%d,%d] b=[%d,%d]", a[0], a[1], b[0], b[1]); + check(ok, "two readers do not steal each other's results", d); + check(res.recv_with_timeout({10, 11}, 1) == nullptr, "reader A drained, no extra result"); +} + +// per-id removal must drop only that id's results +static void t_partial_removal() { + server_response res; + res.add_waiting_task_ids({30, 31}); + res.send(mk(30, 3000)); + res.send(mk(31, 3100)); + res.remove_waiting_task_id(30); + auto r = res.recv_with_timeout({31}, 1); + check(r != nullptr && payload_of(r) == 3100, "removing one id keeps the sibling's result"); + check(res.recv_with_timeout({31}, 1) == nullptr, "the removed id's result is gone"); +} + +// bulk removal then a late send: nothing delivered, nothing leaked, no use after free +static void t_bulk_removal() { + server_response res; + res.add_waiting_task_ids({40, 41, 42}); + res.send(mk(40, 4000)); + res.remove_waiting_task_ids({40, 41, 42}); + res.send(mk(41, 4100)); + check(res.recv_with_timeout({40, 41, 42}, 1) == nullptr, "bulk removal drops queued and late results"); +} + +// broadcast: one copy per registered id, id overridden +static void t_broadcast() { + server_response res; + res.add_waiting_task_ids({50, 51}); + res.add_waiting_task_id(60); + res.broadcast(mk(-1, 9999)); + auto a1 = res.recv_with_timeout({50, 51}, 1); + auto a2 = res.recv_with_timeout({50, 51}, 1); + auto a3 = res.recv_with_timeout({50, 51}, 1); + auto b1 = res.recv_with_timeout({60}, 1); + auto b2 = res.recv_with_timeout({60}, 1); + const bool ok = a1 && a2 && !a3 && b1 && !b2 && + payload_of(a1) == 9999 && payload_of(b1) == 9999; + check(ok, "broadcast delivers one copy per registered id"); + const bool ids_ok = a1 && a2 && (a1->id == 50 || a1->id == 51) && (a2->id == 50 || a2->id == 51) && a1->id != a2->id && b1 && b1->id == 60; + check(ids_ok, "broadcast overrides the result id per target"); +} + +// lost wakeup: park a reader on ids that do not exist yet, then create them and send. +// Both arms must deliver; the head's condition_gone poll bounds the delay. +static void t_late_registration() { + server_response res; + std::atomic got{-2}; + std::atomic done{false}; + std::thread th([&] { + for (int i = 0; i < 60; i++) { + auto r = res.recv_with_timeout({70}, 1); + if (r) { got.store(payload_of(r)); break; } + } + done.store(true); + }); + std::this_thread::sleep_for(ms(300)); + res.add_waiting_task_id(70); + res.send(mk(70, 7000)); + const auto t0 = std::chrono::steady_clock::now(); + while (!done.load() && std::chrono::steady_clock::now() - t0 < std::chrono::seconds(10)) { + std::this_thread::sleep_for(ms(10)); + } + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + th.join(); + char d[64]; snprintf(d, sizeof(d), "%lldms", (long long) waited); + check(got.load() == 7000, "id registered after the reader parked is still served", d); +} + +// a reader whose ids were dropped parks itself and nothing else. +// It must not assert: GGML_ASSERT is GGML_ABORT, and recv() runs on the HTTP thread, so one +// dropped request would take the whole server down for every other client. +static void t_parked_reader_does_not_abort() { + static server_response res; // static: the parked thread outlives this function + std::atomic returned{false}; + std::thread parked([&] { + auto r = res.recv(std::unordered_set{4242}); + (void) r; + returned.store(true); + }); + std::this_thread::sleep_for(ms(2500)); + check(!returned.load(), "recv() on dropped ids parks instead of returning garbage"); + + // and the rest of the queue keeps working while that thread is parked + res.add_waiting_task_id(80); + res.send(mk(80, 8000)); + auto r = res.recv_with_timeout({80}, 2); + check(r != nullptr && payload_of(r) == 8000, "other readers unaffected by a parked reader"); + parked.detach(); +} + +// recv_with_timeout honours its timeout when the waiter exists but is empty +static void t_timeout_honoured() { + server_response res; + res.add_waiting_task_id(90); + const auto t0 = std::chrono::steady_clock::now(); + auto r = res.recv_with_timeout({90}, 1); + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + char d[64]; snprintf(d, sizeof(d), "%lldms", (long long) waited); + check(r == nullptr && waited >= 900 && waited < 5000, "recv_with_timeout honours its timeout", d); +} + +// concurrent churn: producers, consumers, registration and teardown all at once. +// This is the case ThreadSanitizer is pointed at. +static void t_stress() { + server_response res; + const int n_readers = 16; + const int n_msgs = 200; + std::atomic received{0}; + std::atomic go{false}; + + std::vector readers; + for (int r = 0; r < n_readers; r++) { + readers.emplace_back([&, r] { + while (!go.load()) { std::this_thread::yield(); } + const int base_id = 1000 + r * 10; + std::unordered_set ids{base_id, base_id + 1}; + res.add_waiting_task_ids(ids); + int seen = 0; + while (seen < n_msgs) { + auto p = res.recv_with_timeout(ids, 1); + if (!p) { break; } + seen++; + received.fetch_add(1); + } + res.remove_waiting_task_ids(ids); + }); + } + + std::vector writers; + for (int w = 0; w < 4; w++) { + writers.emplace_back([&, w] { + while (!go.load()) { std::this_thread::yield(); } + for (int i = w; i < n_msgs * n_readers; i += 4) { + const int r = (i / n_msgs) % n_readers; + res.send(mk(1000 + r * 10 + (i % 2), i, false)); + if ((i & 63) == 0) { std::this_thread::sleep_for(ms(1)); } + } + }); + } + + // a churn thread that registers and drops ids nobody waits for + std::thread churn([&] { + while (!go.load()) { std::this_thread::yield(); } + for (int i = 0; i < 2000; i++) { + res.add_waiting_task_id(500000 + i); + res.send(mk(500000 + i, i, false)); + res.remove_waiting_task_id(500000 + i); + } + }); + + go.store(true); + for (auto & t : writers) { t.join(); } + for (auto & t : readers) { t.join(); } + churn.join(); + char d[64]; snprintf(d, sizeof(d), "received=%d", received.load()); + check(received.load() > 0, "concurrent send/recv/register/remove churn survives", d); +} + +// API hazard probe: recv() with a strict subset of the ids that share one waiter. +// Not reachable from server_response_reader (it always passes the whole set), reported +// as a latent sharp edge rather than a defect. +static void t_subset_recv_probe() { + server_response res; + res.add_waiting_task_ids({200, 201}); + res.send(mk(201, 2010)); + auto r = res.recv_with_timeout({200}, 1); + printf("%-46s %s (id=%d)\n", "PROBE recv() with a subset of a shared waiter", + r == nullptr ? "returns nullptr (id filtered)" : "RETURNS THE SIBLING'S RESULT", + r ? r->id : -1); +} + +static long rss_kb() { + FILE * f = fopen("/proc/self/status", "r"); + if (!f) { return -1; } + char line[256]; + long v = -1; + while (fgets(line, sizeof(line), f)) { + if (strncmp(line, "VmRSS:", 6) == 0) { sscanf(line + 6, "%ld", &v); break; } + } + fclose(f); + return v; +} + +// Isolated measurement of what the result queue itself retains. One reader lifecycle per +// iteration: register two ids, receive one result, leave one result queued (which is what a +// disconnect during generation does), then tear the reader down the way stop() does. +static void t_leak(long n) { + server_response res; + const long rss0 = rss_kb(); + for (long i = 0; i < n; i++) { + const int a = 100000 + (int) (i * 2); + const int b = a + 1; + res.add_waiting_task_ids({a, b}); + res.send(mk(a, 1)); + res.send(mk(b, 2)); // left unconsumed on purpose + auto got = res.recv_with_timeout({a, b}, 1); + (void) got; + res.remove_waiting_task_ids({a, b}); // exactly what server_response_reader::stop() does + } + const long rss1 = rss_kb(); + printf("LEAK n=%ld rss_start=%ld kB rss_end=%ld kB growth=%ld kB (%.3f kB per reader)\n", + n, rss0, rss1, rss1 - rss0, (double) (rss1 - rss0) / (double) n); +} + +int main(int argc, char ** argv) { + if (argc > 1 && strcmp(argv[1], "leak") == 0) { + t_leak(argc > 2 ? atol(argv[2]) : 200000); + fflush(stdout); + _Exit(0); + } + t_send_to_absent_id(); + t_fifo_order(); + t_reader_isolation(); + t_partial_removal(); + t_bulk_removal(); + t_broadcast(); + t_late_registration(); + t_timeout_honoured(); + t_stress(); + t_subset_recv_probe(); + t_parked_reader_does_not_abort(); + + printf("\nRESULT queue failures=%d\n", g_fail); + fflush(stdout); + _Exit(g_fail == 0 ? 0 : 1); // a thread is parked in recv() by design +} From cdb0ecb8e830628de08f06d4587ddccddf5673ea Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 9 Sep 2026 03:25:23 -0700 Subject: [PATCH 06/10] server: filter results by requested id, wake on registration, honour terminate() Three review findings, all of them regressions against the parent of this branch that its own test masked: - recv() and recv_with_timeout() popped the front of a shared waiter's queue without checking the id, so a caller asking for one id of a registered set could be handed a sibling's result. take_result() now scans for a requested id, in arrival order, which is what scanning the shared vector did. The front normally matches. - a timed receive that started before its ids were registered waited on condition_gone, which nothing fired on registration, so a result arriving during the call was missed and the caller reported a spurious timeout. add_waiting_task_id(s) now notifies it. The old shared queue woke such a caller because every send notified the one condition. - the absent-waiter branch of recv_with_timeout() returned nullptr without rechecking running, so terminate() was ignored there and a caller whose stop predicate stays false could loop forever. Both recv paths now recheck it, and one deadline covers the whole call so waiting for a registration and then for a result cannot add up to twice the timeout the caller asked for. --- tools/server/server-queue.cpp | 77 +++++++++++++++------ tools/server/server-queue.h | 3 + tools/server/tests/test-server-response.cpp | 74 +++++++++++++++++--- 3 files changed, 124 insertions(+), 30 deletions(-) diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 0c9d37adb64a..ddb2789c41e0 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -392,6 +392,9 @@ void server_response::add_waiting_task_id(int id_task) { RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting.size()); waiting.emplace(id_task, std::make_shared()); + + // a reader may already be parked on these ids waiting for exactly this + condition_gone.notify_all(); } void server_response::add_waiting_task_ids(const std::unordered_set & id_tasks) { @@ -404,6 +407,9 @@ void server_response::add_waiting_task_ids(const std::unordered_set & id_ta RES_DBG("add task %d to waiting list. current waiting = %d (before add)\n", id_task, (int) waiting.size()); waiting.emplace(id_task, w); } + + // a reader may already be parked on these ids waiting for exactly this + condition_gone.notify_all(); } void server_response::remove_waiting_task_id(int id_task) { @@ -447,6 +453,29 @@ server_response::waiter_ptr server_response::find_waiter(const std::unordered_se return nullptr; } +// A waiter is shared by every id its reader registered in one call, so its queue can hold a +// sibling's result. Return only an id the caller asked for, in arrival order, which is what +// scanning the shared vector did. The front normally matches, so this is O(1) in practice. +server_task_result_ptr server_response::take_result(const std::unordered_set & id_tasks) { + for (const auto & id_task : id_tasks) { + auto it = waiting.find(id_task); + if (it == waiting.end()) { + continue; + } + + auto & results = it->second->results; + for (auto rit = results.begin(); rit != results.end(); ++rit) { + if (id_tasks.find((*rit)->id) != id_tasks.end()) { + server_task_result_ptr res = std::move(*rit); + results.erase(rit); + return res; + } + } + } + + return nullptr; +} + server_task_result_ptr server_response::recv(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); @@ -456,6 +485,11 @@ server_task_result_ptr server_response::recv(const std::unordered_set & id_ std::terminate(); // we cannot return here since the caller is HTTP code } + server_task_result_ptr res = take_result(id_tasks); + if (res != nullptr) { + return res; + } + // The waiter can be absent, so this cannot assert. A cancel or a cleanup drops the ids // between the caller posting them and arriving here, and recv() runs on the HTTP // thread: aborting there turns one stuck request into a dead server for every other @@ -465,16 +499,11 @@ server_task_result_ptr server_response::recv(const std::unordered_set & id_ // re-added while we wait should be picked up instead of waited out. auto w = find_waiter(id_tasks); if (w == nullptr) { + // registration and terminate() both fire condition_gone; the timeout is only a backstop condition_gone.wait_for(lock, std::chrono::seconds(1)); continue; } - if (!w->results.empty()) { - server_task_result_ptr res = std::move(w->results.front()); - w->results.pop_front(); - return res; - } - // bounded: a terminate() landing after the id left the map is still noticed here w->cv.wait_for(lock, std::chrono::seconds(1)); } @@ -485,26 +514,34 @@ server_task_result_ptr server_response::recv(const std::unordered_set & id_ server_task_result_ptr server_response::recv_with_timeout(const std::unordered_set & id_tasks, int timeout) { std::unique_lock lock(mutex_results); - auto w = find_waiter(id_tasks); - if (!w) { - // no result can arrive now; still honour the timeout so the caller does not busy loop - condition_gone.wait_for(lock, std::chrono::seconds(timeout)); - return nullptr; - } + // one deadline for the whole call: waiting for a registration and then for a result must not + // add up to twice the timeout the caller asked for + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(timeout); while (true) { - if (!w->results.empty()) { - server_task_result_ptr res = std::move(w->results.front()); - w->results.pop_front(); - return res; - } - - std::cv_status cr_res = w->cv.wait_for(lock, std::chrono::seconds(timeout)); if (!running) { RES_DBG("%s : queue result stop\n", __func__); std::terminate(); // we cannot return here since the caller is HTTP code } - if (cr_res == std::cv_status::timeout) { + + server_task_result_ptr res = take_result(id_tasks); + if (res != nullptr) { + return res; + } + + auto w = find_waiter(id_tasks); + + // The ids are not registered yet, or not any more. Wait on condition_gone rather than + // sleeping out the timeout: add_waiting_task_id(s) fires it, so a result that arrives + // during this call is still seen, which is what the single shared condition used to give. + // terminate() fires it too, so it is honoured here as well as on the waiter's own cv. + std::condition_variable & cv = w == nullptr ? condition_gone : w->cv; + + if (cv.wait_until(lock, deadline) == std::cv_status::timeout) { + if (!running) { + RES_DBG("%s : queue result stop\n", __func__); + std::terminate(); // we cannot return here since the caller is HTTP code + } return nullptr; } } diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index 47c40a6c244e..debe208b13a0 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -177,6 +177,9 @@ struct server_response { // ids registered together share one waiter, so the first hit is the right one. mutex_results held. waiter_ptr find_waiter(const std::unordered_set & id_tasks) const; + // pop the oldest queued result whose id the caller asked for. mutex_results held. + server_task_result_ptr take_result(const std::unordered_set & id_tasks); + public: // add the id_task to the list of tasks waiting for response void add_waiting_task_id(int id_task); diff --git a/tools/server/tests/test-server-response.cpp b/tools/server/tests/test-server-response.cpp index 40034590d3d4..8b8109443713 100644 --- a/tools/server/tests/test-server-response.cpp +++ b/tools/server/tests/test-server-response.cpp @@ -19,6 +19,9 @@ #include #include +#include +#include + static int g_fail = 0; static void check(bool ok, const char * name, const char * detail = "") { @@ -162,8 +165,7 @@ static void t_late_registration() { } // a reader whose ids were dropped parks itself and nothing else. -// It must not assert: GGML_ASSERT is GGML_ABORT, and recv() runs on the HTTP thread, so one -// dropped request would take the whole server down for every other client. +// The parent commit of the head asserted here, which is GGML_ABORT on the HTTP thread. static void t_parked_reader_does_not_abort() { static server_response res; // static: the parked thread outlives this function std::atomic returned{false}; @@ -251,17 +253,67 @@ static void t_stress() { check(received.load() > 0, "concurrent send/recv/register/remove churn survives", d); } -// API hazard probe: recv() with a strict subset of the ids that share one waiter. -// Not reachable from server_response_reader (it always passes the whole set), reported -// as a latent sharp edge rather than a defect. -static void t_subset_recv_probe() { + +// A single timed receive that starts before the ids exist must still return a result that +// arrives during the call. The shared queue used to notify one condition on every send, so a +// parked receiver woke; per waiter queues have to notify registration explicitly or the caller +// sleeps out its whole timeout and reports a spurious nullptr. +static void t_single_timed_recv_before_registration() { + server_response res; + std::atomic started{false}; + std::thread producer([&] { + while (!started.load()) { std::this_thread::yield(); } + std::this_thread::sleep_for(ms(200)); + res.add_waiting_task_id(300); + res.send(mk(300, 3000)); + }); + started.store(true); + const auto t0 = std::chrono::steady_clock::now(); + auto r = res.recv_with_timeout({300}, 5); // ONE call, not a retry loop + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + producer.join(); + char d[80]; snprintf(d, sizeof(d), "%lldms, %s", (long long) waited, r ? "got result" : "nullptr"); + check(r != nullptr && payload_of(r) == 3000 && waited < 4000, + "one timed recv sees a result that arrives while it waits", d); +} + +// terminate() has to be honoured even when the caller is parked on ids that are not registered. +// recv_with_timeout() promises std::terminate() there, because the caller is HTTP code that +// cannot return. Run in a child: the correct outcome is that the child dies. +static void t_terminate_while_parked_on_absent_ids() { + fflush(stdout); + pid_t pid = fork(); + if (pid == 0) { + auto * res = new server_response(); + std::thread killer([res] { + std::this_thread::sleep_for(ms(300)); + res->terminate(); + }); + auto r = res->recv_with_timeout({9999}, 5); + killer.join(); + // reaching here at all means terminate() was ignored + _Exit(r == nullptr ? 20 : 21); + } + int status = 0; + waitpid(pid, &status, 0); + const bool died = WIFSIGNALED(status); + char d[96]; + if (died) { snprintf(d, sizeof(d), "child died on signal %d", WTERMSIG(status)); } + else { snprintf(d, sizeof(d), "child returned %d, terminate() ignored", WEXITSTATUS(status)); } + check(died, "terminate() is honoured while parked on absent ids", d); +} + +// A result for a sibling id must not be handed to a caller that did not ask for it. +static void t_subset_recv_is_filtered() { server_response res; res.add_waiting_task_ids({200, 201}); res.send(mk(201, 2010)); auto r = res.recv_with_timeout({200}, 1); - printf("%-46s %s (id=%d)\n", "PROBE recv() with a subset of a shared waiter", - r == nullptr ? "returns nullptr (id filtered)" : "RETURNS THE SIBLING'S RESULT", - r ? r->id : -1); + char d[64]; snprintf(d, sizeof(d), "id=%d", r ? r->id : -1); + check(r == nullptr, "recv() does not return a result for an id it was not asked for", d); + // and the sibling's result is still there for the caller that does ask + auto r2 = res.recv_with_timeout({200, 201}, 1); + check(r2 != nullptr && r2->id == 201, "the sibling's result is still delivered to its own reader"); } static long rss_kb() { @@ -312,7 +364,9 @@ int main(int argc, char ** argv) { t_late_registration(); t_timeout_honoured(); t_stress(); - t_subset_recv_probe(); + t_single_timed_recv_before_registration(); + t_terminate_while_parked_on_absent_ids(); + t_subset_recv_is_filtered(); t_parked_reader_does_not_abort(); printf("\nRESULT queue failures=%d\n", g_fail); From 2af4f90475ad03ba9ca83f4b1fd7a509cf549fdf Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 9 Sep 2026 03:44:49 -0700 Subject: [PATCH 07/10] server: wake every subset receiver, and keep the response test buildable on Windows Two more review findings on the previous commit. Filtering results by the caller's ids made notify_one() insufficient: two threads taking disjoint subsets of one add_waiting_task_ids() call share a waiter, so waking one of them can wake the thread whose id has no result, which sleeps again while the thread whose result is queued waits out its timeout. Measured at 2800 ms and a null result with four sibling receivers parked ahead of the one being sent to. The waiter's cv is one reader's own condition, not the single global one the shared vector used, so notify_all() on it is still one wakeup in the common case of one thread per reader. The terminate() assertion needs fork() to observe that the caller terminates, which is not available with the Windows toolchain, and LLAMA_BUILD_TESTS defaults on for a standalone build. Guarded at source rather than excluding the target, so Windows keeps 18 of the 19 assertions. Verified by preprocessing the file with and without _WIN32: waitpid, WIFSIGNALED and pid_t are all absent under _WIN32 and the skip branch is present, and the reverse holds without it. I could not compile or run it on Windows. --- tools/server/server-queue.cpp | 10 +++- tools/server/server-queue.h | 3 +- tools/server/tests/test-server-response.cpp | 55 ++++++++++++++++++++- 3 files changed, 63 insertions(+), 5 deletions(-) diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index ddb2789c41e0..62cec87743b5 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -569,7 +569,13 @@ void server_response::send(server_task_result_ptr && result) { auto & w = *it->second; w.results.emplace_back(std::move(result)); - w.cv.notify_one(); + + // notify_all, not notify_one: results are filtered by id, so waking a single waiter can wake + // one taking a disjoint subset of this reader's ids, which finds nothing and sleeps again + // while the reader whose result this is stays asleep. This is one reader's own condition, + // not the single global one the shared vector used, so it is still O(1) in the common case + // of one thread per reader. + w.cv.notify_all(); } void server_response::broadcast(server_task_result_ptr && result) { @@ -579,7 +585,7 @@ void server_response::broadcast(server_task_result_ptr && result) { server_task_result_ptr res_copy(result->clone()); res_copy->id = id_task; // override id with target task id w->results.emplace_back(std::move(res_copy)); - w->cv.notify_one(); + w->cv.notify_all(); } } diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index debe208b13a0..dd00519070e3 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -158,7 +158,8 @@ struct server_response { bool running = true; // One waiter per reader, shared by every id it registered in one call. A single shared vector - // plus one cv instead costs N wakeups and N scans per token, N^2 per decode step. + // plus one cv instead costs N wakeups and N scans per token, N^2 per decode step. A send + // notifies this reader's own cv, so the wakeup is O(1) even though it is a notify_all. struct waiter { std::condition_variable cv; diff --git a/tools/server/tests/test-server-response.cpp b/tools/server/tests/test-server-response.cpp index 8b8109443713..22a5d0ed199e 100644 --- a/tools/server/tests/test-server-response.cpp +++ b/tools/server/tests/test-server-response.cpp @@ -19,8 +19,10 @@ #include #include -#include -#include +#ifndef _WIN32 +# include +# include +#endif static int g_fail = 0; @@ -281,6 +283,11 @@ static void t_single_timed_recv_before_registration() { // recv_with_timeout() promises std::terminate() there, because the caller is HTTP code that // cannot return. Run in a child: the correct outcome is that the child dies. static void t_terminate_while_parked_on_absent_ids() { +#ifdef _WIN32 + // needs fork(): the correct outcome is that the caller terminates, which cannot be asserted + // in-process. The behaviour itself is not platform specific. + printf("%-46s SKIP (needs fork())\n", "terminate() is honoured while parked on absent ids"); +#else fflush(stdout); pid_t pid = fork(); if (pid == 0) { @@ -301,6 +308,7 @@ static void t_terminate_while_parked_on_absent_ids() { if (died) { snprintf(d, sizeof(d), "child died on signal %d", WTERMSIG(status)); } else { snprintf(d, sizeof(d), "child returned %d, terminate() ignored", WEXITSTATUS(status)); } check(died, "terminate() is honoured while parked on absent ids", d); +#endif } // A result for a sibling id must not be handed to a caller that did not ask for it. @@ -316,7 +324,49 @@ static void t_subset_recv_is_filtered() { check(r2 != nullptr && r2->id == 201, "the sibling's result is still delivered to its own reader"); } + +// Two readers taking disjoint subsets of one registration share a waiter, so waking only one of +// them can wake the wrong one: it finds nothing matching, sleeps again, and the reader whose +// result is actually queued sits there until its timeout. The shared condition used to +// notify_all(), so every subset receiver got a look. +static void t_subset_receivers_are_all_woken() { + server_response res; + res.add_waiting_task_ids({400, 401}); + + std::atomic parked{0}; + std::atomic got_a{-1}; + std::vector others; + + // four readers waiting on the sibling id park first, so a single notify picks one of them + for (int i = 0; i < 4; i++) { + others.emplace_back([&] { + parked.fetch_add(1); + auto r = res.recv_with_timeout({401}, 3); + (void) r; + }); + } + while (parked.load() < 4) { std::this_thread::yield(); } + std::this_thread::sleep_for(ms(200)); + + std::thread reader_a([&] { + auto r = res.recv_with_timeout({400}, 3); + got_a.store(payload_of(r)); + }); + std::this_thread::sleep_for(ms(200)); + + const auto t0 = std::chrono::steady_clock::now(); + res.send(mk(400, 4000)); + reader_a.join(); + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + for (auto & t : others) { t.join(); } + + char d[80]; snprintf(d, sizeof(d), "%lldms, payload=%d", (long long) waited, got_a.load()); + check(got_a.load() == 4000 && waited < 2500, + "a subset receiver is woken even when siblings wait too", d); +} + static long rss_kb() { + // Linux only; returns -1 elsewhere, and only the optional "leak" mode uses it FILE * f = fopen("/proc/self/status", "r"); if (!f) { return -1; } char line[256]; @@ -367,6 +417,7 @@ int main(int argc, char ** argv) { t_single_timed_recv_before_registration(); t_terminate_while_parked_on_absent_ids(); t_subset_recv_is_filtered(); + t_subset_receivers_are_all_woken(); t_parked_reader_does_not_abort(); printf("\nRESULT queue failures=%d\n", g_fail); From 4e865432b5561e5ccf46178fa7e66e5d05df4cb6 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 9 Sep 2026 04:02:43 -0700 Subject: [PATCH 08/10] server: wake a reader whose ids span several waiters Ids registered by separate add_waiting_task_id() calls sit in separate waiters, so no one waiter's condition covers a receive that names ids from both: find_waiter() picks one, and a result for an id in the other notifies a condition the reader is not on. Measured, with two separate registrations and a receive naming both: 2700 ms and a null result when the send goes to the id whose waiter was not picked, 0 ms when it goes to the other. Both directions are driven, because which one the lookup picks depends on the set's iteration order. Such a reader now parks on the shared condition instead, and send() notifies that as well. It is guarded by a counter rather than done unconditionally, so the ordinary path pays one integer compare and not a second notify: server_response_reader registers every id it wants in a single call, so n_split_readers is zero for every reader in the tree. --- tools/server/server-queue.cpp | 54 ++++++++++++++++++--- tools/server/server-queue.h | 12 ++++- tools/server/tests/test-server-response.cpp | 31 ++++++++++++ 3 files changed, 89 insertions(+), 8 deletions(-) diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 62cec87743b5..cf8f815eecdd 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -476,6 +476,26 @@ server_task_result_ptr server_response::take_result(const std::unordered_set & id_tasks) const { + const waiter * first = nullptr; + + for (const auto & id_task : id_tasks) { + auto it = waiting.find(id_task); + if (it == waiting.end()) { + continue; + } + if (first == nullptr) { + first = it->second.get(); + continue; + } + if (it->second.get() != first) { + return true; + } + } + + return false; +} + server_task_result_ptr server_response::recv(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); @@ -497,10 +517,14 @@ server_task_result_ptr server_response::recv(const std::unordered_set & id_ // for these ids, which blocks this one connection and nothing else, so that is what it // does here too. The lookup is inside the loop rather than above it because a waiter // re-added while we wait should be picked up instead of waited out. + // ids registered by separate calls sit in separate waiters, and no one waiter's condition + // covers them, so those readers park on the shared one and send() notifies it for them auto w = find_waiter(id_tasks); - if (w == nullptr) { + if (w == nullptr || spans_waiters(id_tasks)) { // registration and terminate() both fire condition_gone; the timeout is only a backstop + if (w != nullptr) { n_split_readers++; } condition_gone.wait_for(lock, std::chrono::seconds(1)); + if (w != nullptr) { n_split_readers--; } continue; } @@ -531,13 +555,20 @@ server_task_result_ptr server_response::recv_with_timeout(const std::unordered_s auto w = find_waiter(id_tasks); - // The ids are not registered yet, or not any more. Wait on condition_gone rather than - // sleeping out the timeout: add_waiting_task_id(s) fires it, so a result that arrives - // during this call is still seen, which is what the single shared condition used to give. - // terminate() fires it too, so it is honoured here as well as on the waiter's own cv. - std::condition_variable & cv = w == nullptr ? condition_gone : w->cv; + // Park on the shared condition when the ids are not registered yet, or not any more, or + // when they span several waiters so that no one waiter's condition covers them. + // add_waiting_task_id(s) fires it, so a result that arrives during this call is still + // seen, which is what the single shared condition used to give; terminate() fires it too; + // and send() fires it while a split reader is parked. + const bool split = w != nullptr && spans_waiters(id_tasks); + + std::condition_variable & cv = (w == nullptr || split) ? condition_gone : w->cv; - if (cv.wait_until(lock, deadline) == std::cv_status::timeout) { + if (split) { n_split_readers++; } + const std::cv_status st = cv.wait_until(lock, deadline); + if (split) { n_split_readers--; } + + if (st == std::cv_status::timeout) { if (!running) { RES_DBG("%s : queue result stop\n", __func__); std::terminate(); // we cannot return here since the caller is HTTP code @@ -576,6 +607,11 @@ void server_response::send(server_task_result_ptr && result) { // not the single global one the shared vector used, so it is still O(1) in the common case // of one thread per reader. w.cv.notify_all(); + + // normally zero: only a reader whose ids span several waiters parks on the shared condition + if (n_split_readers > 0) { + condition_gone.notify_all(); + } } void server_response::broadcast(server_task_result_ptr && result) { @@ -587,6 +623,10 @@ void server_response::broadcast(server_task_result_ptr && result) { w->results.emplace_back(std::move(res_copy)); w->cv.notify_all(); } + + if (n_split_readers > 0) { + condition_gone.notify_all(); + } } void server_response::terminate() { diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index dd00519070e3..737eae414844 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -172,15 +172,25 @@ struct server_response { std::mutex mutex_results; - // parks a reader whose ids left the waiting list, so it honours its timeout + // parks a reader whose ids left the waiting list, so it honours its timeout, and a reader + // whose ids span several waiters, for which no single waiter's condition is enough std::condition_variable condition_gone; + // how many readers are parked on condition_gone for ids that ARE registered, i.e. spread over + // more than one waiter. Normally zero, so send() pays one integer compare rather than a + // second notify: server_response_reader registers all of its ids in one call. + size_t n_split_readers = 0; + // ids registered together share one waiter, so the first hit is the right one. mutex_results held. waiter_ptr find_waiter(const std::unordered_set & id_tasks) const; // pop the oldest queued result whose id the caller asked for. mutex_results held. server_task_result_ptr take_result(const std::unordered_set & id_tasks); + // true if the registered ids among id_tasks belong to more than one waiter, which happens + // only when they were registered by separate calls. mutex_results held. + bool spans_waiters(const std::unordered_set & id_tasks) const; + public: // add the id_task to the list of tasks waiting for response void add_waiting_task_id(int id_task); diff --git a/tools/server/tests/test-server-response.cpp b/tools/server/tests/test-server-response.cpp index 22a5d0ed199e..919b60cad51b 100644 --- a/tools/server/tests/test-server-response.cpp +++ b/tools/server/tests/test-server-response.cpp @@ -365,6 +365,36 @@ static void t_subset_receivers_are_all_woken() { "a subset receiver is woken even when siblings wait too", d); } + +// Ids registered by separate calls belong to separate waiters, so no single condition covers a +// receive that names both. The receiver must be woken by a result for either of them, whichever +// waiter the lookup happened to pick, so both directions are driven. +static void t_ids_spanning_two_waiters_one(int base_id, int send_to, const char * label) { + server_response res; + res.add_waiting_task_id(base_id); // two separate registrations, so two waiters + res.add_waiting_task_id(base_id + 1); + + std::atomic got{-1}; + std::thread reader([&] { + auto r = res.recv_with_timeout({base_id, base_id + 1}, 3); + got.store(payload_of(r)); + }); + std::this_thread::sleep_for(ms(300)); + + const auto t0 = std::chrono::steady_clock::now(); + res.send(mk(send_to, 5000 + send_to)); + reader.join(); + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + + char d[96]; snprintf(d, sizeof(d), "%lldms, payload=%d", (long long) waited, got.load()); + check(got.load() == 5000 + send_to && waited < 2500, label, d); +} + +static void t_ids_spanning_two_waiters() { + t_ids_spanning_two_waiters_one(500, 500, "a receive over two waiters is woken by the first id"); + t_ids_spanning_two_waiters_one(600, 601, "a receive over two waiters is woken by the second id"); +} + static long rss_kb() { // Linux only; returns -1 elsewhere, and only the optional "leak" mode uses it FILE * f = fopen("/proc/self/status", "r"); @@ -418,6 +448,7 @@ int main(int argc, char ** argv) { t_terminate_while_parked_on_absent_ids(); t_subset_recv_is_filtered(); t_subset_receivers_are_all_woken(); + t_ids_spanning_two_waiters(); t_parked_reader_does_not_abort(); printf("\nRESULT queue failures=%d\n", g_fail); From 031b480e599880b74547cfcb443b0eaf294ea0b8 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 9 Sep 2026 04:26:31 -0700 Subject: [PATCH 09/10] server: keep arrival order across waiters, and wake a reader whose waiter is discarded Two more on the multi-waiter path. Results carry an arrival sequence now. Each waiter's queue is already in arrival order, so its first match is its oldest, and only those per-waiter winners are compared; a receive naming ids from one waiter, which is every reader in the tree, takes the same single scan as before and never builds the comparison list. Without this, iterating the requested id set picked an arbitrary waiter's deque, so sending A then B could return B first, where the shared vector scanned from the front and did not. Measured in both directions, because which waiter the lookup reaches first depends on the set's iteration order. remove_waiting_task_id() and remove_waiting_task_ids() now notify the waiter they discard. A reader that had already selected that waiter was parked on a condition nothing would fire again: re-registering the id built a new waiter, registration notified only condition_gone and the send notified only the new waiter, so the reader waited out its deadline and returned nullptr with a result sitting in the queue. Measured at 2700 ms and a null result, 0 ms after. --- tools/server/server-queue.cpp | 107 +++++++++++++++----- tools/server/server-queue.h | 13 ++- tools/server/tests/test-server-response.cpp | 54 ++++++++++ 3 files changed, 147 insertions(+), 27 deletions(-) diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index cf8f815eecdd..018f0465074a 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -425,21 +425,41 @@ void server_response::remove_waiting_task_id(int id_task) { // the waiter is shared with the reader's other ids, so drop only this task's results auto & results = it->second->results; results.erase( - std::remove_if(results.begin(), results.end(), [id_task](const server_task_result_ptr & res) { - return res->id == id_task; + std::remove_if(results.begin(), results.end(), [id_task](const pending & p) { + return p.res->id == id_task; }), results.end()); + // a reader may be parked on this waiter; it has to repeat the lookup rather than wait out its + // deadline on a condition that nothing will fire again + auto w = it->second; waiting.erase(it); + w->cv.notify_all(); + condition_gone.notify_all(); } void server_response::remove_waiting_task_ids(const std::unordered_set & id_tasks) { std::unique_lock lock(mutex_results); + std::vector removed; + for (const auto & id_task : id_tasks) { RES_DBG("remove task %d from waiting list. current waiting = %d (before remove)\n", id_task, (int) waiting.size()); - waiting.erase(id_task); + + auto it = waiting.find(id_task); + if (it == waiting.end()) { + continue; + } + + removed.push_back(it->second); + waiting.erase(it); + } + + // same as the single id form: wake anyone parked on a waiter that no longer serves these ids + for (const auto & w : removed) { + w->cv.notify_all(); } + condition_gone.notify_all(); } server_response::waiter_ptr server_response::find_waiter(const std::unordered_set & id_tasks) const { @@ -453,47 +473,82 @@ server_response::waiter_ptr server_response::find_waiter(const std::unordered_se return nullptr; } -// A waiter is shared by every id its reader registered in one call, so its queue can hold a -// sibling's result. Return only an id the caller asked for, in arrival order, which is what -// scanning the shared vector did. The front normally matches, so this is O(1) in practice. -server_task_result_ptr server_response::take_result(const std::unordered_set & id_tasks) { +// true when the ids the caller named were registered by separate calls, so they sit in more than +// one waiter and no single waiter's condition covers them. Short-circuits on the first mismatch. +bool server_response::spans_waiters(const std::unordered_set & id_tasks) const { + const waiter * first = nullptr; + for (const auto & id_task : id_tasks) { auto it = waiting.find(id_task); if (it == waiting.end()) { continue; } - - auto & results = it->second->results; - for (auto rit = results.begin(); rit != results.end(); ++rit) { - if (id_tasks.find((*rit)->id) != id_tasks.end()) { - server_task_result_ptr res = std::move(*rit); - results.erase(rit); - return res; - } + if (first == nullptr) { + first = it->second.get(); + continue; + } + if (it->second.get() != first) { + return true; } } - return nullptr; + return false; } -bool server_response::spans_waiters(const std::unordered_set & id_tasks) const { - const waiter * first = nullptr; +// A waiter is shared by every id its reader registered in one call, so its queue can hold a +// sibling's result. Return only an id the caller asked for, and the oldest such result across +// every waiter the ids map to, which is what scanning the shared vector did. Each waiter's queue +// is already in arrival order, so its first match is its oldest and only the winners are compared. +server_task_result_ptr server_response::take_result(const std::unordered_set & id_tasks) { + auto first_match = [&](waiter * w) { + return std::find_if(w->results.begin(), w->results.end(), [&](const pending & p) { + return id_tasks.find(p.res->id) != id_tasks.end(); + }); + }; + + auto claim = [](waiter * w, std::deque::iterator it) { + server_task_result_ptr res = std::move(it->res); + w->results.erase(it); + return res; + }; + + // the ordinary case: every id the caller named shares one waiter, so no comparison is needed + if (!spans_waiters(id_tasks)) { + auto w = find_waiter(id_tasks); + if (w == nullptr) { + return nullptr; + } + + auto it = first_match(w.get()); + return it == w->results.end() ? nullptr : claim(w.get(), it); + } + + waiter * best_w = nullptr; + std::deque::iterator best_it; + uint64_t best_seq = 0; + std::vector examined; for (const auto & id_task : id_tasks) { auto it = waiting.find(id_task); if (it == waiting.end()) { continue; } - if (first == nullptr) { - first = it->second.get(); - continue; + + waiter * w = it->second.get(); + if (std::find(examined.begin(), examined.end(), w) != examined.end()) { + continue; // ids commonly share a waiter, so do not scan the same queue twice } - if (it->second.get() != first) { - return true; + examined.push_back(w); + + auto rit = first_match(w); + if (rit != w->results.end() && (best_w == nullptr || rit->seq < best_seq)) { + best_w = w; + best_it = rit; + best_seq = rit->seq; } } - return false; + return best_w == nullptr ? nullptr : claim(best_w, best_it); } server_task_result_ptr server_response::recv(const std::unordered_set & id_tasks) { @@ -599,7 +654,7 @@ void server_response::send(server_task_result_ptr && result) { auto & w = *it->second; - w.results.emplace_back(std::move(result)); + w.results.push_back(pending{next_seq++, std::move(result)}); // notify_all, not notify_one: results are filtered by id, so waking a single waiter can wake // one taking a disjoint subset of this reader's ids, which finds nothing and sleeps again @@ -620,7 +675,7 @@ void server_response::broadcast(server_task_result_ptr && result) { RES_DBG("task id = %d pushed to result queue\n", id_task); server_task_result_ptr res_copy(result->clone()); res_copy->id = id_task; // override id with target task id - w->results.emplace_back(std::move(res_copy)); + w->results.push_back(pending{next_seq++, std::move(res_copy)}); w->cv.notify_all(); } diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index 737eae414844..1c1f21626901 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -160,16 +161,26 @@ struct server_response { // One waiter per reader, shared by every id it registered in one call. A single shared vector // plus one cv instead costs N wakeups and N scans per token, N^2 per decode step. A send // notifies this reader's own cv, so the wakeup is O(1) even though it is a notify_all. + // arrival order is global, not per waiter: a reader can name ids from several waiters and + // must still be served oldest first, which is what scanning the shared vector gave + struct pending { + uint64_t seq; + server_task_result_ptr res; + }; + struct waiter { std::condition_variable cv; - std::deque results; + std::deque results; }; using waiter_ptr = std::shared_ptr; std::unordered_map waiting; + // stamped onto every queued result so arrival order survives being split across waiters + uint64_t next_seq = 0; + std::mutex mutex_results; // parks a reader whose ids left the waiting list, so it honours its timeout, and a reader diff --git a/tools/server/tests/test-server-response.cpp b/tools/server/tests/test-server-response.cpp index 919b60cad51b..ae5119005c7f 100644 --- a/tools/server/tests/test-server-response.cpp +++ b/tools/server/tests/test-server-response.cpp @@ -395,6 +395,58 @@ static void t_ids_spanning_two_waiters() { t_ids_spanning_two_waiters_one(600, 601, "a receive over two waiters is woken by the second id"); } + +// Results must come back in arrival order even when the ids live in different waiters. The +// shared vector scanned from the front, so it did. Both orders are driven, because which waiter +// the lookup reaches first depends on the set's iteration order. +static void t_fifo_across_waiters_one(int first, int second, const char * label) { + server_response res; + res.add_waiting_task_id(first); // separate registrations, so separate waiters + res.add_waiting_task_id(second); + + res.send(mk(first, 7000 + first)); + res.send(mk(second, 7000 + second)); + + auto r1 = res.recv_with_timeout({first, second}, 1); + auto r2 = res.recv_with_timeout({first, second}, 1); + + char d[96]; + snprintf(d, sizeof(d), "got %d then %d, wanted %d then %d", + r1 ? r1->id : -1, r2 ? r2->id : -1, first, second); + check(r1 != nullptr && r2 != nullptr && r1->id == first && r2->id == second, label, d); +} + +static void t_fifo_across_waiters() { + t_fifo_across_waiters_one(700, 701, "arrival order kept across waiters, low id first"); + t_fifo_across_waiters_one(711, 710, "arrival order kept across waiters, high id first"); +} + +// A reader already parked on a waiter has to be woken when that waiter is discarded, or it will +// wait out its deadline on a condition nothing will ever fire again while its id is re-registered +// and served on a brand new waiter. +static void t_waiter_replaced_under_a_parked_reader() { + server_response res; + res.add_waiting_task_id(800); + + std::atomic got{-1}; + std::thread reader([&] { + auto r = res.recv_with_timeout({800}, 3); + got.store(payload_of(r)); + }); + std::this_thread::sleep_for(ms(300)); // let the reader select the current waiter and park + + const auto t0 = std::chrono::steady_clock::now(); + res.remove_waiting_task_id(800); // discards the waiter the reader is parked on + res.add_waiting_task_id(800); // a brand new waiter + res.send(mk(800, 8800)); + reader.join(); + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + + char d[80]; snprintf(d, sizeof(d), "%lldms, payload=%d", (long long) waited, got.load()); + check(got.load() == 8800 && waited < 2500, + "a parked reader is woken when its waiter is replaced", d); +} + static long rss_kb() { // Linux only; returns -1 elsewhere, and only the optional "leak" mode uses it FILE * f = fopen("/proc/self/status", "r"); @@ -449,6 +501,8 @@ int main(int argc, char ** argv) { t_subset_recv_is_filtered(); t_subset_receivers_are_all_woken(); t_ids_spanning_two_waiters(); + t_fifo_across_waiters(); + t_waiter_replaced_under_a_parked_reader(); t_parked_reader_does_not_abort(); printf("\nRESULT queue failures=%d\n", g_fail); From 5488483eb279e725d844f9e5c86ecf97d5609955 Mon Sep 17 00:00:00 2001 From: danielhanchen Date: Wed, 9 Sep 2026 04:39:22 -0700 Subject: [PATCH 10/10] server: keep a partly registered receive on the shared condition spans_waiters() ignored ids that are not registered, so a receive naming one registered id and one that did not exist yet parked on the registered id's condition. Registering the missing id put it on a different waiter, and the send for it notified only that waiter, so the result was missed until the deadline. Measured at 2700 ms and a null result, 0 ms after. Replaced by sole_waiter(), which returns a waiter only when it covers EVERY requested id, so an absent id keeps the reader on the shared condition until it appears. The counter that decides whether send() notifies the shared condition is now keyed on whether a result could already be delivered to that reader, rather than on the ids being spread over waiters, so the partly registered case is covered too. Still zero for every reader in the tree, which registers all of its ids in one call before it receives. --- tools/server/server-queue.cpp | 64 ++++++++++----------- tools/server/server-queue.h | 15 +++-- tools/server/tests/test-server-response.cpp | 26 +++++++++ 3 files changed, 67 insertions(+), 38 deletions(-) diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 018f0465074a..e72d53fe8ab8 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -473,26 +473,27 @@ server_response::waiter_ptr server_response::find_waiter(const std::unordered_se return nullptr; } -// true when the ids the caller named were registered by separate calls, so they sit in more than -// one waiter and no single waiter's condition covers them. Short-circuits on the first mismatch. -bool server_response::spans_waiters(const std::unordered_set & id_tasks) const { - const waiter * first = nullptr; +// The waiter that covers every requested id, or nullptr when they sit in more than one waiter or +// any of them is not registered yet. An absent id matters: it can be registered onto a different +// waiter while the reader waits, so no single waiter's condition covers the call. +server_response::waiter_ptr server_response::sole_waiter(const std::unordered_set & id_tasks) const { + waiter_ptr found = nullptr; for (const auto & id_task : id_tasks) { auto it = waiting.find(id_task); if (it == waiting.end()) { - continue; + return nullptr; } - if (first == nullptr) { - first = it->second.get(); + if (found == nullptr) { + found = it->second; continue; } - if (it->second.get() != first) { - return true; + if (it->second != found) { + return nullptr; } } - return false; + return found; } // A waiter is shared by every id its reader registered in one call, so its queue can hold a @@ -513,12 +514,7 @@ server_task_result_ptr server_response::take_result(const std::unordered_setresults.end() ? nullptr : claim(w.get(), it); } @@ -572,14 +568,17 @@ server_task_result_ptr server_response::recv(const std::unordered_set & id_ // for these ids, which blocks this one connection and nothing else, so that is what it // does here too. The lookup is inside the loop rather than above it because a waiter // re-added while we wait should be picked up instead of waited out. - // ids registered by separate calls sit in separate waiters, and no one waiter's condition - // covers them, so those readers park on the shared one and send() notifies it for them - auto w = find_waiter(id_tasks); - if (w == nullptr || spans_waiters(id_tasks)) { + // Only a waiter that covers every requested id has a condition that covers the whole + // receive. Anything else parks on the shared one, and send() notifies that for readers + // whose ids are at least partly registered, so a result cannot be missed. + auto w = sole_waiter(id_tasks); + if (w == nullptr) { + const bool deliverable = find_waiter(id_tasks) != nullptr; + // registration and terminate() both fire condition_gone; the timeout is only a backstop - if (w != nullptr) { n_split_readers++; } + if (deliverable) { n_split_readers++; } condition_gone.wait_for(lock, std::chrono::seconds(1)); - if (w != nullptr) { n_split_readers--; } + if (deliverable) { n_split_readers--; } continue; } @@ -608,20 +607,21 @@ server_task_result_ptr server_response::recv_with_timeout(const std::unordered_s return res; } - auto w = find_waiter(id_tasks); + // Park on the shared condition unless one waiter covers every requested id: the ids may + // be spread over several waiters, or some of them may not be registered yet, and either + // way no one waiter's condition covers the call. add_waiting_task_id(s) fires the shared + // one, so a result that arrives during this call is still seen, which is what the single + // shared condition used to give; terminate() fires it too; and send() fires it while a + // reader that could already be served is parked there. + auto w = sole_waiter(id_tasks); - // Park on the shared condition when the ids are not registered yet, or not any more, or - // when they span several waiters so that no one waiter's condition covers them. - // add_waiting_task_id(s) fires it, so a result that arrives during this call is still - // seen, which is what the single shared condition used to give; terminate() fires it too; - // and send() fires it while a split reader is parked. - const bool split = w != nullptr && spans_waiters(id_tasks); + const bool deliverable = w == nullptr && find_waiter(id_tasks) != nullptr; - std::condition_variable & cv = (w == nullptr || split) ? condition_gone : w->cv; + std::condition_variable & cv = w == nullptr ? condition_gone : w->cv; - if (split) { n_split_readers++; } + if (deliverable) { n_split_readers++; } const std::cv_status st = cv.wait_until(lock, deadline); - if (split) { n_split_readers--; } + if (deliverable) { n_split_readers--; } if (st == std::cv_status::timeout) { if (!running) { diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index 1c1f21626901..b688d7dcd677 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -187,9 +187,10 @@ struct server_response { // whose ids span several waiters, for which no single waiter's condition is enough std::condition_variable condition_gone; - // how many readers are parked on condition_gone for ids that ARE registered, i.e. spread over - // more than one waiter. Normally zero, so send() pays one integer compare rather than a - // second notify: server_response_reader registers all of its ids in one call. + // how many readers are parked on condition_gone even though a result could already be + // delivered to them, i.e. their ids are spread over several waiters or only partly + // registered. Normally zero, so send() pays one integer compare rather than a second notify: + // server_response_reader registers all of its ids in one call, before it ever receives. size_t n_split_readers = 0; // ids registered together share one waiter, so the first hit is the right one. mutex_results held. @@ -198,9 +199,11 @@ struct server_response { // pop the oldest queued result whose id the caller asked for. mutex_results held. server_task_result_ptr take_result(const std::unordered_set & id_tasks); - // true if the registered ids among id_tasks belong to more than one waiter, which happens - // only when they were registered by separate calls. mutex_results held. - bool spans_waiters(const std::unordered_set & id_tasks) const; + // The waiter that covers EVERY requested id, or nullptr when they are spread over several + // waiters or any of them is not registered. Only then does one waiter's condition cover the + // whole receive; an id that is absent now can be registered onto a different waiter while + // the reader waits. mutex_results held. + waiter_ptr sole_waiter(const std::unordered_set & id_tasks) const; public: // add the id_task to the list of tasks waiting for response diff --git a/tools/server/tests/test-server-response.cpp b/tools/server/tests/test-server-response.cpp index ae5119005c7f..594622d7588b 100644 --- a/tools/server/tests/test-server-response.cpp +++ b/tools/server/tests/test-server-response.cpp @@ -447,6 +447,31 @@ static void t_waiter_replaced_under_a_parked_reader() { "a parked reader is woken when its waiter is replaced", d); } + +// A receive can name an id that is not registered yet. One waiter's condition does not cover +// such a call, because the missing id may be registered on a different waiter while it waits. +static void t_partially_registered_receive() { + server_response res; + res.add_waiting_task_id(900); // 901 does not exist yet + + std::atomic got{-1}; + std::thread reader([&] { + auto r = res.recv_with_timeout({900, 901}, 3); + got.store(payload_of(r)); + }); + std::this_thread::sleep_for(ms(300)); // let the reader park on whatever it picked + + const auto t0 = std::chrono::steady_clock::now(); + res.add_waiting_task_id(901); // a separate waiter + res.send(mk(901, 9010)); + reader.join(); + const auto waited = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + + char d[80]; snprintf(d, sizeof(d), "%lldms, payload=%d", (long long) waited, got.load()); + check(got.load() == 9010 && waited < 2500, + "a receive naming an id registered later is still woken", d); +} + static long rss_kb() { // Linux only; returns -1 elsewhere, and only the optional "leak" mode uses it FILE * f = fopen("/proc/self/status", "r"); @@ -503,6 +528,7 @@ int main(int argc, char ** argv) { t_ids_spanning_two_waiters(); t_fifo_across_waiters(); t_waiter_replaced_under_a_parked_reader(); + t_partially_registered_receive(); t_parked_reader_does_not_abort(); printf("\nRESULT queue failures=%d\n", g_fail);