Re-introduce support for io_close hook.

Author: samuel-williams-shopify

internal/io.h

@@ -131,6 +131,7 @@ struct rb_io {
     struct ccan_list_head blocking_operations;
     struct rb_execution_context_struct *closing_ec;
     VALUE wakeup_mutex;
+    size_t closing_count;
 
     // The fork generation of the blocking operations list.
     rb_serial_t fork_generation;

io.c

@@ -5546,23 +5546,28 @@ fptr_finalize_flush(rb_io_t *fptr, int noraise, int keepgvl)
         // Need to keep FILE objects of stdin, stdout and stderr, so we are done:
         done = 1;
     }
+    // The check for Qnil here is purely defensive, we should never invoke the fiber scheduler if there is no IO object.
+    else if (fptr->self != Qnil && fptr->closing_count == 0) {
+        VALUE scheduler = rb_fiber_scheduler_current();
+        if (scheduler != Qnil) {
+            fptr->closing_count += 1;
+            VALUE result = rb_fiber_scheduler_io_close(scheduler, fptr->self);
+            fptr->closing_count -= 1;
+
+            if (!UNDEF_P(result)) {
+                // Scheduler handled it:
+                if (fptr->fd == -1) return;
+            }
+        }
+    }
 
     fptr->fd = -1;
     fptr->stdio_file = 0;
     fptr->mode &= ~(FMODE_READABLE|FMODE_WRITABLE);
 
-    // wait for blocking operations to ensure they do not hit EBADF:
+    // Wait for blocking operations to ensure they do not hit EBADF:
     rb_thread_io_close_wait(fptr);
 
-    // Disable for now.
-    // if (!done && fd >= 0) {
-    //     VALUE scheduler = rb_fiber_scheduler_current();
-    //     if (scheduler != Qnil) {
-    //         VALUE result = rb_fiber_scheduler_io_close(scheduler, fptr->self);
-    //         if (!UNDEF_P(result)) done = 1;
-    //     }
-    // }
-
     if (!done && stdio_file) {
         // stdio_file is deallocated anyway even if fclose failed.
         if ((maygvl_fclose(stdio_file, noraise) < 0) && NIL_P(error)) {
@@ -5724,10 +5729,12 @@ io_close_fptr(VALUE io)
     if (!fptr) return 0;
     if (fptr->fd < 0) return 0;
 
+    // This guards against multiple threads closing the same IO object:
     if (rb_thread_io_close_interrupt(fptr)) {
         /* calls close(fptr->fd): */
         fptr_finalize_flush(fptr, FALSE, KEEPGVL);
     }
+
     rb_io_fptr_cleanup(fptr, FALSE);
     return fptr;
 }
@@ -8553,6 +8560,7 @@ rb_io_init_copy(VALUE dest, VALUE io)
     ccan_list_head_init(&fptr->blocking_operations);
     fptr->closing_ec = NULL;
     fptr->wakeup_mutex = Qnil;
+    fptr->closing_count = 0;
     fptr->fork_generation = GET_VM()->fork_gen;
 
     if (!NIL_P(orig->pathv)) fptr->pathv = orig->pathv;
@@ -9283,6 +9291,7 @@ rb_io_open_descriptor(VALUE klass, int descriptor, int mode, VALUE path, VALUE t
     ccan_list_head_init(&io->blocking_operations);
     io->closing_ec = NULL;
     io->wakeup_mutex = Qnil;
+    io->closing_count = 0;
     io->fork_generation = GET_VM()->fork_gen;
 
     if (encoding) {
@@ -9427,6 +9436,7 @@ rb_io_fptr_new(void)
     ccan_list_head_init(&fp->blocking_operations);
     fp->closing_ec = NULL;
     fp->wakeup_mutex = Qnil;
+    fp->closing_count = 0;
     fp->fork_generation = GET_VM()->fork_gen;
     return fp;
 }
@@ -9561,6 +9571,7 @@ io_initialize(VALUE io, VALUE fnum, VALUE vmode, VALUE opt)
     ccan_list_head_init(&fp->blocking_operations);
     fp->closing_ec = NULL;
     fp->wakeup_mutex = Qnil;
+    fp->closing_count = 0;
     fp->fork_generation = GET_VM()->fork_gen;
     clear_codeconv(fp);
     io_check_tty(fp);

test/fiber/scheduler.rb

@@ -255,6 +255,12 @@ def io_select(...)
     end.value
   end
 
+  # This hook is invoked by `IO#close`. Using a separate IO object
+  # demonstrates that the close operation is asynchronous.
+  def io_close(io)
+    io.close
+  end
+
   # This hook is invoked by `Kernel#sleep` and `Thread::Mutex#sleep`.
   def kernel_sleep(duration = nil)
     # $stderr.puts [__method__, duration, Fiber.current].inspect

thread.c

@@ -1773,6 +1773,14 @@ rb_io_blocking_operations(struct rb_io *io)
 static void
 rb_io_blocking_operation_enter(struct rb_io *io, struct rb_io_blocking_operation *blocking_operation)
 {
+    // Prevent new blocking operations from starting if the IO is being closed.
+    if (io->closing_ec || io->fd == -1) {
+        rb_execution_context_t *ec = GET_EC();
+        rb_vm_t *vm = ec->thread_ptr->vm;
+        VALUE error = vm->special_exceptions[ruby_error_stream_closed];
+        rb_exc_raise(error);
+    }
+
     ccan_list_add(rb_io_blocking_operations(io), &blocking_operation->list);
 }
 
@@ -2895,7 +2903,6 @@ rb_thread_io_close_wait(struct rb_io* io)
 
     // We are done closing:
     io->wakeup_mutex = Qnil;
-    io->closing_ec = NULL;
 }
 
 void