From a7b9b3991bcee6f4a24c2cf46e52b6dc5ef83176 Mon Sep 17 00:00:00 2001 From: Anton Date: Mon, 21 Sep 2026 15:45:51 +0200 Subject: [PATCH] perf(pool): wake only the workers a dispatch can keep busy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ray_pool_dispatch and ray_pool_dispatch_n signalled the whole pool on every dispatch, however narrow the window. The main thread participates as worker 0, so at most n_tasks-1 helpers can ever claim anything; the surplus threads woke, raced to an already-drained window, and went straight back to the semaphore. On a dispatch narrower than the machine that is pure overhead — a partition-parallel step over three partitions woke every core on the box to run two tasks. Signal min(n_tasks-1, n_workers) instead. Signalling FEWER is safe: completion is governed by the `pending` spin-wait, never by the signal count, so an unsignalled worker simply stays asleep, and under-signalling cannot produce the surplus-signal problem between consecutive dispatches that the spin-wait exists to avoid. Measured (10M-row ClickBench, 43 queries x 3, splayed): sum of per-query hot times 5690.7 ms -> 5667.3 ms (-0.4%, within noise), user CPU 55.1 s -> 54.8 s. So this is waste removal rather than a speedup on workloads whose dispatches are already wider than the pool. Not a fix for #599. That report's shape — a 130k-row join probe, which splits into 16 tasks against 8 workers — is unaffected, because the clamp lands on the same worker count it already used (measured: 5.19 s -> 5.18 s of CPU, unchanged). The investigation there points at the pool defaulting to logical rather than physical cores; that is a separate change with its own benchmark trade and is filed on its own. test_pool.c gains pool/dispatch_narrow, which is the case the existing coverage missed: dispatch_n_small uses 4 tasks on a 2-worker pool, so the clamp never binds. The new test drives windows of 1..4 tasks on a 4-worker pool, then alternates narrow and wide windows 25 times over the same pool, asserting exact task and element counts throughout — the two risks the change introduces are a task nobody claims and signal accounting that drifts across dispatches and starves a later wide window. --- src/core/pool.c | 24 ++++++++++++++++++---- test/test_pool.c | 52 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 4 deletions(-) diff --git a/src/core/pool.c b/src/core/pool.c index 0b81070a..c3cac800 100644 --- a/src/core/pool.c +++ b/src/core/pool.c @@ -340,8 +340,16 @@ void ray_pool_dispatch(ray_pool_t* pool, ray_pool_fn fn, void* ctx, const bool prog = ray_qstats_mode() & RAY_QS_PROGRESS; if (prog) ray_progress_dispatch_begin((uint64_t)total_elems); - /* Wake worker threads */ - for (uint32_t i = 0; i < pool->n_workers; i++) { + /* Wake worker threads — only as many as have something to claim. Main + * participates as worker 0, so at most n_tasks-1 helpers can be useful; + * signalling the whole pool woke threads that raced to an already-drained + * window and went straight back to sleep, which is pure overhead on any + * dispatch narrower than the machine (#599). Signalling FEWER is safe: + * completion is governed by the `pending` spin-wait below, never by the + * signal count, so an unsignalled worker simply stays asleep. */ + uint32_t wake = n_tasks > 0 ? n_tasks - 1 : 0; + if (wake > pool->n_workers) wake = pool->n_workers; + for (uint32_t i = 0; i < wake; i++) { ray_sem_signal(&pool->work_ready); } @@ -413,8 +421,16 @@ static void dispatch_n_round(ray_pool_t* pool, ray_pool_fn fn, void* ctx, const bool prog = ray_qstats_mode() & RAY_QS_PROGRESS; if (prog) ray_progress_dispatch_begin((uint64_t)n_tasks); - /* Wake worker threads */ - for (uint32_t i = 0; i < pool->n_workers; i++) { + /* Wake worker threads — only as many as have something to claim. Main + * participates as worker 0, so at most n_tasks-1 helpers can be useful; + * signalling the whole pool woke threads that raced to an already-drained + * window and went straight back to sleep, which is pure overhead on any + * dispatch narrower than the machine (#599). Signalling FEWER is safe: + * completion is governed by the `pending` spin-wait below, never by the + * signal count, so an unsignalled worker simply stays asleep. */ + uint32_t wake = n_tasks > 0 ? n_tasks - 1 : 0; + if (wake > pool->n_workers) wake = pool->n_workers; + for (uint32_t i = 0; i < wake; i++) { ray_sem_signal(&pool->work_ready); } diff --git a/test/test_pool.c b/test/test_pool.c index 7e2f177d..8eb499a1 100644 --- a/test/test_pool.c +++ b/test/test_pool.c @@ -462,6 +462,57 @@ static test_result_t test_dispatch_small(void) { * spin-wait for completion). * -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- + * Test: a dispatch NARROWER than the pool still runs every task, and leaves + * the pool fully usable for a later wide one (#599). + * + * ray_pool_dispatch/_n now signal only min(n_tasks-1, n_workers) workers + * instead of the whole pool, so the under-signalled workers stay asleep. The + * risks that buys are (a) a task nobody claims and (b) signal accounting that + * drifts across dispatches, starving a later wide window. Alternating narrow + * and wide dispatches over one pool catches both: every window is verified for + * exact task and element counts. + * -------------------------------------------------------------------------- */ +static test_result_t test_dispatch_narrower_than_pool(void) { + ray_heap_init(); + + ray_pool_t pool; + TEST_ASSERT_EQ_I(ray_pool_create(&pool, 4), RAY_OK); + + /* n_tasks below the worker count: 1 wakes nobody, 2 wakes one, etc. */ + for (uint32_t n = 1; n <= 4; n++) { + pool_count_ctx_t ctx = {0}; + ray_pool_dispatch_n(&pool, pool_count_fn, &ctx, n); + TEST_ASSERT_EQ_I(atomic_load(&ctx.calls), n); + TEST_ASSERT_EQ_I(atomic_load(&ctx.elem_sum), n); + } + + /* The element form, sized to a single task. */ + { + pool_count_ctx_t ctx = {0}; + ray_pool_dispatch(&pool, pool_count_fn, &ctx, 1); + TEST_ASSERT_EQ_I(atomic_load(&ctx.calls), 1); + TEST_ASSERT_EQ_I(atomic_load(&ctx.elem_sum), 1); + } + + /* Alternating narrow/wide over the same pool: a wide window must still be + * fully served after windows that signalled fewer workers than exist. */ + for (int rep = 0; rep < 25; rep++) { + pool_count_ctx_t narrow = {0}; + ray_pool_dispatch_n(&pool, pool_count_fn, &narrow, 1); + TEST_ASSERT_EQ_I(atomic_load(&narrow.calls), 1); + + pool_count_ctx_t wide = {0}; + ray_pool_dispatch_n(&pool, pool_count_fn, &wide, 32); + TEST_ASSERT_EQ_I(atomic_load(&wide.calls), 32); + TEST_ASSERT_EQ_I(atomic_load(&wide.elem_sum), 32); + } + + ray_pool_free(&pool); + ray_heap_destroy(); + PASS(); +} + static test_result_t test_dispatch_n_small(void) { ray_heap_init(); @@ -1357,6 +1408,7 @@ const test_entry_t pool_entries[] = { { "pool/dispatch_zero_elems", test_dispatch_zero_elems, NULL, NULL }, { "pool/dispatch_small", test_dispatch_small, NULL, NULL }, { "pool/dispatch_n_small", test_dispatch_n_small, NULL, NULL }, + { "pool/dispatch_narrow", test_dispatch_narrower_than_pool, NULL, NULL }, { "pool/dispatch_n_ring_grow", test_dispatch_n_ring_growth, NULL, NULL }, { "pool/dispatch_ring_grow", test_dispatch_ring_growth, NULL, NULL }, { "pool/dispatch_n_cancelled", test_dispatch_n_cancelled, NULL, NULL },