diff --git a/doc/language/fiber.md b/doc/language/fiber.md index 2107a7b9934..ff029a7ec8c 100644 --- a/doc/language/fiber.md +++ b/doc/language/fiber.md @@ -234,7 +234,7 @@ non-blocking*, the operation will invoke the scheduler. Closing an IO interrupts all blocking operations on that IO. When a thread calls `IO#close`, it first attempts to interrupt any threads or fibers that are blocked on that IO. The closing thread waits until all blocked threads and fibers have been properly interrupted and removed from the IO's blocking list. Each interrupted thread or fiber receives an `IOError` and is cleanly removed from the blocking operation. Only after all blocking operations have been interrupted and cleaned up will the actual file descriptor be closed, ensuring proper resource cleanup and preventing potential race conditions. -For fibers managed by a scheduler, the interruption process involves calling `rb_fiber_scheduler_fiber_interrupt` on the scheduler. The corresponding `Fiber::Scheduler#fiber_interrupt` hook is required. This allows the scheduler to handle the interruption in a way that's appropriate for its event loop implementation. The scheduler can then notify the fiber, which will receive an `IOError` and be removed from the blocking operation. This mechanism ensures that fiber-based concurrency works correctly with IO operations, even when those operations are interrupted by `IO#close`. +For fibers managed by a scheduler, the interruption process involves calling `rb_fiber_scheduler_fiber_interrupt` on the scheduler. The corresponding `Fiber::Scheduler#fiber_interrupt` hook is required. For IO operations, the first argument is an operation-scoped proxy which responds to `alive?`, `raise`, and `transfer`, matching the subset of the Fiber interface needed to deliver an interruption. The scheduler can raise the supplied exception on the target, or enqueue the target directly and later call `transfer` to raise its stored exception. If the operation completes before the queued interruption is processed, `alive?` returns false and both `raise` and `transfer` have no effect. This prevents a delayed exception from escaping the operation which caused it and interrupting a later operation on the same fiber. ```mermaid sequenceDiagram @@ -266,16 +266,16 @@ sequenceDiagram ThreadB->>IO: thread_io_close_notify_all Note over ThreadB: rb_mutex_sleep - IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Fiber1) - Scheduler->>Fiber1: fiber_interrupt with IOError + IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Target1) + Scheduler->>Target1: raise IOError if alive? activate Fiber1 Note over IO: fiber_interrupt causes removal from blocking list Fiber1->>IO: rb_io_blocking_operation_exit() IO-->>ThreadB: Wakeup thread deactivate Fiber1 - IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Fiber2) - Scheduler->>Fiber2: fiber_interrupt with IOError + IO->>Scheduler: rb_fiber_scheduler_fiber_interrupt(Target2) + Scheduler->>Target2: raise IOError if alive? activate Fiber2 Note over IO: fiber_interrupt causes removal from blocking list Fiber2->>IO: rb_io_blocking_operation_exit() diff --git a/include/ruby/fiber/scheduler.h b/include/ruby/fiber/scheduler.h index 73a1ed40640..10eb1352777 100644 --- a/include/ruby/fiber/scheduler.h +++ b/include/ruby/fiber/scheduler.h @@ -25,7 +25,8 @@ RBIMPL_SYMBOL_EXPORT_BEGIN() // Version 3: Adds support for `fiber_interrupt`. // Version 4: IO hooks use single-transfer `(offset, length)` semantics. -#define RUBY_FIBER_SCHEDULER_VERSION 4 +// Version 5: `fiber_interrupt` may receive an operation-scoped proxy. +#define RUBY_FIBER_SCHEDULER_VERSION 5 struct timeval; struct rb_thread_struct; @@ -475,16 +476,22 @@ int rb_fiber_scheduler_blocking_operation_cancel(rb_fiber_scheduler_blocking_ope VALUE rb_fiber_scheduler_blocking_operation_wait(VALUE scheduler, void* (*function)(void *), void *data, rb_unblock_function_t *unblock_function, void *data2, int flags, struct rb_fiber_scheduler_blocking_operation_state *state); /** - * Interrupt a fiber by raising an exception. You can construct an exception using `rb_make_exception`. + * Interrupt a target by raising an exception. You can construct an exception using `rb_make_exception`. + * + * The target is usually a Fiber. For an IO operation, it may instead be an + * operation-scoped proxy which responds to `alive?`, `raise`, and `transfer`. + * Calling `transfer` raises the stored exception in the target Fiber, allowing + * the proxy to be queued directly. Schedulers should restrict their use of the + * target to those methods. * * This hook may be invoked by a different thread. * * @param[in] scheduler Target scheduler. - * @param[in] fiber The fiber to interrupt. + * @param[in] target The Fiber or operation-scoped proxy to interrupt. * @param[in] exception The exception to raise in the fiber. * @return What `scheduler.fiber_interrupt` returns. */ -VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE fiber, VALUE exception); +VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE target, VALUE exception); /** * Create and schedule a non-blocking fiber. diff --git a/internal/io.h b/internal/io.h index 2110f0b0876..f47aba42d1e 100644 --- a/internal/io.h +++ b/internal/io.h @@ -27,6 +27,9 @@ struct rb_io_blocking_operation { // The execution context of the blocking operation. struct rb_execution_context_struct *ec; + + // An operation-scoped target passed to the Fiber scheduler if interrupted. + VALUE scheduler_interrupt_target; }; /** Ruby's IO, metadata and buffers. */ diff --git a/internal/scheduler.h b/internal/scheduler.h new file mode 100644 index 00000000000..f79b5d02a6d --- /dev/null +++ b/internal/scheduler.h @@ -0,0 +1,7 @@ +#pragma once + +#include "ruby/ruby.h" + +VALUE rb_fiber_scheduler_interrupt_target_new(VALUE fiber, VALUE exception); +VALUE rb_fiber_scheduler_interrupt_target_exception(VALUE target); +void rb_fiber_scheduler_interrupt_target_invalidate(VALUE target); diff --git a/scheduler.c b/scheduler.c index d315b275fba..0c955a28747 100644 --- a/scheduler.c +++ b/scheduler.c @@ -17,6 +17,7 @@ #include "ruby/thread.h" // For `ruby_thread_has_gvl_p`: +#include "internal/scheduler.h" #include "internal/thread.h" // For atomic operations: @@ -51,6 +52,118 @@ static ID id_fiber_schedule; // Our custom blocking operation class static VALUE rb_cFiberSchedulerBlockingOperation; +static VALUE rb_cFiberSchedulerInterruptTarget; + +struct rb_fiber_scheduler_interrupt_target { + VALUE fiber; + VALUE exception; + bool active; +}; + +static void +interrupt_target_mark(void *ptr) +{ + struct rb_fiber_scheduler_interrupt_target *target = ptr; + + rb_gc_mark(target->fiber); + rb_gc_mark(target->exception); +} + +static size_t +interrupt_target_memsize(const void *ptr) +{ + return sizeof(struct rb_fiber_scheduler_interrupt_target); +} + +static const rb_data_type_t interrupt_target_data_type = { + "Fiber::Scheduler::InterruptTarget", + { + interrupt_target_mark, + RUBY_DEFAULT_FREE, + interrupt_target_memsize, + }, + 0, 0, RUBY_TYPED_FREE_IMMEDIATELY +}; + +static struct rb_fiber_scheduler_interrupt_target * +get_interrupt_target(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target; + TypedData_Get_Struct(self, struct rb_fiber_scheduler_interrupt_target, &interrupt_target_data_type, target); + return target; +} + +static VALUE +interrupt_target_alive_p(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (!target->active) { + return Qfalse; + } + + return rb_fiber_alive_p(target->fiber); +} + +static VALUE +interrupt_target_raise(int argc, VALUE *argv, VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (!target->active) { + return Qnil; + } + + return rb_fiber_raise(target->fiber, argc, argv); +} + +static VALUE +interrupt_target_transfer(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (!target->active || !RTEST(rb_fiber_alive_p(target->fiber))) { + return Qnil; + } + + VALUE exception = target->exception; + return rb_fiber_raise(target->fiber, 1, &exception); +} + +VALUE +rb_fiber_scheduler_interrupt_target_new(VALUE fiber, VALUE exception) +{ + struct rb_fiber_scheduler_interrupt_target *target; + VALUE self = TypedData_Make_Struct(rb_cFiberSchedulerInterruptTarget, struct rb_fiber_scheduler_interrupt_target, &interrupt_target_data_type, target); + + target->fiber = fiber; + target->exception = exception; + target->active = true; + + return self; +} + +VALUE +rb_fiber_scheduler_interrupt_target_exception(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + if (target->active) { + return target->exception; + } + + return Qnil; +} + +void +rb_fiber_scheduler_interrupt_target_invalidate(VALUE self) +{ + struct rb_fiber_scheduler_interrupt_target *target = get_interrupt_target(self); + + target->active = false; + target->fiber = Qnil; + target->exception = Qnil; +} /* * Custom blocking operation structure for blocking operations @@ -344,6 +457,15 @@ Init_Fiber_Scheduler(void) // Register the anonymous class as a GC root so it doesn't get collected rb_gc_register_mark_object(rb_cFiberSchedulerBlockingOperation); + // Define an anonymous interrupt target class. Instances are passed to + // fiber_interrupt and cannot be instantiated directly. + rb_cFiberSchedulerInterruptTarget = rb_class_new(rb_cObject); + rb_undef_alloc_func(rb_cFiberSchedulerInterruptTarget); + rb_define_method(rb_cFiberSchedulerInterruptTarget, "alive?", interrupt_target_alive_p, 0); + rb_define_method(rb_cFiberSchedulerInterruptTarget, "raise", interrupt_target_raise, -1); + rb_define_method(rb_cFiberSchedulerInterruptTarget, "transfer", interrupt_target_transfer, 0); + rb_gc_register_mark_object(rb_cFiberSchedulerInterruptTarget); + #if 0 /* for RDoc */ rb_cFiberScheduler = rb_define_class_under(rb_cFiber, "Scheduler", rb_cObject); rb_define_method(rb_cFiberScheduler, "close", rb_fiber_scheduler_close, 0); @@ -1154,17 +1276,21 @@ VALUE rb_fiber_scheduler_blocking_operation_wait(VALUE scheduler, void* (*functi /* * Document-method: Fiber::Scheduler#fiber_interrupt - * call-seq: fiber_interrupt(fiber, exception) + * call-seq: fiber_interrupt(target, exception) * - * Invoked by Ruby's core methods to notify the scheduler that the blocked fiber should be interrupted - * with an exception. For example, IO#close uses this method to interrupt fibers that are performing - * blocking IO operations. + * Invoked by Ruby's core methods to notify the scheduler that a blocked fiber + * should be interrupted with an exception. For IO operations, +target+ is an + * operation-scoped proxy which responds to #alive?, #raise, and #transfer. The + * scheduler should enqueue the target on its owning thread and use only those + * methods. #transfer raises the interruption's stored exception in the target. + * Once the operation completes, #alive? returns false and #raise has no effect, + * preventing a delayed exception from escaping into a later operation. * */ -VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE fiber, VALUE exception) +VALUE rb_fiber_scheduler_fiber_interrupt(VALUE scheduler, VALUE target, VALUE exception) { VALUE arguments[] = { - fiber, exception + target, exception }; VALUE result; diff --git a/test/fiber/scheduler.rb b/test/fiber/scheduler.rb index af66ba10600..2cb04812fa2 100644 --- a/test/fiber/scheduler.rb +++ b/test/fiber/scheduler.rb @@ -313,24 +313,9 @@ class Scheduler io.write_nonblock('.') end - class FiberInterrupt - def initialize(fiber, exception) - @fiber = fiber - @exception = exception - end - - def alive? - @fiber.alive? - end - - def transfer - @fiber.raise(@exception) - end - end - - def fiber_interrupt(fiber, exception) + def fiber_interrupt(target, _exception) @lock.synchronize do - @ready << FiberInterrupt.new(fiber, exception) + @ready << target end io = @urgent.last diff --git a/test/fiber/test_io_close.rb b/test/fiber/test_io_close.rb index 742b40841d9..14c6731acbb 100644 --- a/test/fiber/test_io_close.rb +++ b/test/fiber/test_io_close.rb @@ -1,5 +1,6 @@ # frozen_string_literal: true require 'test/unit' +require 'io/wait' require_relative 'scheduler' class TestFiberIOClose < Test::Unit::TestCase @@ -44,6 +45,46 @@ class TestFiberIOClose < Test::Unit::TestCase end end + def test_io_close_interrupt_does_not_escape_blocking_operation + with_socket_pair do |source, source_peer| + with_socket_pair do |unrelated, unrelated_peer| + errors = [] + + thread = Thread.new do + scheduler = Scheduler.new + Fiber.set_scheduler scheduler + + 5.times do + Fiber.schedule do + begin + source.wait_readable(0.01) + source.close + rescue => error + errors << [:source, error] + end + + begin + unrelated.wait_readable(0.01) + rescue => error + errors << [:unrelated, error] + end + end + end + end + + thread.join + + assert_equal 4, errors.size + assert_equal [:source], errors.map(&:first).uniq + errors.each do |location, error| + assert_equal :source, location + assert_instance_of IOError, error + assert_match(/closed/, error.message) + end + end + end + end + def test_io_close_blocking_thread omit "Interrupting a io_wait read is not supported!" if RUBY_PLATFORM =~ /mswin|mingw/ @@ -80,9 +121,19 @@ class TestFiberIOClose < Test::Unit::TestCase with_socket_pair do |i, o| error = nil + scheduler = nil + + scheduler_class = Class.new(Scheduler) do + attr_reader :interrupt_target + + def fiber_interrupt(target, exception) + @interrupt_target = target + super + end + end thread = Thread.new do - scheduler = Scheduler.new + scheduler = scheduler_class.new Fiber.set_scheduler scheduler Fiber.schedule do @@ -102,6 +153,11 @@ class TestFiberIOClose < Test::Unit::TestCase assert_instance_of IOError, error assert_match(/closed/, error.message) + + interrupt_target = scheduler.interrupt_target + assert_not_predicate interrupt_target, :alive? + assert_nil interrupt_target.transfer + assert_nil interrupt_target.raise(IOError.new) end end end diff --git a/thread.c b/thread.c index 5de075cf73a..992f8119088 100644 --- a/thread.c +++ b/thread.c @@ -87,6 +87,7 @@ #include "internal/object.h" #include "internal/proc.h" #include "ruby/fiber/scheduler.h" +#include "internal/scheduler.h" #include "internal/signal.h" #include "internal/thread.h" #include "internal/time.h" @@ -2023,6 +2024,10 @@ rb_io_blocking_operation_exit(struct rb_io *io, struct rb_io_blocking_operation // Indicate that the blocking operation is no longer active: blocking_operation->ec = NULL; + if (!NIL_P(blocking_operation->scheduler_interrupt_target)) { + rb_fiber_scheduler_interrupt_target_invalidate(blocking_operation->scheduler_interrupt_target); + } + if (RB_TEST(wakeup_mutex)) { struct io_blocking_operation_arguments arguments = { .io = io, @@ -2047,6 +2052,30 @@ rb_thread_io_blocking_operation_ensure(VALUE _argument) return Qnil; } +struct thread_io_blocking_operation_arguments { + VALUE (*function)(VALUE); + VALUE argument; + struct rb_io_blocking_operation *blocking_operation; +}; + +static VALUE +rb_thread_io_blocking_operation_body(VALUE _arguments) +{ + struct thread_io_blocking_operation_arguments *arguments = (void *)_arguments; + VALUE result = arguments->function(arguments->argument); + VALUE target = arguments->blocking_operation->scheduler_interrupt_target; + + if (!NIL_P(target)) { + VALUE exception = rb_fiber_scheduler_interrupt_target_exception(target); + + if (!NIL_P(exception)) { + rb_exc_raise(exception); + } + } + + return result; +} + /* * Executes a function that performs a blocking IO operation, while properly tracking * the operation in the IO's blocking_operations list. This ensures proper cleanup @@ -2069,6 +2098,7 @@ rb_thread_io_blocking_operation(VALUE self, VALUE(*function)(VALUE), VALUE argum rb_execution_context_t *ec = GET_EC(); struct rb_io_blocking_operation blocking_operation = { .ec = ec, + .scheduler_interrupt_target = Qnil, }; rb_io_blocking_operation_enter(io, &blocking_operation); @@ -2077,7 +2107,13 @@ rb_thread_io_blocking_operation(VALUE self, VALUE(*function)(VALUE), VALUE argum .blocking_operation = &blocking_operation }; - return rb_ensure(function, argument, rb_thread_io_blocking_operation_ensure, (VALUE)&io_blocking_operation_arguments); + struct thread_io_blocking_operation_arguments arguments = { + .function = function, + .argument = argument, + .blocking_operation = &blocking_operation, + }; + + return rb_ensure(rb_thread_io_blocking_operation_body, (VALUE)&arguments, rb_thread_io_blocking_operation_ensure, (VALUE)&io_blocking_operation_arguments); } static bool @@ -2187,6 +2223,7 @@ rb_thread_io_blocking_call(struct rb_io* io, rb_blocking_function_t *func, void struct rb_io_blocking_operation blocking_operation = { .ec = ec, + .scheduler_interrupt_target = Qnil, }; rb_io_blocking_operation_enter(io, &blocking_operation); @@ -3086,7 +3123,16 @@ thread_io_close_notify_all(VALUE _io) rb_thread_t *thread = ec->thread_ptr; if (thread->scheduler != Qnil) { - rb_fiber_scheduler_fiber_interrupt(thread->scheduler, rb_fiberptr_self(ec->fiber_ptr), error); + VALUE target = blocking_operation->scheduler_interrupt_target; + + if (NIL_P(target)) { + VALUE fiber = rb_fiberptr_self(ec->fiber_ptr); + target = rb_fiber_scheduler_interrupt_target_new(fiber, error); + blocking_operation->scheduler_interrupt_target = target; + } + + rb_fiber_scheduler_fiber_interrupt(thread->scheduler, target, error); + RB_GC_GUARD(target); } else { // If the thread is not the current thread, we need to enqueue an error: @@ -4884,6 +4930,7 @@ thread_io_wait(rb_thread_t *th, struct rb_io *io, int fd, int events, struct tim if (io) { blocking_operation.ec = ec; + blocking_operation.scheduler_interrupt_target = Qnil; COMPILER_WARNING_PUSH #if RBIMPL_COMPILER_SINCE(GCC, 12, 0, 0) COMPILER_WARNING_IGNORED(-Wdangling-pointer) @@ -5054,12 +5101,14 @@ thread_io_wait(rb_thread_t *th, struct rb_io *io, int fd, int events, struct tim if (io) { args.io = io; blocking_operation.ec = th->ec; + blocking_operation.scheduler_interrupt_target = Qnil; rb_io_blocking_operation_enter(io, &blocking_operation); args.blocking_operation = &blocking_operation; } else { args.io = NULL; blocking_operation.ec = NULL; + blocking_operation.scheduler_interrupt_target = Qnil; args.blocking_operation = NULL; }