Repair IO#wait_readable, IO#wait_writable, IO#wait to be interruptible

* Fixes #3504

Author: andrykonchin

CHANGELOG.md

@@ -11,6 +11,7 @@ Bug fixes:
 * Fix `rb_gc_register_mark_object()` for `Float` and bignum values (#3502, @eregon, @andrykonchin).
 * Fix parsing literal floats when the locale does not use `.` for the decimal separator (e.g. `LANG=fr_FR.UTF-8`) (#3512, @eregon).
 * Fix `IO#{read_nonblock,readpartial,sysread}`, `BasicSocket#{recv,recv_nonblock}`, `{Socket,UDPSocket}#recvfrom_nonblock`, `UnixSocket#recvfrom` and preserve a provided buffer's encoding (#3506, @andrykonchyn).
+* Repair `IO#{wait_readable,wait_writable,wait}` to be interruptible (#3504, @andrykonchin).
 
 Compatibility:
 

spec/ruby/library/io-wait/fixtures/classes.rb spec/ruby/fixtures/io.rb

@@ -1,12 +1,12 @@
-module IOWaitSpec
+module IOSpec
   def self.exhaust_write_buffer(io)
     written = 0
     buf = " " * 4096
 
-    begin
+    while true
       written += io.write_nonblock(buf)
-    rescue Errno::EAGAIN, Errno::EWOULDBLOCK
-      return written
-    end while true
+    end
+  rescue Errno::EAGAIN, Errno::EWOULDBLOCK
+    written
   end
 end

spec/ruby/library/io-wait/wait_readable_spec.rb

@@ -24,4 +24,23 @@
   it "waits for the IO to become readable with the given large timeout" do
     @io.wait_readable(365 * 24 * 60 * 60).should == @io
   end
+
+  it "can be interrupted" do
+    rd, _wr = IO.pipe
+    start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+    t = Thread.new do
+      rd.wait_readable(10)
+    end
+
+    Thread.pass until t.stop?
+    t.kill
+    t.join
+
+    finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    (finish - start).should < 9
+  ensure
+    rd.close
+    _wr.close
+  end
 end

spec/ruby/library/io-wait/wait_spec.rb

@@ -1,5 +1,5 @@
 require_relative '../../spec_helper'
-require_relative 'fixtures/classes'
+require_relative '../../fixtures/io'
 
 ruby_version_is ''...'3.2' do
   require 'io/wait'
@@ -55,7 +55,7 @@
     end
 
     it "waits for the WRITABLE event to be ready" do
-      written_bytes = IOWaitSpec.exhaust_write_buffer(@w)
+      written_bytes = IOSpec.exhaust_write_buffer(@w)
       @w.wait(IO::WRITABLE, 0).should == nil
 
       @r.read(written_bytes)
@@ -67,7 +67,7 @@
     end
 
     it "returns nil when the WRITABLE event is not ready during the timeout" do
-      IOWaitSpec.exhaust_write_buffer(@w)
+      IOSpec.exhaust_write_buffer(@w)
       @w.wait(IO::WRITABLE, 0).should == nil
     end
 
@@ -92,14 +92,45 @@
     end
 
     it "changes thread status to 'sleep' when waits for WRITABLE event" do
-      written_bytes = IOWaitSpec.exhaust_write_buffer(@w)
+      IOSpec.exhaust_write_buffer(@w)
 
       t = Thread.new { @w.wait(IO::WRITABLE, 10) }
       sleep 1
       t.status.should == 'sleep'
       t.kill
       t.join # Thread#kill doesn't wait for the thread to end
     end
+
+    it "can be interrupted when waiting for READABLE event" do
+      start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+      t = Thread.new do
+        @r.wait(IO::READABLE, 10)
+      end
+
+      Thread.pass until t.stop?
+      t.kill
+      t.join # Thread#kill doesn't wait for the thread to end
+
+      finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      (finish - start).should < 9
+    end
+
+    it "can be interrupted when waiting for WRITABLE event" do
+      IOSpec.exhaust_write_buffer(@w)
+      start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+      t = Thread.new do
+        @w.wait(IO::WRITABLE, 10)
+      end
+
+      Thread.pass until t.stop?
+      t.kill
+      t.join # Thread#kill doesn't wait for the thread to end
+
+      finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      (finish - start).should < 9
+    end
   end
 
   context "[timeout, mode] passed" do

spec/ruby/library/io-wait/wait_writable_spec.rb

@@ -1,4 +1,5 @@
 require_relative '../../spec_helper'
+require_relative '../../fixtures/io'
 
 ruby_version_is ''...'3.2' do
   require 'io/wait'
@@ -17,4 +18,24 @@
     # Represents one year and is larger than a 32-bit int
     STDOUT.wait_writable(365 * 24 * 60 * 60).should == STDOUT
   end
+
+  it "can be interrupted" do
+    _rd, wr = IO.pipe
+    IOSpec.exhaust_write_buffer(wr)
+    start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+    t = Thread.new do
+      wr.wait_writable(10)
+    end
+
+    Thread.pass until t.stop?
+    t.kill
+    t.join
+
+    finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    (finish - start).should < 9
+  ensure
+    _rd.close unless _rd.closed?
+    wr.close unless wr.closed?
+  end
 end

spec/ruby/optional/capi/io_spec.rb

@@ -1,4 +1,5 @@
 require_relative 'spec_helper'
+require_relative '../../fixtures/io'
 
 load_extension('io')
 
@@ -279,6 +280,22 @@
       it "raises an IOError if the IO is not initialized" do
         -> { @o.rb_io_maybe_wait_writable(0, IO.allocate, nil) }.should raise_error(IOError, "uninitialized stream")
       end
+
+      it "can be interrupted" do
+        IOSpec.exhaust_write_buffer(@w_io)
+        start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+        t = Thread.new do
+          @o.rb_io_maybe_wait_writable(0, @w_io, 10)
+        end
+
+        Thread.pass until t.stop?
+        t.kill
+        t.join
+
+        finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        (finish - start).should < 9
+      end
     end
   end
 
@@ -355,6 +372,21 @@
           thr.join
         end
 
+        it "can be interrupted" do
+          start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+          t = Thread.new do
+            @o.rb_io_maybe_wait_readable(0, @r_io, 10, false)
+          end
+
+          Thread.pass until t.stop?
+          t.kill
+          t.join
+
+          finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+          (finish - start).should < 9
+        end
+
         it "raises an IOError if the IO is closed" do
           @r_io.close
           -> { @o.rb_io_maybe_wait_readable(0, @r_io, nil, false) }.should raise_error(IOError, "closed stream")
@@ -438,6 +470,37 @@
       it "raises an IOError if the IO is not initialized" do
         -> { @o.rb_io_maybe_wait(0, IO.allocate, IO::WRITABLE, nil) }.should raise_error(IOError, "uninitialized stream")
       end
+
+      it "can be interrupted when waiting for READABLE event" do
+        start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+        t = Thread.new do
+          @o.rb_io_maybe_wait(0, @r_io, IO::READABLE, 10)
+        end
+
+        Thread.pass until t.stop?
+        t.kill
+        t.join
+
+        finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        (finish - start).should < 9
+      end
+
+      it "can be interrupted when waiting for WRITABLE event" do
+        IOSpec.exhaust_write_buffer(@w_io)
+        start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+        t = Thread.new do
+          @o.rb_io_maybe_wait(0, @w_io, IO::WRITABLE, 10)
+        end
+
+        Thread.pass until t.stop?
+        t.kill
+        t.join
+
+        finish = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        (finish - start).should < 9
+      end
     end
   end
 

src/main/ruby/truffleruby/core/posix.rb

@@ -242,6 +242,9 @@ def self.attach_function_eagerly(native_name, argument_types, return_type,
       # We should capture the non-lazy method
       attach_function_eagerly :poll, [:pointer, :nfds_t, :int], :int, LIBC, false, :poll, self
       POLL = method(:poll)
+
+      attach_function_eagerly :truffleposix_poll_single_fd, [:int, :int, :int], :int, LIBTRUFFLEPOSIX, false, :truffleposix_poll_single_fd, self
+      POLL_SINGLE_FD = method(:truffleposix_poll_single_fd)
     end
   end
 

src/main/ruby/truffleruby/core/truffle/io_operations.rb

@@ -265,7 +265,9 @@ def self.poll(io, event_mask, timeout)
         status = Primitive.thread_set_status(Thread.current, :sleep)
 
         begin
-          returned_events = Truffle::POSIX.truffleposix_poll_single_fd(Primitive.io_fd(io), event_mask, remaining_timeout)
+          returned_events = Primitive.thread_run_blocking_nfi_system_call(
+            Truffle::POSIX::POLL_SINGLE_FD,
+            [Primitive.io_fd(io), event_mask, remaining_timeout])
         ensure
           Primitive.thread_set_status(Thread.current, status)
         end