Adding IO#timeout.

CHANGELOG.md

@@ -20,6 +20,7 @@ Compatibility:
 * Add `String#bytesplice` (#3039, @itarato).
 * Add `String#byteindex` and `String#byterindex` (#3039, @itarato).
 * Add implementations of `rb_proc_call_with_block`, `rb_proc_call_kw`, `rb_proc_call_with_block_kw` and `rb_funcall_with_block_kw` (#3068, @andrykonchin).
+* Adding `IO#timeout` and `IO#timeout=` (#3039, @itarato).
 
 Performance:
 

spec/ruby/core/io/timeout_spec.rb

@@ -0,0 +1,104 @@
+# -*- encoding: utf-8 -*-
+require_relative '../../spec_helper'
+
+describe "IO#timeout" do
+  before :each do
+    @fname = tmp("io_timeout.txt")
+    @file = File.open(@fname, "a+")
+
+    @rpipe, @wpipe = IO.pipe
+    # There is no strict long term standard for pipe limits (2**16 bytes currently). This is an attempt to set a safe
+    # enough size to test a full pipe.
+    @more_than_pipe_limit = 1 << 18
+  end
+
+  after :each do
+    @rpipe.close
+    @wpipe.close
+
+    @file.close
+    rm_r @fname
+  end
+
+  ruby_version_is "3.2" do
+    it "files have timeout attribute" do
+      @fname = tmp("io_timeout_attribute.txt")
+      touch(@fname)
+
+      @file.timeout.should == nil
+
+      @file.timeout = 1.23
+      @file.timeout.should == 1.23
+    end
+
+    it "IO instances have timeout attribute" do
+      @rpipe.timeout.should == nil
+      @wpipe.timeout.should == nil
+
+      @rpipe.timeout = 1.23
+      @wpipe.timeout = 4.56
+
+      @rpipe.timeout.should == 1.23
+      @wpipe.timeout.should == 4.56
+    end
+
+    it "raises IO::TimeoutError when timeout is exceeded for .read" do
+      @rpipe.timeout = 0.01
+      -> { @rpipe.read.should }.should raise_error(IO::TimeoutError)
+    end
+
+    it "raises IO::TimeoutError when timeout is exceeded for .read(n)" do
+      @rpipe.timeout = 0.01
+      -> { @rpipe.read(3) }.should raise_error(IO::TimeoutError)
+    end
+
+    it "raises IO::TimeoutError when timeout is exceeded for .gets" do
+      @rpipe.timeout = 0.01
+      -> { @rpipe.gets }.should raise_error(IO::TimeoutError)
+    end
+
+    it "raises IO::TimeoutError when timeout is exceeded for .write" do
+      @wpipe.timeout = 0.01
+      -> { @wpipe.write("x" * @more_than_pipe_limit) }.should raise_error(IO::TimeoutError)
+    end
+
+    it "raises IO::TimeoutError when timeout is exceeded for .puts" do
+      @wpipe.timeout = 0.01
+      -> { @wpipe.puts("x" * @more_than_pipe_limit) }.should raise_error(IO::TimeoutError)
+    end
+
+    it "times out with .read when there is no EOF" do
+      @wpipe.write("hello")
+      @rpipe.timeout = 0.01
+
+      -> { @rpipe.read }.should raise_error(IO::TimeoutError)
+    end
+
+    it "returns content with .read when there is EOF" do
+      @wpipe.write("hello")
+      @wpipe.close
+
+      @rpipe.timeout = 0.01
+
+      @rpipe.read.should == "hello"
+    end
+
+    it "times out with .read(N) when there is not enough bytes" do
+      @wpipe.write("hello")
+      @rpipe.timeout = 0.01
+
+      @rpipe.read(2).should == "he"
+      -> { @rpipe.read(5) }.should raise_error(IO::TimeoutError)
+    end
+
+    it "returns partial content with .read(N) when there is not enough bytes but there is EOF" do
+      @wpipe.write("hello")
+      @rpipe.timeout = 0.01
+
+      @rpipe.read(2).should == "he"
+
+      @wpipe.close
+      @rpipe.read(5).should == "llo"
+    end
+  end
+end

spec/tags/truffle/methods_tags.txt

@@ -114,3 +114,5 @@ fails:Public methods on UnboundMethod should include private?
 fails:Public methods on UnboundMethod should include protected?
 fails:Public methods on UnboundMethod should include public?
 fails:Public methods on String should not include bytesplice
+fails:Public methods on IO should not include timeout
+fails:Public methods on IO should not include timeout=

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

@@ -38,6 +38,8 @@ class IO
 
   include Enumerable
 
+  class TimeoutError < IOError; end
+
   module WaitReadable; end
   module WaitWritable; end
 
@@ -1648,6 +1650,18 @@ def printf(fmt, *args)
     write sprintf(fmt, *args)
   end
 
+  def timeout
+    @timeout || nil
+  end
+
+  def timeout=(new_timeout)
+    if Primitive.nil?(timeout) ^ Primitive.nil?(new_timeout)
+      self.nonblock = !Primitive.nil?(new_timeout)
+    end
+
+    @timeout = new_timeout
+  end
+
   def read(length = nil, buffer = nil)
     ensure_open_and_readable
     buffer = StringValue(buffer) if buffer

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

@@ -401,10 +401,10 @@ def self.read_string_nonblock(io, count, exception)
   # by IO#sysread
 
   def self.read_string_native(io, length)
-    fd = io.fileno
     buffer = Primitive.io_thread_buffer_allocate(length)
     begin
-      bytes_read = Truffle::POSIX.read(fd, buffer, length)
+      bytes_read = execute_posix_read(io, buffer, length)
+
       if bytes_read < 0
         bytes_read, errno = bytes_read, Errno.errno
       elsif bytes_read == 0 # EOF
@@ -425,11 +425,62 @@ def self.read_string_native(io, length)
     end
   end
 
-  def self.read_to_buffer_native(io, length)
+  def self.execute_posix_read(io, buffer, length)
+    fd = io.fileno
+    return Truffle::POSIX.read(fd, buffer, length) unless io.timeout
+
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + io.timeout
+
+    loop do
+      current_timeout = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      raise IO::TimeoutError if current_timeout < 0
+
+      poll_result = Truffle::IOOperations.poll(io, Truffle::IOOperations::POLLIN, current_timeout)
+      if poll_result == 0
+        raise IO::TimeoutError
+      elsif poll_result == -1
+        Errno.handle_errno(Errno.errno)
+      end
+
+      if (bytes_read = Truffle::POSIX.read(fd, buffer, length)) == -1
+        continue if Errno.errno == Errno::EAGAIN
+        break
+      end
+
+      return bytes_read
+    end
+  end
+
+  def self.execute_posix_write(io, buffer, length)
     fd = io.fileno
+    return Truffle::POSIX.write(fd, buffer, length) unless io.timeout
+
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + io.timeout
+
+    loop do
+      current_timeout = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      raise IO::TimeoutError if current_timeout < 0
+
+      poll_result = Truffle::IOOperations.poll(io, Truffle::IOOperations::POLLOUT, current_timeout)
+      if poll_result == 0
+        raise IO::TimeoutError
+      elsif poll_result == -1
+        Errno.handle_errno(Errno.errno)
+      end
+
+      if (bytes_written = Truffle::POSIX.write(fd, buffer, length)) == -1
+        continue if Errno.errno == Errno::EAGAIN
+        break
+      end
+
+      return bytes_written
+    end
+  end
+
+  def self.read_to_buffer_native(io, length)
     buffer = Primitive.io_thread_buffer_allocate(length)
     begin
-      bytes_read = Truffle::POSIX.read(fd, buffer, length)
+      bytes_read = execute_posix_read(io, buffer, length)
       if bytes_read < 0
         bytes_read, errno = bytes_read, Errno.errno
       elsif bytes_read == 0 # EOF
@@ -495,7 +546,7 @@ def self.write_string_native(io, string, continue_on_eagain)
 
       written = 0
       while written < length
-        ret = Truffle::POSIX.write(fd, buffer + written, length - written)
+        ret = execute_posix_write(io, buffer + written, length - written)
         if ret < 0
           errno = Errno.errno
           if errno == EAGAIN_ERRNO
@@ -540,12 +591,11 @@ def self.write_string_polyglot(io, string, continue_on_eagain)
   # #write_string_nonblock_polylgot) is called by IO#write_nonblock
 
   def self.write_string_nonblock_native(io, string)
-    fd = io.fileno
     length = string.bytesize
     buffer = Primitive.io_thread_buffer_allocate(length)
     begin
       buffer.write_bytes string
-      written = Truffle::POSIX.write(fd, buffer, length)
+      written = execute_posix_write(io, buffer, length)
 
       if written < 0
         errno = Errno.errno