Wait for blocking SSLSocket operations through the fiber scheduler

src/main/java/org/jruby/ext/openssl/SSLSocket.java

@@ -550,9 +550,16 @@ private Object waitSelect(final int operations, final boolean blocking, final bo
         if ( ! channel.isSelectable() ) return Boolean.TRUE;
 
         final Ruby runtime = getRuntime();
-        final RubyThread thread = runtime.getCurrentContext().getThread();
+        final ThreadContext context = runtime.getCurrentContext();
+        final RubyThread thread = context.getThread();
 
         channel.configureBlocking(false);
+
+        if (blocking) {
+            IRubyObject scheduler = current_scheduler(context);
+            if (!scheduler.isNil()) return scheduler_io_wait(context, scheduler, operations);
+        }
+
         final Selector selector = runtime.getSelectorPool().get();
         SelectionKey key = null;
 
@@ -650,6 +657,25 @@ public void wakeup() {
         }
     }
 
+    // Fiber.current_scheduler if Fiber.respond_to?(:current_scheduler)
+    // (defined by default since JRuby 10; on 9.4.5+ only with -Xexperimental.fiber.scheduler)
+    private static IRubyObject current_scheduler(final ThreadContext context) {
+        final IRubyObject fiber = context.runtime.getObject().getConstantAt("Fiber");
+        if (fiber.respondsTo("current_scheduler")) return fiber.callMethod(context, "current_scheduler");
+        return context.nil;
+    }
+
+    // MRI: blocking SSL operations wait in rb_io_wait, which defers to the fiber scheduler
+    private boolean scheduler_io_wait(final ThreadContext context, IRubyObject scheduler, int operations) {
+        // scheduler.io_wait(@io, events, nil)
+        int events = 0;
+        if ((operations & SelectionKey.OP_READ) != 0) events |= 1; // IO::READABLE
+        if ((operations & SelectionKey.OP_WRITE) != 0) events |= 4; // IO::WRITABLE
+
+        IRubyObject[] args = { io, context.runtime.newFixnum(events), context.nil };
+        return scheduler.callMethod(context, "io_wait", args).isTrue();
+    }
+
     // return values are -1 (EOF) and >= 0 (byte counts), so any value < -1 is safe to use
     private static final int READ_WOULD_BLOCK_RESULT  = -2;
     private static final int WRITE_WOULD_BLOCK_RESULT = -3;

test/ssl/test_ssl.rb

@@ -1313,4 +1313,73 @@ def test_sysread_syswrite_raise_before_handshake
     end
   end
 
+  # Just enough of a Fiber scheduler to record io_wait calls and wait with IO.select
+  class RecordingScheduler
+    attr_reader :waits
+
+    def initialize
+      @waits = []
+      @readable = {}
+      @writable = {}
+    end
+
+    def io_wait(io, events, timeout)
+      @waits << events
+      @readable[io] = Fiber.current if events & IO::READABLE != 0
+      @writable[io] = Fiber.current if events & IO::WRITABLE != 0
+      Fiber.yield
+      events
+    ensure
+      @readable.delete(io)
+      @writable.delete(io)
+    end
+
+    def fiber(&block)
+      Fiber.new(blocking: false, &block).tap(&:resume)
+    end
+
+    def close
+      until @readable.empty? && @writable.empty?
+        readable, writable = IO.select(@readable.keys, @writable.keys)
+        (readable.map(&@readable) | writable.map(&@writable)).each(&:resume)
+      end
+    end
+
+    def kernel_sleep(duration = nil) raise NotImplementedError end
+    def block(blocker, timeout = nil) raise NotImplementedError end
+    def unblock(blocker, fiber) raise NotImplementedError end
+  end
+
+  def test_blocking_operations_wait_through_fiber_scheduler
+    assert_waits_through_fiber_scheduler(OpenSSL::SSL::SSLSocket)
+  end
+
+  def test_blocking_operations_wait_through_fiber_scheduler_in_subclass
+    assert_waits_through_fiber_scheduler(Class.new(OpenSSL::SSL::SSLSocket))
+  end
+
+  def assert_waits_through_fiber_scheduler(socket_class)
+    omit "no Fiber scheduler" unless Fiber.respond_to?(:set_scheduler)
+
+    start_server0(PORT, OpenSSL::SSL::VERIFY_NONE, true) do |server, port|
+      scheduler = RecordingScheduler.new
+      line = nil
+      Thread.new do
+        sock = TCPSocket.new("127.0.0.1", port)
+        Fiber.set_scheduler(scheduler)
+        Fiber.schedule do
+          ssl = socket_class.new(sock)
+          ssl.connect
+          ssl.puts "hello"
+          line = ssl.gets
+          ssl.close
+        end
+        Fiber.set_scheduler(nil) # closes the scheduler, running it until the fiber is done
+      end.join
+
+      assert_equal "hello\n", line
+      assert_not_empty scheduler.waits
+    end
+  end
+
 end