From a6b2ee3f50ee6c589b1d213074323684e113c1ae Mon Sep 17 00:00:00 2001 From: Martin Ek Date: Fri, 9 Oct 2026 13:23:43 -0700 Subject: [PATCH 1/2] 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 d08ceef2d7ac3b68bcd0791690690bf5a716ac76 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sat, 17 Oct 2026 16:53:23 +1300 Subject: [PATCH 2/2] Prevent worker pool re-entry with a blocking fiber scope. --- ext/io/event/worker_pool.c | 50 ++++++----- releases.md | 1 + test/io/event/worker_pool.rb | 169 ++++++++++++++++++++++++++++++----- 3 files changed, 180 insertions(+), 40 deletions(-) diff --git a/ext/io/event/worker_pool.c b/ext/io/event/worker_pool.c index c58ec825..372b2530 100644 --- a/ext/io/event/worker_pool.c +++ b/ext/io/event/worker_pool.c @@ -1,5 +1,5 @@ // Released under the MIT License. -// Copyright, 2025, by Samuel Williams. +// Copyright, 2025-2026, by Samuel Williams. #include "worker_pool.h" #include "worker_pool_test.h" @@ -19,7 +19,9 @@ enum { }; static VALUE IO_Event_WorkerPool; +static VALUE rb_Fiber; static ID id_maximum_worker_count; +static ID id_blocking; // Thread pool structure struct IO_Event_WorkerPool_Worker { @@ -318,17 +320,11 @@ static VALUE worker_pool_work_block(VALUE _work) { static VALUE worker_pool_work_wait(VALUE _work) { struct IO_Event_WorkerPool_Work *work = (void*)_work; - while (true) { + // A stale unblock is not cancellation. Only a non-local exit cancels work. + do { 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); - } + } while (!work->completed); return Qtrue; } @@ -354,6 +350,21 @@ static VALUE worker_pool_work_ensure(VALUE _work) { return Qnil; } +static VALUE worker_pool_work_submit(RB_BLOCK_CALL_FUNC_ARGLIST(fiber, _work)) { + struct IO_Event_WorkerPool_Work *work = (void*)_work; + struct IO_Event_WorkerPool *pool; + TypedData_Get_Struct(work->blocker, struct IO_Event_WorkerPool, &IO_Event_WorkerPool_type, pool); + + pthread_mutex_lock(&pool->mutex); + enqueue_work(pool, work); + pthread_cond_signal(&pool->work_available); + pthread_mutex_unlock(&pool->mutex); + + // Keep cancellation and draining inside the blocking scope too. rb_ensure + // preserves the original exception or other non-local exit. + return rb_ensure(worker_pool_work_wait, _work, worker_pool_work_ensure, _work); +} + // Ruby method to submit work and wait for completion static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) { struct IO_Event_WorkerPool *pool; @@ -389,17 +400,13 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) { .fiber = fiber, .next = NULL }; - - // Enqueue work: - pthread_mutex_lock(&pool->mutex); - enqueue_work(pool, &work); - pthread_cond_signal(&pool->work_available); - pthread_mutex_unlock(&pool->mutex); - // Block the current fiber until work is completed. If blocking exits via an - // exception or another non-local jump, ensure the work is cancelled and fully - // drained before Ruby restores and rethrows the original control-flow state. - return rb_ensure(worker_pool_work_wait, (VALUE)&work, worker_pool_work_ensure, (VALUE)&work); + // A finalizer inside the scheduler's block hook must not offload another + // operation on this fiber and consume this work's wakeup. Fiber.blocking + // makes nested operations synchronous, while explicit fiber transfers still + // work and other fibers retain their own blocking state. Capture the scheduler + // above, before Fiber.blocking hides it from rb_fiber_scheduler_current(). + return rb_block_call(rb_Fiber, id_blocking, 0, NULL, worker_pool_work_submit, (VALUE)&work); } static VALUE worker_pool_allocate(VALUE klass) { @@ -495,6 +502,9 @@ static VALUE worker_pool_statistics(VALUE self) { void Init_IO_Event_WorkerPool(VALUE IO_Event) { // Initialize symbols id_maximum_worker_count = rb_intern("maximum_worker_count"); + id_blocking = rb_intern("blocking"); + rb_Fiber = rb_const_get(rb_cObject, rb_intern("Fiber")); + rb_global_variable(&rb_Fiber); IO_Event_WorkerPool = rb_define_class_under(IO_Event, "WorkerPool", rb_cObject); rb_define_alloc_func(IO_Event_WorkerPool, worker_pool_allocate); diff --git a/releases.md b/releases.md index b3a99aec..9ac97816 100644 --- a/releases.md +++ b/releases.md @@ -2,6 +2,7 @@ ## Unreleased + - Prevent nested worker-pool waits by running `WorkerPool#call` in a `Fiber.blocking` scope. Blocking operations triggered inside scheduler callbacks, including finalizers, execute synchronously. Cancellation still drains submitted work, and spurious wakeups no longer cancel operations. - 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. ## v1.22.1 diff --git a/test/io/event/worker_pool.rb b/test/io/event/worker_pool.rb index 283c419a..61c672b4 100644 --- a/test/io/event/worker_pool.rb +++ b/test/io/event/worker_pool.rb @@ -9,6 +9,56 @@ return unless defined?(IO::Event::WorkerPool) +# Simulate a finalizer performing a blocking operation before the scheduler suspends the fiber. The operation must run synchronously to avoid consuming the outer operation's wakeup. +class ReentrantBlockScheduler < IO::Event::TestScheduler + attr_reader :nested_result, :blocking_states + + def initialize(...) + super + @blocking_states = [] + end + + def block(blocker, timeout = nil) + @blocking_states << Fiber.current.blocking? if blocker.is_a?(IO::Event::WorkerPool) + + # 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 + +# Exercise both exceptional exits from block and synchronous operations during cancellation cleanup. +class InterruptedBlockScheduler < IO::Event::TestScheduler + attr_reader :nested_result, :blocking_states + + def initialize(...) + super + @blocking_states = [] + end + + def block(blocker, timeout = nil) + return super unless blocker.is_a?(IO::Event::WorkerPool) + + @blocking_states << Fiber.current.blocking? + case @blocking_states.size + when 1 + raise WaitTimeout, "original interruption" + when 2 + @nested_result = IO::Event::WorkerPool.busy(duration: 0.01) + raise WaitTimeout, "cleanup interruption" + end + + super + end +end + describe IO::Event::WorkerPool do with "an instance" do let(:worker_pool) {subject.new} @@ -303,6 +353,7 @@ def block(...) it "preserves non-exception control flow" do result = nil rescued = false + blocking = nil Thread.new do scheduler = ThrowingBlockScheduler.new @@ -313,6 +364,7 @@ def block(...) IO::Event::WorkerPool.busy(duration: 2.0) :missing end + blocking = Fiber.current.blocking? end rescued = scheduler.rescued ensure @@ -322,39 +374,26 @@ def block(...) expect(result).to be == :expected expect(rescued).to be == true + expect(blocking).to be == false 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 - + with "blocking operations inside the scheduler block hook" do it "wakes the outer operation and runs the nested operation" do result = nil nested_result = nil + statistics = nil + blocking_states = nil + blocking = nil Thread.new do - # A single worker guarantees that the nested operation queues behind the outer one: + # The nested operation must finish synchronously without using the single worker: scheduler = ReentrantBlockScheduler.new(maximum_worker_count: 1) Fiber.set_scheduler(scheduler) fiber = Fiber.schedule do result = IO::Event::WorkerPool.busy(duration: 0.1) + blocking = Fiber.current.blocking? end # Bound the outer wait so a lost wakeup fails the test instead of hanging it: @@ -370,6 +409,8 @@ class WaitTimeout < StandardError scheduler.run nested_result = scheduler.nested_result + statistics = scheduler.worker_pool.statistics + blocking_states = scheduler.blocking_states ensure Fiber.set_scheduler(nil) scheduler&.close @@ -387,6 +428,94 @@ class WaitTimeout < StandardError executed: be == true, result: be == :completed ) + expect(statistics).to have_keys(call_count: be == 1, completed_count: be == 1, current_queue_size: be == 0) + expect(blocking_states).to be == [true] + expect(blocking).to be == false + end + end + + with "a blocking wait" do + it "allows other fibers to offload operations while waiting" do + results = [] + blocking_states = [] + statistics = nil + + Thread.new do + scheduler = IO::Event::TestScheduler.new(maximum_worker_count: 1) + Fiber.set_scheduler(scheduler) + + 2.times do + Fiber.schedule do + blocking_states << Fiber.current.blocking? + results << IO::Event::WorkerPool.busy(duration: 0.01) + blocking_states << Fiber.current.blocking? + end + end + + # Both fibers have submitted their operations and suspended: + statistics = scheduler.worker_pool.statistics + scheduler.run + ensure + Fiber.set_scheduler(nil) + end.value + + expect(statistics[:call_count]).to be == 2 + expect(results.size).to be == 2 + results.each do |result| + expect(result).to have_keys(executed: be == true, result: be == :completed) + end + expect(blocking_states).to be == [false, false, false, false] + end + + it "keeps cancellation cleanup blocking and restores the fiber after an exception" do + result = nil + next_result = nil + blocking = nil + statistics = nil + scheduler = nil + + Thread.new do + scheduler = InterruptedBlockScheduler.new + Fiber.set_scheduler(scheduler) + + Fiber.schedule do + result = IO::Event::WorkerPool.busy(duration: 2.0) + blocking = Fiber.current.blocking? + next_result = IO::Event::WorkerPool.busy(duration: 0.01) + end + + scheduler.run + statistics = scheduler.worker_pool.statistics + ensure + Fiber.set_scheduler(nil) + end.value + + expect(result).to have_keys(result: be == :exception, exception: be_a(WaitTimeout).and(have_attributes(message: be == "original interruption"))) + expect(scheduler.nested_result).to have_keys(executed: be == true, result: be == :completed) + expect(next_result).to have_keys(executed: be == true, result: be == :completed) + expect(statistics).to have_keys(call_count: be == 2, completed_count: be == 2, current_queue_size: be == 0) + expect(scheduler.blocking_states).not.to be(:include?, false) + expect(blocking).to be == false + end + + it "does not cancel work on a spurious wakeup" do + result = nil + + Thread.new do + scheduler = IO::Event::TestScheduler.new + Fiber.set_scheduler(scheduler) + + fiber = Fiber.schedule do + result = IO::Event::WorkerPool.busy(duration: 0.1) + end + + scheduler.unblock(nil, fiber) + scheduler.run + ensure + Fiber.set_scheduler(nil) + end.value + + expect(result).to have_keys(executed: be == true, result: be == :completed) end end end