From 8c87f226c12de4a21450137fd06a0132d71b4ede Mon Sep 17 00:00:00 2001 From: Martin Ek Date: Fri, 9 Oct 2026 13:23:43 -0700 Subject: [PATCH 1/4] Add failing test for lost wakeup when WorkerPool is re-entered from `block`. If the scheduler's `block` hook performs a nested `WorkerPool#call` on the same fiber before suspending (as a GC finalizer can, e.g. `Tempfile` unlinking its file), the outer operation's wakeup resumes the nested wait instead. The nested operation is cancelled and the outer wait never wakes again. Also record whether the `busy` test operation actually executed, since an operation cancelled before it starts is otherwise reported as completed. Co-Authored-By: Claude Opus 5.5 (1M context) --- ext/io/event/worker_pool_test.c | 6 +++ test/io/event/worker_pool.rb | 65 +++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) diff --git a/ext/io/event/worker_pool_test.c b/ext/io/event/worker_pool_test.c index 3def74df..c6b45bde 100644 --- a/ext/io/event/worker_pool_test.c +++ b/ext/io/event/worker_pool_test.c @@ -18,6 +18,7 @@ struct BusyOperationData { int read_fd; int write_fd; volatile int cancelled; + volatile int executed; double duration; // How long to wait (for testing) clock_t start_time; clock_t end_time; @@ -29,6 +30,9 @@ struct BusyOperationData { static void* busy_blocking_operation(void *data) { struct BusyOperationData *busy_data = (struct BusyOperationData*)data; + // Record that the operation actually ran (a cancelled operation may be skipped entirely): + busy_data->executed = 1; + // Use select() to wait for the pipe to become readable fd_set read_fds; struct timeval timeout; @@ -143,6 +147,7 @@ static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) { .read_fd = pipe_fds[0], .write_fd = pipe_fds[1], .cancelled = 0, + .executed = 0, .duration = duration, .start_time = 0, .end_time = 0, @@ -165,6 +170,7 @@ static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) { VALUE result = rb_hash_new(); rb_hash_aset(result, ID2SYM(rb_intern("duration")), DBL2NUM(duration)); rb_hash_aset(result, ID2SYM(rb_intern("elapsed")), DBL2NUM(elapsed)); + rb_hash_aset(result, ID2SYM(rb_intern("executed")), busy_data.executed ? Qtrue : Qfalse); // Determine result based on operation outcome if (busy_data.exception != Qnil) { diff --git a/test/io/event/worker_pool.rb b/test/io/event/worker_pool.rb index 8ac3543d..283c419a 100644 --- a/test/io/event/worker_pool.rb +++ b/test/io/event/worker_pool.rb @@ -324,4 +324,69 @@ def block(...) expect(rescued).to be == true end end + + with "scheduler that re-enters the worker pool from block" do + # Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing one nested blocking operation from inside the first worker pool `block`. + class ReentrantBlockScheduler < IO::Event::TestScheduler + attr_reader :nested_result + + def block(blocker, timeout = nil) + # Simulate a finalizer that performs an offloaded blocking operation, exactly once: + if blocker.is_a?(IO::Event::WorkerPool) and !@reentered + @reentered = true + @nested_result = IO::Event::WorkerPool.busy(duration: 0.01) + end + + super + end + end + + class WaitTimeout < StandardError + end + + it "wakes the outer operation and runs the nested operation" do + result = nil + nested_result = nil + + Thread.new do + # A single worker guarantees that the nested operation queues behind the outer one: + scheduler = ReentrantBlockScheduler.new(maximum_worker_count: 1) + Fiber.set_scheduler(scheduler) + + fiber = Fiber.schedule do + result = IO::Event::WorkerPool.busy(duration: 0.1) + end + + # Bound the outer wait so a lost wakeup fails the test instead of hanging it: + Fiber.schedule do + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 2.0 + + while result.nil? and Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline + sleep(0.01) + end + + Fiber.scheduler.fiber_interrupt(fiber, WaitTimeout) if result.nil? + end + + scheduler.run + nested_result = scheduler.nested_result + ensure + Fiber.set_scheduler(nil) + scheduler&.close + end.join + + # The nested operation should run to completion, rather than being cancelled when the outer operation's wakeup resumes the nested wait (depending on timing, the cancelled operation is either interrupted or skipped entirely): + expect(nested_result).to have_keys( + executed: be == true, + result: be == :completed + ) + + # The outer operation should complete without being interrupted by the timeout: + expect(result[:exception]).to be_nil + expect(result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + end end From f03f371b0d388d6ae0bd73de23b00ed44c79ea25 Mon Sep 17 00:00:00 2001 From: Martin Ek Date: Fri, 9 Oct 2026 13:51:12 -0700 Subject: [PATCH 2/4] Fix lost wakeup when the waiting fiber re-enters `WorkerPool#call`. Ruby can run GC finalizers inside the scheduler's `block` hook before the fiber suspends, and a finalizer that offloads a blocking operation (e.g. a `Tempfile` finalizer closing and unlinking its file) re-enters `WorkerPool#call` on the same fiber. Wakeups are addressed to the fiber, so the outer work's wakeup could resume the nested wait, which cancelled the nested operation and consumed the only wakeup for the outer one, which then never woke. - Only an exception or other non-local exit cancels the work. A plain wakeup no longer cancels it; the wait loops until the work's own wakeup has been accounted for. - Track the work each fiber is waiting for (innermost first, per pool). A wakeup is attributed to the innermost completed work which hasn't been woken yet, or ignored as spurious. When a nested call returns, any wakeup it consumed on behalf of outer work is re-issued with `rb_fiber_scheduler_unblock`. Add tests for the outer work completing before or during the nested call, the nested work finishing first, two levels of nesting with one or more workers, an exception raised while nested, and a spurious wakeup. Co-Authored-By: Claude Opus 5.5 (1M context) --- ext/io/event/worker_pool.c | 86 ++++++++++++--- releases.md | 1 + test/io/event/worker_pool.rb | 196 ++++++++++++++++++++++++++++++++--- 3 files changed, 254 insertions(+), 29 deletions(-) diff --git a/ext/io/event/worker_pool.c b/ext/io/event/worker_pool.c index c58ec825..1333e5a7 100644 --- a/ext/io/event/worker_pool.c +++ b/ext/io/event/worker_pool.c @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -37,14 +38,23 @@ struct IO_Event_WorkerPool_Worker { // Work item structure struct IO_Event_WorkerPool_Work { + struct IO_Event_WorkerPool *pool; + rb_fiber_scheduler_blocking_operation_t *blocking_operation; + // Set by the worker (holding the GVL) before it unblocks the waiting fiber, exactly once: bool completed; + // Set when a wakeup of the waiting fiber has been attributed to this work's completion. See `worker_pool_work_wakeup`. + bool woken; + VALUE scheduler; VALUE blocker; VALUE fiber; + // The work the same fiber was already waiting for when this work was submitted, if any. This happens when the scheduler's `block` hook performs a blocking operation before the fiber suspends, e.g. a GC finalizer that unlinks a file. + struct IO_Event_WorkerPool_Work *outer; + struct IO_Event_WorkerPool_Work *next; }; @@ -64,6 +74,9 @@ struct IO_Event_WorkerPool { size_t completed_count; size_t cancelled_count; + // Maps each fiber waiting in `call` to the innermost work it is waiting for (protected by the GVL): + st_table *waiting; + bool shutdown; }; @@ -83,6 +96,11 @@ static void worker_pool_free(void *ptr) { // Note: We don't free worker structures or wait for threads during GC // as this can cause deadlocks. The Ruby GC will handle the thread objects. // Workers will see the shutdown flag and exit cleanly. + + if (pool->waiting) { + st_free_table(pool->waiting); + pool->waiting = NULL; + } } } @@ -110,7 +128,9 @@ static void worker_pool_compact(void *ptr) // Size functions for Ruby GC static size_t worker_pool_size(const void *ptr) { - return sizeof(struct IO_Event_WorkerPool); + const struct IO_Event_WorkerPool *pool = (const struct IO_Event_WorkerPool *)ptr; + + return sizeof(struct IO_Event_WorkerPool) + (pool->waiting ? st_memsize(pool->waiting) : 0); } // Ruby TypedData structures @@ -315,21 +335,28 @@ static VALUE worker_pool_work_block(VALUE _work) { return Qnil; } +// Account for the waiting fiber being resumed without an exception. +// +// Each completed work unblocks its fiber exactly once, but wakeups are addressed to the fiber, not to the work. If the fiber re-enters `call` while waiting for outer work, the outer work's wakeup can resume the nested wait instead, so attribute the wakeup to the innermost work (starting from `work`) which has completed but not yet been woken. If there is no such work, the wakeup is spurious (e.g. a stale `unblock`) and is ignored. +static void worker_pool_work_wakeup(struct IO_Event_WorkerPool_Work *work) { + for (; work; work = work->outer) { + if (work->completed && !work->woken) { + work->woken = true; + return; + } + } +} + static VALUE worker_pool_work_wait(VALUE _work) { struct IO_Event_WorkerPool_Work *work = (void*)_work; - - while (true) { + + // Only an exception or other non-local exit cancels the work (see `worker_pool_work_ensure`). Any other wakeup was either for this work, for outer work (re-issued when this call returns), or spurious, so keep waiting: + while (!work->woken) { worker_pool_work_block(_work); - if (DEBUG) fprintf(stderr, "-- worker_pool_work_wait:work completed=%d\n", work->completed); - - if (work->completed) { - break; - } - - if (DEBUG) fprintf(stderr, "worker_pool_work_wait:rb_fiber_scheduler_blocking_operation_cancel\n"); - rb_fiber_scheduler_blocking_operation_cancel(work->blocking_operation); + worker_pool_work_wakeup(work); + if (DEBUG) fprintf(stderr, "-- worker_pool_work_wait:work completed=%d woken=%d\n", work->completed, work->woken); } - + return Qtrue; } @@ -348,9 +375,28 @@ static VALUE worker_pool_work_ensure(VALUE _work) { // Ignore errors raised while waiting for cancellation to complete. rb_ensure // will restore and rethrow the original control-flow state after this returns. rb_set_errinfo(Qnil); + } else { + worker_pool_work_wakeup(work); } } - + + // This work is no longer being waited for: + st_data_t fiber = (st_data_t)work->fiber; + if (work->outer) { + st_insert(work->pool->waiting, fiber, (st_data_t)work->outer); + } else { + st_delete(work->pool->waiting, &fiber, NULL); + } + + // If we consumed the wakeup for outer work, the fiber won't be resumed again for it, so re-issue it: + for (struct IO_Event_WorkerPool_Work *outer = work->outer; outer; outer = outer->outer) { + if (outer->woken) { + if (DEBUG) fprintf(stderr, "worker_pool_work_ensure:rb_fiber_scheduler_unblock outer=%p\n", outer); + outer->woken = false; + rb_fiber_scheduler_unblock(outer->scheduler, outer->blocker, outer->fiber); + } + } + return Qnil; } @@ -382,14 +428,24 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) { // Create work item struct IO_Event_WorkerPool_Work work = { + .pool = pool, .blocking_operation = blocking_operation, .completed = false, + .woken = false, .scheduler = scheduler, .blocker = self, .fiber = fiber, + .outer = NULL, .next = NULL }; - + + // Track the work this fiber is waiting for, so that a nested call can account for its wakeup: + st_data_t outer = 0; + if (st_lookup(pool->waiting, (st_data_t)fiber, &outer)) { + work.outer = (struct IO_Event_WorkerPool_Work *)outer; + } + st_insert(pool->waiting, (st_data_t)fiber, (st_data_t)&work); + // Enqueue work: pthread_mutex_lock(&pool->mutex); enqueue_work(pool, &work); @@ -409,6 +465,8 @@ static VALUE worker_pool_allocate(VALUE klass) { // Initialize to NULL/zero so we can detect uninitialized pools memset(pool, 0, sizeof(struct IO_Event_WorkerPool)); + pool->waiting = st_init_numtable(); + return self; } diff --git a/releases.md b/releases.md index b3a99aec..411cf7ea 100644 --- a/releases.md +++ b/releases.md @@ -3,6 +3,7 @@ ## Unreleased - Add `IO::Event::Futex` on Linux with Ruby 4.1+, including atomic value operations and blocking and scheduler-aware single and vector waits over shared memory. Futexes retain a counted allocation lock until explicitly closed or finalized; pending waits prevent closing until completion or cancellation has finished. + - Fix a lost wakeup in `IO::Event::WorkerPool` when the waiting fiber re-enters `WorkerPool#call` from inside the scheduler's `block` hook, e.g. a GC finalizer that closes or unlinks a `Tempfile`. The outer wait no longer hangs, and the nested operation is no longer cancelled. More generally, a wakeup without an exception no longer cancels the operation; only an exception or other non-local exit does. ## v1.22.1 diff --git a/test/io/event/worker_pool.rb b/test/io/event/worker_pool.rb index 283c419a..14b475f0 100644 --- a/test/io/event/worker_pool.rb +++ b/test/io/event/worker_pool.rb @@ -326,15 +326,33 @@ def block(...) end with "scheduler that re-enters the worker pool from block" do - # Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing one nested blocking operation from inside the first worker pool `block`. + # Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing nested blocking operations from inside the worker pool `block`, one per duration in `nested`, each from inside the previous one. class ReentrantBlockScheduler < IO::Event::TestScheduler - attr_reader :nested_result + def initialize(nested: [0.01], before_nested: nil, **options) + super(**options) + + @nested = nested.dup + @before_nested = before_nested + @nested_results = [] + end + + # @attribute [Array(Hash)] The results of the nested operations, innermost first. + attr_reader :nested_results + + def nested_result + @nested_results.first + end def block(blocker, timeout = nil) - # Simulate a finalizer that performs an offloaded blocking operation, exactly once: - if blocker.is_a?(IO::Event::WorkerPool) and !@reentered - @reentered = true - @nested_result = IO::Event::WorkerPool.busy(duration: 0.01) + # Simulate a finalizer that performs an offloaded blocking operation: + if blocker.is_a?(IO::Event::WorkerPool) and duration = @nested.shift + @before_nested&.call(self) + + result = IO::Event::WorkerPool.busy(duration: duration) + @nested_results << result + + # `busy` rescues the exception that cancelled it, so propagate it to the outer wait: + raise result[:exception] if result[:exception] end super @@ -344,22 +362,27 @@ def block(blocker, timeout = nil) class WaitTimeout < StandardError end - it "wakes the outer operation and runs the nested operation" do + class Stop < StandardError + end + + # Run `busy(duration:)` in a fiber on a new `ReentrantBlockScheduler`, interrupting it with `WaitTimeout` if it hasn't finished within `timeout` seconds, so a lost wakeup fails the test instead of hanging it. + # + # @returns [Array(Hash, ReentrantBlockScheduler)] The result of the outer operation, and the scheduler. + def run_busy(duration:, timeout: 2.0, **options, &block) result = nil - nested_result = nil + scheduler = nil Thread.new do - # A single worker guarantees that the nested operation queues behind the outer one: - scheduler = ReentrantBlockScheduler.new(maximum_worker_count: 1) + scheduler = ReentrantBlockScheduler.new(**options) Fiber.set_scheduler(scheduler) fiber = Fiber.schedule do - result = IO::Event::WorkerPool.busy(duration: 0.1) + result = IO::Event::WorkerPool.busy(duration: duration) + result[:statistics] = scheduler.worker_pool.statistics end - # Bound the outer wait so a lost wakeup fails the test instead of hanging it: Fiber.schedule do - deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 2.0 + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout while result.nil? and Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline sleep(0.01) @@ -368,15 +391,23 @@ class WaitTimeout < StandardError Fiber.scheduler.fiber_interrupt(fiber, WaitTimeout) if result.nil? end + block&.call(fiber) + scheduler.run - nested_result = scheduler.nested_result ensure Fiber.set_scheduler(nil) scheduler&.close end.join + return result, scheduler + end + + it "wakes the outer operation and runs the nested operation" do + # A single worker guarantees that the nested operation queues behind the outer one: + result, scheduler = run_busy(duration: 0.1, maximum_worker_count: 1) + # The nested operation should run to completion, rather than being cancelled when the outer operation's wakeup resumes the nested wait (depending on timing, the cancelled operation is either interrupted or skipped entirely): - expect(nested_result).to have_keys( + expect(scheduler.nested_result).to have_keys( executed: be == true, result: be == :completed ) @@ -388,5 +419,140 @@ class WaitTimeout < StandardError result: be == :completed ) end + + it "wakes the outer operation when it completed before the nested operation started" do + completed_count = nil + + before_nested = proc do |scheduler| + # Give the worker time to complete the outer operation and unblock the fiber: + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 0.1 + Thread.pass while Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline + + completed_count = scheduler.worker_pool.statistics[:completed_count] + end + + result, scheduler = run_busy(duration: 0.0, maximum_worker_count: 1, before_nested: before_nested) + + expect(completed_count).to be == 1 + expect(scheduler.nested_result).to have_keys( + executed: be == true, + result: be == :completed + ) + expect(result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + + it "wakes the outer operation when the nested operation finishes first" do + # With two workers, the nested operation runs concurrently and finishes long before the outer one: + result, scheduler = run_busy(duration: 0.2, maximum_worker_count: 2, nested: [0.01]) + + expect(scheduler.nested_result).to have_keys( + executed: be == true, + result: be == :completed + ) + expect(result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + + it "wakes every operation with two levels of nesting" do + result, scheduler = run_busy(duration: 0.05, maximum_worker_count: 1, nested: [0.02, 0.01]) + + expect(scheduler.nested_results.size).to be == 2 + scheduler.nested_results.each do |nested_result| + expect(nested_result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + + expect(result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + + it "wakes every operation with two levels of nesting and multiple workers" do + result, scheduler = run_busy(duration: 0.05, maximum_worker_count: 3, nested: [0.1, 0.01]) + + expect(scheduler.nested_results.size).to be == 2 + scheduler.nested_results.each do |nested_result| + expect(nested_result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + + expect(result).to have_keys( + executed: be == true, + result: be == :completed + ) + end + + it "cancels and drains both operations when interrupted while nested" do + start_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) + + # With two workers, both operations are in flight when the exception is raised: + result, scheduler = run_busy(duration: 2.0, maximum_worker_count: 2, nested: [2.0]) do |fiber| + Fiber.schedule do + sleep(0.1) + Fiber.scheduler.fiber_interrupt(fiber, Stop) + end + end + + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start_time + + # The exception cancels the nested operation and then, propagating from `block`, the outer operation: + expect(scheduler.nested_result).to have_keys( + result: be == :exception, + exception: be_a(Stop) + ) + expect(result).to have_keys( + result: be == :exception, + exception: be_a(Stop) + ) + expect(elapsed).to be < 1.0 + + # Both operations were finished by the workers before returning: + expect(result[:statistics]).to have_keys( + call_count: be == 2, + completed_count: be == 2, + current_queue_size: be == 0 + ) + end + end + + with "spurious wakeup" do + let(:scheduler) {IO::Event::TestScheduler.new} + + it "does not cancel the operation" do + result = nil + + Thread.new do + Fiber.set_scheduler(scheduler) + + fiber = Fiber.schedule do + result = IO::Event::WorkerPool.busy(duration: 0.1) + end + + # Resume the waiting fiber without an exception, e.g. a stale `unblock`: + Fiber.schedule do + Fiber.scheduler.unblock(nil, fiber) + end + + scheduler.run + ensure + Fiber.set_scheduler(nil) + scheduler.close + end.join + + expect(result).to have_keys( + executed: be == true, + result: be == :completed + ) + end end end From db27094e96bd0c4523fd48f0086183d6352678f6 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sat, 17 Oct 2026 16:05:25 +1300 Subject: [PATCH 3/4] Preserve outer worker wakeups when nested waits are interrupted. A nested block can consume the outer operation's completion wakeup and then raise before wakeup accounting runs. If a finalizer swallows that exception, the outer operation remains blocked even after both operations have finished. Account for exceptional exits from the initial wait and from cancellation cleanup, while preserving the original exception. Since block does not report whether it consumed a wakeup, exceptional exits are accounted for conservatively and may replay an extra wakeup. Add regressions for interruptions during the normal wait and cancellation drain loop, using a queue to control worker progress. Both tests fail without the fix. Update the release note to cover interrupted nested waits. Validation on macOS with Ruby 4.0.6: - Full suite: 325 passed, 21 skipped. - Worker pool: 23 passed with both KQueue and Select. - RuboCop: no offenses in the changed Ruby test file. --- ext/io/event/worker_pool.c | 13 ++++-- releases.md | 2 +- test/io/event/worker_pool.rb | 86 ++++++++++++++++++++++++++++++++++-- 3 files changed, 92 insertions(+), 9 deletions(-) diff --git a/ext/io/event/worker_pool.c b/ext/io/event/worker_pool.c index 1333e5a7..5148fc69 100644 --- a/ext/io/event/worker_pool.c +++ b/ext/io/event/worker_pool.c @@ -335,9 +335,10 @@ static VALUE worker_pool_work_block(VALUE _work) { return Qnil; } -// Account for the waiting fiber being resumed without an exception. +// Account for an exit from the scheduler's block hook, including an exceptional exit. // // Each completed work unblocks its fiber exactly once, but wakeups are addressed to the fiber, not to the work. If the fiber re-enters `call` while waiting for outer work, the outer work's wakeup can resume the nested wait instead, so attribute the wakeup to the innermost work (starting from `work`) which has completed but not yet been woken. If there is no such work, the wakeup is spurious (e.g. a stale `unblock`) and is ignored. +// An exception may arrive after a completion wakeup has resumed the fiber. The scheduler doesn't tell us whether that wakeup was consumed, so account for it conservatively: replaying an unconsumed wakeup is preferable to leaving completed outer work blocked forever. static void worker_pool_work_wakeup(struct IO_Event_WorkerPool_Work *work) { for (; work; work = work->outer) { if (work->completed && !work->woken) { @@ -362,7 +363,12 @@ static VALUE worker_pool_work_wait(VALUE _work) { static VALUE worker_pool_work_ensure(VALUE _work) { struct IO_Event_WorkerPool_Work *work = (void*)_work; - + + // A normal return from `worker_pool_work_wait` already accounted for its wakeup. An exceptional exit may have consumed a wakeup before reaching that accounting: + if (!work->woken) { + worker_pool_work_wakeup(work); + } + while (!work->completed) { if (DEBUG) fprintf(stderr, "worker_pool_work_ensure:rb_fiber_scheduler_blocking_operation_cancel\n"); rb_fiber_scheduler_blocking_operation_cancel(work->blocking_operation); @@ -375,9 +381,8 @@ static VALUE worker_pool_work_ensure(VALUE _work) { // Ignore errors raised while waiting for cancellation to complete. rb_ensure // will restore and rethrow the original control-flow state after this returns. rb_set_errinfo(Qnil); - } else { - worker_pool_work_wakeup(work); } + worker_pool_work_wakeup(work); } // This work is no longer being waited for: diff --git a/releases.md b/releases.md index 411cf7ea..6bd0d0be 100644 --- a/releases.md +++ b/releases.md @@ -3,7 +3,7 @@ ## Unreleased - Add `IO::Event::Futex` on Linux with Ruby 4.1+, including atomic value operations and blocking and scheduler-aware single and vector waits over shared memory. Futexes retain a counted allocation lock until explicitly closed or finalized; pending waits prevent closing until completion or cancellation has finished. - - Fix a lost wakeup in `IO::Event::WorkerPool` when the waiting fiber re-enters `WorkerPool#call` from inside the scheduler's `block` hook, e.g. a GC finalizer that closes or unlinks a `Tempfile`. The outer wait no longer hangs, and the nested operation is no longer cancelled. More generally, a wakeup without an exception no longer cancels the operation; only an exception or other non-local exit does. + - Fix a lost wakeup in `IO::Event::WorkerPool` when the waiting fiber re-enters `WorkerPool#call` from inside the scheduler's `block` hook, e.g. a GC finalizer that closes or unlinks a `Tempfile`. The outer wait no longer hangs, including when the nested wait is interrupted after resuming, and the nested operation is no longer cancelled by an unrelated wakeup. More generally, a wakeup without an exception no longer cancels the operation; only an exception or other non-local exit does. ## v1.22.1 diff --git a/test/io/event/worker_pool.rb b/test/io/event/worker_pool.rb index 14b475f0..7d1fed05 100644 --- a/test/io/event/worker_pool.rb +++ b/test/io/event/worker_pool.rb @@ -328,11 +328,12 @@ def block(...) with "scheduler that re-enters the worker pool from block" do # Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing nested blocking operations from inside the worker pool `block`, one per duration in `nested`, each from inside the previous one. class ReentrantBlockScheduler < IO::Event::TestScheduler - def initialize(nested: [0.01], before_nested: nil, **options) + def initialize(nested: [0.01], before_nested: nil, propagate_nested_errors: true, **options) super(**options) @nested = nested.dup @before_nested = before_nested + @propagate_nested_errors = propagate_nested_errors @nested_results = [] end @@ -352,7 +353,7 @@ def block(blocker, timeout = nil) @nested_results << result # `busy` rescues the exception that cancelled it, so propagate it to the outer wait: - raise result[:exception] if result[:exception] + raise result[:exception] if @propagate_nested_errors and result[:exception] end super @@ -365,15 +366,64 @@ class WaitTimeout < StandardError class Stop < StandardError end + # Interrupt a nested wait after it consumes the outer completion's wakeup. + class InterruptedBlockScheduler < ReentrantBlockScheduler + def initialize(interrupt_before_block: false, **options) + # Finalizers swallow exceptions instead of propagating them to the outer wait: + super(propagate_nested_errors: false, **options) + + @interrupt_before_block = interrupt_before_block + @block_depth = 0 + @interrupted = false + @unblocked = false + @resume_worker = Thread::Queue.new + end + + def unblock(blocker, fiber) + first_completion = blocker.is_a?(IO::Event::WorkerPool) && !@unblocked + @unblocked = true if first_completion + + super + + # Keep the single worker from completing the nested operation until its wait has been interrupted and entered cancellation cleanup: + @resume_worker.pop if first_completion + end + + def block(blocker, timeout = nil) + return super unless blocker.is_a?(IO::Event::WorkerPool) + + @block_depth += 1 + begin + if @block_depth == 2 + if @interrupt_before_block + @interrupt_before_block = false + raise Stop, "before suspension" + end + + @resume_worker.push(true) if @interrupted + end + + super + + if @block_depth == 2 and !@interrupted + @interrupted = true + raise Stop, "after resumption" + end + ensure + @block_depth -= 1 + end + end + end + # Run `busy(duration:)` in a fiber on a new `ReentrantBlockScheduler`, interrupting it with `WaitTimeout` if it hasn't finished within `timeout` seconds, so a lost wakeup fails the test instead of hanging it. # # @returns [Array(Hash, ReentrantBlockScheduler)] The result of the outer operation, and the scheduler. - def run_busy(duration:, timeout: 2.0, **options, &block) + def run_busy(duration:, timeout: 2.0, scheduler_class: ReentrantBlockScheduler, **options, &block) result = nil scheduler = nil Thread.new do - scheduler = ReentrantBlockScheduler.new(**options) + scheduler = scheduler_class.new(**options) Fiber.set_scheduler(scheduler) fiber = Fiber.schedule do @@ -523,6 +573,34 @@ def run_busy(duration:, timeout: 2.0, **options, &block) current_queue_size: be == 0 ) end + + with "an interrupted nested wait" do + [false, true].each do |interrupt_before_block| + with "interrupt_before_block=#{interrupt_before_block}", interrupt_before_block: interrupt_before_block do + it "preserves the outer wakeup when the nested exception is swallowed" do + result, scheduler = run_busy( + duration: 0.01, + nested: [2.0], + maximum_worker_count: 1, + scheduler_class: InterruptedBlockScheduler, + interrupt_before_block: interrupt_before_block + ) + + # When cancellation starts before suspension, the second exception occurs in the drain loop and must not replace the original one: + expect(scheduler.nested_result).to have_keys( + result: be == :exception, + exception: be_a(Stop).and(have_attributes(message: be == (interrupt_before_block ? "before suspension" : "after resumption"))) + ) + + expect(result).to have_keys( + executed: be == true, + result: be == :completed, + statistics: have_keys(call_count: be == 2, completed_count: be == 2, current_queue_size: be == 0) + ) + end + end + end + end end with "spurious wakeup" do From f1cbe77745140297d4fba19839ac7b990d9b1b2c Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sat, 17 Oct 2026 16:12:24 +1300 Subject: [PATCH 4/4] Move reentrant worker pool test helpers to file scope. --- test/io/event/worker_pool.rb | 178 +++++++++++++++++------------------ 1 file changed, 89 insertions(+), 89 deletions(-) diff --git a/test/io/event/worker_pool.rb b/test/io/event/worker_pool.rb index 7d1fed05..1fb3c5c4 100644 --- a/test/io/event/worker_pool.rb +++ b/test/io/event/worker_pool.rb @@ -9,6 +9,95 @@ return unless defined?(IO::Event::WorkerPool) +# Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing nested blocking operations from inside the worker pool `block`, one per duration in `nested`, each from inside the previous one. +class ReentrantBlockScheduler < IO::Event::TestScheduler + def initialize(nested: [0.01], before_nested: nil, propagate_nested_errors: true, **options) + super(**options) + + @nested = nested.dup + @before_nested = before_nested + @propagate_nested_errors = propagate_nested_errors + @nested_results = [] + end + + # @attribute [Array(Hash)] The results of the nested operations, innermost first. + attr_reader :nested_results + + def nested_result + @nested_results.first + end + + def block(blocker, timeout = nil) + # Simulate a finalizer that performs an offloaded blocking operation: + if blocker.is_a?(IO::Event::WorkerPool) and duration = @nested.shift + @before_nested&.call(self) + + result = IO::Event::WorkerPool.busy(duration: duration) + @nested_results << result + + # `busy` rescues the exception that cancelled it, so propagate it to the outer wait: + raise result[:exception] if @propagate_nested_errors and result[:exception] + end + + super + end +end + +class WaitTimeout < StandardError +end + +class Stop < StandardError +end + +# Interrupt a nested wait after it consumes the outer completion's wakeup. +class InterruptedBlockScheduler < ReentrantBlockScheduler + def initialize(interrupt_before_block: false, **options) + # Finalizers swallow exceptions instead of propagating them to the outer wait: + super(propagate_nested_errors: false, **options) + + @interrupt_before_block = interrupt_before_block + @block_depth = 0 + @interrupted = false + @unblocked = false + @resume_worker = Thread::Queue.new + end + + def unblock(blocker, fiber) + first_completion = blocker.is_a?(IO::Event::WorkerPool) && !@unblocked + @unblocked = true if first_completion + + super + + # Keep the single worker from completing the nested operation until its wait has been interrupted and entered cancellation cleanup: + @resume_worker.pop if first_completion + end + + def block(blocker, timeout = nil) + return super unless blocker.is_a?(IO::Event::WorkerPool) + + @block_depth += 1 + begin + if @block_depth == 2 + if @interrupt_before_block + @interrupt_before_block = false + raise Stop, "before suspension" + end + + @resume_worker.push(true) if @interrupted + end + + super + + if @block_depth == 2 and !@interrupted + @interrupted = true + raise Stop, "after resumption" + end + ensure + @block_depth -= 1 + end + end +end + describe IO::Event::WorkerPool do with "an instance" do let(:worker_pool) {subject.new} @@ -326,95 +415,6 @@ def block(...) end with "scheduler that re-enters the worker pool from block" do - # Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing nested blocking operations from inside the worker pool `block`, one per duration in `nested`, each from inside the previous one. - class ReentrantBlockScheduler < IO::Event::TestScheduler - def initialize(nested: [0.01], before_nested: nil, propagate_nested_errors: true, **options) - super(**options) - - @nested = nested.dup - @before_nested = before_nested - @propagate_nested_errors = propagate_nested_errors - @nested_results = [] - end - - # @attribute [Array(Hash)] The results of the nested operations, innermost first. - attr_reader :nested_results - - def nested_result - @nested_results.first - end - - def block(blocker, timeout = nil) - # Simulate a finalizer that performs an offloaded blocking operation: - if blocker.is_a?(IO::Event::WorkerPool) and duration = @nested.shift - @before_nested&.call(self) - - result = IO::Event::WorkerPool.busy(duration: duration) - @nested_results << result - - # `busy` rescues the exception that cancelled it, so propagate it to the outer wait: - raise result[:exception] if @propagate_nested_errors and result[:exception] - end - - super - end - end - - class WaitTimeout < StandardError - end - - class Stop < StandardError - end - - # Interrupt a nested wait after it consumes the outer completion's wakeup. - class InterruptedBlockScheduler < ReentrantBlockScheduler - def initialize(interrupt_before_block: false, **options) - # Finalizers swallow exceptions instead of propagating them to the outer wait: - super(propagate_nested_errors: false, **options) - - @interrupt_before_block = interrupt_before_block - @block_depth = 0 - @interrupted = false - @unblocked = false - @resume_worker = Thread::Queue.new - end - - def unblock(blocker, fiber) - first_completion = blocker.is_a?(IO::Event::WorkerPool) && !@unblocked - @unblocked = true if first_completion - - super - - # Keep the single worker from completing the nested operation until its wait has been interrupted and entered cancellation cleanup: - @resume_worker.pop if first_completion - end - - def block(blocker, timeout = nil) - return super unless blocker.is_a?(IO::Event::WorkerPool) - - @block_depth += 1 - begin - if @block_depth == 2 - if @interrupt_before_block - @interrupt_before_block = false - raise Stop, "before suspension" - end - - @resume_worker.push(true) if @interrupted - end - - super - - if @block_depth == 2 and !@interrupted - @interrupted = true - raise Stop, "after resumption" - end - ensure - @block_depth -= 1 - end - end - end - # Run `busy(duration:)` in a fiber on a new `ReentrantBlockScheduler`, interrupting it with `WaitTimeout` if it hasn't finished within `timeout` seconds, so a lost wakeup fails the test instead of hanging it. # # @returns [Array(Hash, ReentrantBlockScheduler)] The result of the outer operation, and the scheduler.