[GR-40333] Ruby 3.1 Add rb_io_maybe_wait_readable, rb_io_maybe_wait_writable and rb_io_maybe_wait C-functions

PullRequest: truffleruby/3707

Author: andrykonchin

CHANGELOG.md

@@ -99,6 +99,7 @@ Compatibility:
 * Deprecate `rb_gc_force_recycle` and make it a no-op function (#2733, @moste00).
 * Add `Refinement#import_methods` method and add deprecation warning for `Refinement#include` and `Refinement#prepend` (#2733, @horakivo).
 * Upgrading `UNICODE` version to 13.0.0 and `EMOJI` version to 13.1 (#2733, @horakivo).
+* Add `rb_io_maybe_wait_readable`, `rb_io_maybe_wait_writable` and `rb_io_maybe_wait` functions (#2733, @andrykonchin).
 
 Performance:
 

lib/cext/ABI_version.txt

@@ -1 +1 @@
-4
+5

lib/truffle/truffle/cext_structs.rb

@@ -365,11 +365,13 @@ def polyglot_has_members?
   end
 
   def polyglot_members(internal)
-    ['stdio_file', 'fd', 'mode', 'pathv', 'pid', 'lineno', 'tied_io_for_writing']
+    ['self', 'stdio_file', 'fd', 'mode', 'pathv', 'pid', 'lineno', 'tied_io_for_writing']
   end
 
   def polyglot_read_member(name)
     case name
+    when 'self'
+      Primitive.cext_wrap(@io)
     when 'fd'
       Primitive.io_fd(@io)
     when 'mode'
@@ -405,11 +407,11 @@ def polyglot_invoke_member(name, *args)
   end
 
   def polyglot_member_readable?(name)
-    name == 'fd' || name == 'mode' || name == 'pathv' || 'tied_io_for_writing'
+    name == 'self' || name == 'fd' || name == 'mode' || name == 'pathv' || name == 'tied_io_for_writing'
   end
 
   def polyglot_member_modifiable?(name)
-    name == 'mode' || name == 'pathv' || 'tied_io_for_writing'
+    name == 'mode' || name == 'pathv' || name == 'tied_io_for_writing'
   end
 
   def polyglot_member_removable?(name)

spec/ruby/optional/capi/ext/io_spec.c

@@ -162,6 +162,60 @@ VALUE io_spec_rb_io_wait_writable(VALUE self, VALUE io) {
   return ret ? Qtrue : Qfalse;
 }
 
+#ifdef RUBY_VERSION_IS_3_1
+VALUE io_spec_rb_io_maybe_wait_writable(VALUE self, VALUE error, VALUE io, VALUE timeout) {
+  int ret = rb_io_maybe_wait_writable(NUM2INT(error), io, timeout);
+  return INT2NUM(ret);
+}
+#endif
+
+#ifdef RUBY_VERSION_IS_3_1
+VALUE io_spec_rb_io_maybe_wait_readable(VALUE self, VALUE error, VALUE io, VALUE timeout, VALUE read_p) {
+  int fd = io_spec_get_fd(io);
+#ifndef SET_NON_BLOCKING_FAILS_ALWAYS
+  char buf[RB_IO_WAIT_READABLE_BUF];
+  int ret, saved_errno;
+#endif
+
+  if (set_non_blocking(fd) == -1)
+    rb_sys_fail("set_non_blocking failed");
+
+#ifndef SET_NON_BLOCKING_FAILS_ALWAYS
+  if(RTEST(read_p)) {
+    if (read(fd, buf, RB_IO_WAIT_READABLE_BUF) != -1) {
+      return Qnil;
+    }
+    saved_errno = errno;
+    rb_ivar_set(self, rb_intern("@write_data"), Qtrue);
+    errno = saved_errno;
+  }
+
+  // main part
+  ret = rb_io_maybe_wait_readable(NUM2INT(error), io, timeout);
+
+  if(RTEST(read_p)) {
+    ssize_t r = read(fd, buf, RB_IO_WAIT_READABLE_BUF);
+    if (r != RB_IO_WAIT_READABLE_BUF) {
+      perror("read");
+      return SSIZET2NUM(r);
+    }
+    rb_ivar_set(self, rb_intern("@read_data"),
+        rb_str_new(buf, RB_IO_WAIT_READABLE_BUF));
+  }
+
+  return INT2NUM(ret);
+#else
+  UNREACHABLE;
+#endif
+}
+#endif
+
+#ifdef RUBY_VERSION_IS_3_1
+VALUE io_spec_rb_io_maybe_wait(VALUE self, VALUE error, VALUE io, VALUE events, VALUE timeout) {
+  return rb_io_maybe_wait(NUM2INT(error), io, events, timeout);
+}
+#endif
+
 VALUE io_spec_rb_thread_wait_fd(VALUE self, VALUE io) {
   rb_thread_wait_fd(io_spec_get_fd(io));
   return Qnil;
@@ -294,6 +348,11 @@ void Init_io_spec(void) {
   rb_define_method(cls, "rb_io_taint_check", io_spec_rb_io_taint_check, 1);
   rb_define_method(cls, "rb_io_wait_readable", io_spec_rb_io_wait_readable, 2);
   rb_define_method(cls, "rb_io_wait_writable", io_spec_rb_io_wait_writable, 1);
+#ifdef RUBY_VERSION_IS_3_1
+  rb_define_method(cls, "rb_io_maybe_wait_writable", io_spec_rb_io_maybe_wait_writable, 3);
+  rb_define_method(cls, "rb_io_maybe_wait_readable", io_spec_rb_io_maybe_wait_readable, 4);
+  rb_define_method(cls, "rb_io_maybe_wait", io_spec_rb_io_maybe_wait, 4);
+#endif
   rb_define_method(cls, "rb_thread_wait_fd", io_spec_rb_thread_wait_fd, 1);
   rb_define_method(cls, "rb_thread_fd_writable", io_spec_rb_thread_fd_writable, 1);
   rb_define_method(cls, "rb_thread_fd_select_read", io_spec_rb_thread_fd_select_read, 1);

spec/ruby/optional/capi/io_spec.rb

@@ -256,6 +256,27 @@
     end
   end
 
+  ruby_version_is "3.1" do
+    describe "rb_io_maybe_wait_writable" do
+      it "returns mask for events if operation was interrupted" do
+        @o.rb_io_maybe_wait_writable(Errno::EINTR::Errno, @w_io, nil).should == IO::WRITABLE
+      end
+
+      it "returns 0 if there is no error condition" do
+        @o.rb_io_maybe_wait_writable(0, @w_io, nil).should == 0
+      end
+
+      it "raises an IOError if the IO is closed" do
+        @w_io.close
+        -> { @o.rb_io_maybe_wait_writable(0, @w_io, nil) }.should raise_error(IOError, "closed stream")
+      end
+
+      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
+    end
+  end
+
   describe "rb_thread_fd_writable" do
     it "waits til an fd is ready for writing" do
       @o.rb_thread_fd_writable(@w_io).should be_nil
@@ -305,6 +326,40 @@
         thr.join
       end
     end
+
+    ruby_version_is "3.1" do
+      describe "rb_io_maybe_wait_readable" do
+        it "returns mask for events if operation was interrupted" do
+          @o.rb_io_maybe_wait_readable(Errno::EINTR::Errno, @r_io, nil, false).should == IO::READABLE
+        end
+
+        it "returns 0 if there is no error condition" do
+          @o.rb_io_maybe_wait_readable(0, @r_io, nil, false).should == 0
+        end
+
+        it "blocks until the io is readable and returns events that actually occurred" do
+          @o.instance_variable_set :@write_data, false
+          thr = Thread.new do
+            Thread.pass until @o.instance_variable_get(:@write_data)
+            @w_io.write "rb_io_wait_readable"
+          end
+
+          @o.rb_io_maybe_wait_readable(Errno::EAGAIN::Errno, @r_io, IO::READABLE, true).should == IO::READABLE
+          @o.instance_variable_get(:@read_data).should == "rb_io_wait_re"
+
+          thr.join
+        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")
+        end
+
+        it "raises an IOError if the IO is not initialized" do
+          -> { @o.rb_io_maybe_wait_readable(0, IO.allocate, nil, false) }.should raise_error(IOError, "uninitialized stream")
+        end
+      end
+    end
   end
 
   describe "rb_thread_wait_fd" do
@@ -344,6 +399,42 @@
       @o.rb_wait_for_single_fd(@r_io, 1, 0, 0).should == 0
     end
   end
+
+  ruby_version_is "3.1" do
+    describe "rb_io_maybe_wait" do
+      it "waits til an fd is ready for reading" do
+        start = false
+        thr = Thread.new do
+          start = true
+          sleep 0.05
+          @w_io.write "rb_io_maybe_wait"
+        end
+
+        Thread.pass until start
+
+        @o.rb_io_maybe_wait(Errno::EAGAIN::Errno, @r_io, IO::READABLE, nil).should == IO::READABLE
+
+        thr.join
+      end
+
+      it "returns mask for events if operation was interrupted" do
+        @o.rb_io_maybe_wait(Errno::EINTR::Errno, @w_io, IO::WRITABLE, nil).should == IO::WRITABLE
+      end
+
+      it "returns false if there is no error condition" do
+        @o.rb_io_maybe_wait(0, @w_io, IO::WRITABLE, nil).should == false
+      end
+
+      it "raises an IOError if the IO is closed" do
+        @w_io.close
+        -> { @o.rb_io_maybe_wait(0, @w_io, IO::WRITABLE, nil) }.should raise_error(IOError, "closed stream")
+      end
+
+      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
+    end
+  end
 end
 
 describe "rb_fd_fix_cloexec" do

spec/tags/optional/capi/io_tags.txt

@@ -0,0 +1,3 @@
+fails:C-API IO function rb_io_maybe_wait_writable raises an IOError if the IO is not initialized
+fails:C-API IO function rb_io_maybe_wait_readable raises an IOError if the IO is not initialized
+fails:C-API IO function rb_io_maybe_wait raises an IOError if the IO is not initialized

src/main/c/cext/io.c

@@ -83,6 +83,98 @@ int rb_wait_for_single_fd(int fd, int events, struct timeval *tv) {
   return polyglot_as_i32(polyglot_invoke(RUBY_CEXT, "rb_wait_for_single_fd", fd, events, tv_sec, tv_usec));
 }
 
+// The only difference between this implementation and CRuby's one
+// is that a fiber scheduler isn't supported here
+VALUE rb_io_wait(VALUE io, VALUE events, VALUE timeout) {
+  rb_io_t * fptr = NULL;
+  RB_IO_POINTER(io, fptr);
+
+  struct timeval tv_storage;
+  struct timeval *tv = NULL;
+
+  if (timeout != Qnil) {
+    tv_storage = rb_time_interval(timeout);
+    tv = &tv_storage;
+  }
+
+  int ready = rb_wait_for_single_fd(fptr->fd, RB_NUM2INT(events), tv);
+
+  if (ready < 0) {
+    rb_sys_fail(0);
+  }
+
+  // Not sure if this is necessary:
+  rb_io_check_closed(fptr);
+
+  if (ready) {
+    return RB_INT2NUM(ready);
+  }
+  else {
+    return Qfalse;
+  }
+}
+
+VALUE rb_io_maybe_wait(int error, VALUE io, VALUE events, VALUE timeout) {
+  // fptr->fd can be set to -1 at any time by another thread when the GVL is
+  // released. Many code, e.g. `io_bufread` didn't check this correctly and
+  // instead relies on `read(-1) -> -1` which causes this code path. We then
+  // check here whether the IO was in fact closed. Probably it's better to
+  // check that `fptr->fd != -1` before using it in syscall.
+  rb_io_check_closed(RFILE(io)->fptr);
+
+  switch (error) {
+  // In old Linux, several special files under /proc and /sys don't handle
+  // select properly. Thus we need avoid to call if don't use O_NONBLOCK.
+  // Otherwise, we face nasty hang up. Sigh.
+  // e.g. http://git.kernel.org/?p=linux/kernel/git/torvalds/linux-2.6.git;a=commit;h=31b07093c44a7a442394d44423e21d783f5523b8
+  // http://git.kernel.org/?p=linux/kernel/git/torvalds/linux-2.6.git;a=commit;h=31b07093c44a7a442394d44423e21d783f5523b8
+  // In EINTR case, we only need to call RUBY_VM_CHECK_INTS_BLOCKING().
+  // Then rb_thread_check_ints() is enough.
+  case EINTR:
+#if defined(ERESTART)
+    case ERESTART:
+#endif
+      // We might have pending interrupts since the previous syscall was interrupted:
+      rb_thread_check_ints();
+
+      // The operation was interrupted, so retry it immediately:
+      return events;
+
+  case EAGAIN:
+#if EWOULDBLOCK != EAGAIN
+  case EWOULDBLOCK:
+#endif
+    // The operation would block, so wait for the specified events:
+    return rb_io_wait(io, events, timeout);
+
+  default:
+    // Non-specific error, no event is ready:
+    return Qfalse;
+  }
+}
+
+int rb_io_maybe_wait_readable(int error, VALUE io, VALUE timeout) {
+  VALUE result = rb_io_maybe_wait(error, io, RB_INT2NUM(RUBY_IO_READABLE), timeout);
+
+  if (RTEST(result)) {
+    return RB_NUM2INT(result);
+  }
+  else {
+    return 0;
+  }
+}
+
+int rb_io_maybe_wait_writable(int error, VALUE io, VALUE timeout) {
+  VALUE result = rb_io_maybe_wait(error, io, RB_INT2NUM(RUBY_IO_WRITABLE), timeout);
+
+  if (RTEST(result)) {
+    return RB_NUM2INT(result);
+  }
+  else {
+    return 0;
+  }
+}
+
 VALUE rb_io_addstr(VALUE io, VALUE str) {
   // use write instead of just #<<, it's closer to what MRI does
   // and avoids stack-overflow in zlib where #<< is defined with this method

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

@@ -75,6 +75,10 @@ class EINPROGRESSWaitWritable < Errno::EINPROGRESS
     end
   end
 
+  READABLE = Truffle::Config['platform.poll.POLLIN']
+  WRITABLE = Truffle::Config['platform.poll.POLLOUT']
+  PRIORITY = Truffle::Config['platform.poll.POLLPRI']
+
   # InternalBuffer provides a sliding window into a region of bytes.
   # The buffer is filled to the +used+ indicator, which is
   # always less than or equal to +total+. As bytes are taken