Fiber scheduler: invoke #io_write hook on IO flush (ruby#15609)

Previously, calling IO#flush or closing an IO with unflushed buffered
writes would just invoke `#blocking_operation_wait` and flush the write
buffer using a `write` syscall. This change adds flushing through the
fiber scheduler by invoking the `#io_write` hook.

* Prefer IO::Buffer#write in IOScheduler

* Use Dir.tmpdir for test file

* Correctly handle errors in io_flush_buffer_fiber_scheduler

Author: noteflakes

io.c

@@ -1418,10 +1418,34 @@ io_flush_buffer_sync(void *arg)
     return (VALUE)-1;
 }
 
+static inline VALUE
+io_flush_buffer_fiber_scheduler(VALUE scheduler, rb_io_t *fptr)
+{
+    VALUE ret = rb_fiber_scheduler_io_write_memory(scheduler, fptr->self, fptr->wbuf.ptr+fptr->wbuf.off, fptr->wbuf.len, 0);
+    if (!UNDEF_P(ret)) {
+        ssize_t result = rb_fiber_scheduler_io_result_apply(ret);
+        if (result > 0) {
+            fptr->wbuf.off += result;
+            fptr->wbuf.len -= result;
+        }
+        return result >= 0 ? (VALUE)0 : (VALUE)-1;
+    }
+    return ret;
+}
+
 static VALUE
 io_flush_buffer_async(VALUE arg)
 {
     rb_io_t *fptr = (rb_io_t *)arg;
+
+    VALUE scheduler = rb_fiber_scheduler_current();
+    if (scheduler != Qnil) {
+        VALUE result = io_flush_buffer_fiber_scheduler(scheduler, fptr);
+        if (!UNDEF_P(result)) {
+            return result;
+        }
+    }
+
     return rb_io_blocking_region_wait(fptr, io_flush_buffer_sync, fptr, RUBY_IO_WRITABLE);
 }
 

test/fiber/scheduler.rb

@@ -488,6 +488,29 @@ def blocking(&block)
   end
 end
 
+class IOScheduler < Scheduler
+  def __io_ops__
+    @__io_ops__ ||= []
+  end
+
+  def io_write(io, buffer, length, offset)
+    fd = io.fileno
+    str = buffer.get_string
+    __io_ops__ << [:io_write, fd, str]
+    Fiber.blocking { buffer.write(IO.for_fd(fd), 0, offset) }
+  end
+end
+
+class IOErrorScheduler < Scheduler
+  def io_read(io, buffer, length, offset)
+    return -Errno::EBADF::Errno
+  end
+
+  def io_write(io, buffer, length, offset)
+    return -Errno::EINVAL::Errno
+  end
+end
+
 # This scheduler has a broken implementation of `unblock`` in the sense that it
 # raises an exception. This is used to test the behavior of the scheduler when
 # unblock raises an exception.

test/fiber/test_scheduler.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 require 'test/unit'
+require 'securerandom'
+require 'fileutils'
 require_relative 'scheduler'
 
 class TestFiberScheduler < Test::Unit::TestCase
@@ -283,4 +285,91 @@ def test_post_fork_fiber_blocking
   ensure
     thread.kill rescue nil
   end
+
+  def test_io_write_on_flush
+    fn = File.join(Dir.tmpdir, "ruby_test_io_write_on_flush_#{SecureRandom.hex}")
+    write_fd = nil
+    io_ops = nil
+    thread = Thread.new do
+      scheduler = IOScheduler.new
+      Fiber.set_scheduler scheduler
+
+      Fiber.schedule do
+        File.open(fn, 'w+') do |f|
+          write_fd = f.fileno
+          f << 'foo'
+          f.flush
+          f << 'bar'
+        end
+      end
+      io_ops = scheduler.__io_ops__
+    end
+    thread.join
+    assert_equal [
+      [:io_write, write_fd, 'foo'],
+      [:io_write, write_fd, 'bar']
+    ], io_ops
+
+    assert_equal 'foobar', IO.read(fn)
+  ensure
+    thread.kill rescue nil
+    FileUtils.rm_f(fn)
+  end
+
+  def test_io_read_error
+    fn = File.join(Dir.tmpdir, "ruby_test_io_read_error_#{SecureRandom.hex}")
+    exception = nil
+    thread = Thread.new do
+      scheduler = IOErrorScheduler.new
+      Fiber.set_scheduler scheduler
+      Fiber.schedule do
+        File.open(fn, 'w+') { it.read }
+      rescue => e
+        exception = e
+      end
+    end
+    thread.join
+    assert_kind_of Errno::EBADF, exception
+  ensure
+    thread.kill rescue nil
+    FileUtils.rm_f(fn)
+  end
+
+  def test_io_write_error
+    fn = File.join(Dir.tmpdir, "ruby_test_io_write_error_#{SecureRandom.hex}")
+    exception = nil
+    thread = Thread.new do
+      scheduler = IOErrorScheduler.new
+      Fiber.set_scheduler scheduler
+      Fiber.schedule do
+        File.open(fn, 'w+') { it.sync = true; it << 'foo' }
+      rescue => e
+        exception = e
+      end
+    end
+    thread.join
+    assert_kind_of Errno::EINVAL, exception
+  ensure
+    thread.kill rescue nil
+    FileUtils.rm_f(fn)
+  end
+
+  def test_io_write_flush_error
+    fn = File.join(Dir.tmpdir, "ruby_test_io_write_flush_error_#{SecureRandom.hex}")
+    exception = nil
+    thread = Thread.new do
+      scheduler = IOErrorScheduler.new
+      Fiber.set_scheduler scheduler
+      Fiber.schedule do
+        File.open(fn, 'w+') { it << 'foo' }
+      rescue => e
+        exception = e
+      end
+    end
+    thread.join
+    assert_kind_of Errno::EINVAL, exception
+  ensure
+    thread.kill rescue nil
+    FileUtils.rm_f(fn)
+  end
 end