diff --git a/ext/io/event/worker_pool.c b/ext/io/event/worker_pool.c index c58ec825..5148fc69 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,27 +335,40 @@ static VALUE worker_pool_work_block(VALUE _work) { return Qnil; } +// 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) { + 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; } 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); @@ -349,8 +382,26 @@ static VALUE worker_pool_work_ensure(VALUE _work) { // will restore and rethrow the original control-flow state after this returns. rb_set_errinfo(Qnil); } + 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 +433,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 +470,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/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/releases.md b/releases.md index b3a99aec..6bd0d0be 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, 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 8ac3543d..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} @@ -324,4 +413,224 @@ def block(...) expect(rescued).to be == true end end + + with "scheduler that re-enters the worker pool from block" do + # 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, scheduler_class: ReentrantBlockScheduler, **options, &block) + result = nil + scheduler = nil + + Thread.new do + scheduler = scheduler_class.new(**options) + Fiber.set_scheduler(scheduler) + + fiber = Fiber.schedule do + result = IO::Event::WorkerPool.busy(duration: duration) + result[:statistics] = scheduler.worker_pool.statistics + end + + Fiber.schedule do + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + + 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 + + block&.call(fiber) + + scheduler.run + 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(scheduler.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 + + 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 + + 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 + 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