Scope fiber_interrupt with an operation proxy. (#18793)

[Bug #21918]

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()

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.

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. */

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);

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;

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

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

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;
     }