Re-introduce support for io_close hook.

Author: samuel-williams-shopify

io.c

@@ -5540,29 +5540,31 @@ fptr_finalize_flush(rb_io_t *fptr, int noraise, int keepgvl)
         }
     }
 
+    // Wait for blocking operations to ensure they do not hit EBADF:
+    rb_thread_io_close_wait(fptr);
+
     int done = 0;
 
     if (RUBY_IO_EXTERNAL_P(fptr) || fd <= 2) {
         // 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) {
+        VALUE scheduler = rb_fiber_scheduler_current();
+        if (scheduler != Qnil) {
+            VALUE result = rb_fiber_scheduler_io_close(scheduler, fptr->self);
+            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:
-    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 +5726,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;
 }
@@ -5743,6 +5747,7 @@ fptr_waitpid(rb_io_t *fptr, int nohang)
     }
 }
 
+
 VALUE
 rb_io_close(VALUE io)
 {

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)
+    Fiber.blocking{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