Thread scheduler for light weight concurrency.

11 files changed, 511 insertions, 47 deletions

diff --git a/test/ruby/test_fiber.rb b/test/ruby/test_fiber.rb
index 7070fdf03c..4d103a7f76 100644
--- a/test/ruby/test_fiber.rb
+++ b/test/ruby/test_fiber.rb
@@ -347,51 +347,6 @@ class TestFiber < Test::Unit::TestCase
     EOS
   end
 
-  def invoke_rec script, vm_stack_size, machine_stack_size, use_length = true
-    env = {}
-    env['RUBY_FIBER_VM_STACK_SIZE'] = vm_stack_size.to_s if vm_stack_size
-    env['RUBY_FIBER_MACHINE_STACK_SIZE'] = machine_stack_size.to_s if machine_stack_size
-    out = Dir.mktmpdir("test_fiber") {|tmpdir|
-      out, err, status = EnvUtil.invoke_ruby([env, '-e', script], '', true, true, chdir: tmpdir, timeout: 30)
-      assert(!status.signaled?, FailDesc[status, nil, err])
-      out
-    }
-    use_length ? out.length : out
-  end
-
-  def test_stack_size
-    skip 'too unstable on riscv' if RUBY_PLATFORM =~ /riscv/
-    h_default = eval(invoke_rec('p RubyVM::DEFAULT_PARAMS', nil, nil, false))
-    h_0 = eval(invoke_rec('p RubyVM::DEFAULT_PARAMS', 0, 0, false))
-    h_large = eval(invoke_rec('p RubyVM::DEFAULT_PARAMS', 1024 * 1024 * 5, 1024 * 1024 * 10, false))
-
-    assert_operator(h_default[:fiber_vm_stack_size], :>, h_0[:fiber_vm_stack_size])
-    assert_operator(h_default[:fiber_vm_stack_size], :<, h_large[:fiber_vm_stack_size])
-    assert_operator(h_default[:fiber_machine_stack_size], :>=, h_0[:fiber_machine_stack_size])
-    assert_operator(h_default[:fiber_machine_stack_size], :<=, h_large[:fiber_machine_stack_size])
-
-    # check VM machine stack size
-    script = '$stdout.sync=true; def rec; print "."; rec; end; Fiber.new{rec}.resume'
-    size_default = invoke_rec script, nil, nil
-    assert_operator(size_default, :>, 0)
-    size_0 = invoke_rec script, 0, nil
-    assert_operator(size_default, :>, size_0)
-    size_large = invoke_rec script, 1024 * 1024 * 5, nil
-    assert_operator(size_default, :<, size_large)
-
-    return if /mswin|mingw/ =~ RUBY_PLATFORM
-
-    # check machine stack size
-    # Note that machine stack size may not change size (depend on OSs)
-    script = '$stdout.sync=true; def rec; print "."; 1.times{1.times{1.times{rec}}}; end; Fiber.new{rec}.resume'
-    vm_stack_size = 1024 * 1024
-    size_default = invoke_rec script, vm_stack_size, nil
-    size_0 = invoke_rec script, vm_stack_size, 0
-    assert_operator(size_default, :>=, size_0)
-    size_large = invoke_rec script, vm_stack_size, 1024 * 1024 * 10
-    assert_operator(size_default, :<=, size_large)
-  end
-
   def test_separate_lastmatch
     bug7678 = '[ruby-core:51331]'
     /a/ =~ "a"
diff --git a/test/ruby/test_stack.rb b/test/ruby/test_stack.rb
new file mode 100644
index 0000000000..6657b9e83c
--- /dev/null
+++ b/test/ruby/test_stack.rb
@@ -0,0 +1,81 @@
+# frozen_string_literal: false
+require 'test/unit'
+require 'tmpdir'
+
+class TestStack < Test::Unit::TestCase
+  LARGE_VM_STACK_SIZE = 1024*1024*5
+  LARGE_MACHINE_STACK_SIZE = 1024*1024*10
+
+  def initialize(*)
+    super
+
+    @h_default = nil
+    @h_0 = nil
+    @h_large = nil
+  end
+
+  def invoke_ruby script, vm_stack_size: nil, machine_stack_size: nil
+    env = {}
+    env['RUBY_FIBER_VM_STACK_SIZE'] = vm_stack_size.to_s if vm_stack_size
+    env['RUBY_FIBER_MACHINE_STACK_SIZE'] = machine_stack_size.to_s if machine_stack_size
+
+    stdout, stderr, status = EnvUtil.invoke_ruby([env, '-e', script], '', true, true, timeout: 30)
+    assert(!status.signaled?, FailDesc[status, nil, stderr])
+
+    return stdout
+  end
+
+  def h_default
+    @h_default ||= eval(invoke_ruby('p RubyVM::DEFAULT_PARAMS'))
+  end
+
+  def h_0
+    @h_0 ||= eval(invoke_ruby('p RubyVM::DEFAULT_PARAMS',
+      vm_stack_size: 0,
+      machine_stack_size: 0
+    ))
+  end
+
+  def h_large
+    @h_large ||= eval(invoke_ruby('p RubyVM::DEFAULT_PARAMS',
+      vm_stack_size: LARGE_VM_STACK_SIZE,
+      machine_stack_size: LARGE_MACHINE_STACK_SIZE
+    ))
+  end
+
+  def test_relative_stack_sizes
+    assert_operator(h_default[:fiber_vm_stack_size], :>, h_0[:fiber_vm_stack_size])
+    assert_operator(h_default[:fiber_vm_stack_size], :<, h_large[:fiber_vm_stack_size])
+    assert_operator(h_default[:fiber_machine_stack_size], :>=, h_0[:fiber_machine_stack_size])
+    assert_operator(h_default[:fiber_machine_stack_size], :<=, h_large[:fiber_machine_stack_size])
+  end
+
+  def test_vm_stack_size
+    script = '$stdout.sync=true; def rec; print "."; rec; end; Fiber.new{rec}.resume'
+    
+    size_default = invoke_ruby(script).bytesize
+    assert_operator(size_default, :>, 0)
+    
+    size_0 = invoke_ruby(script, vm_stack_size: 0).bytesize
+    assert_operator(size_default, :>, size_0)
+    
+    size_large = invoke_ruby(script, vm_stack_size: LARGE_VM_STACK_SIZE).bytesize
+    assert_operator(size_default, :<, size_large)
+  end
+
+  # Depending on OS, machine stack size may not change size.
+  def test_machine_stack_size
+    return if /mswin|mingw/ =~ RUBY_PLATFORM
+
+    script = '$stdout.sync=true; def rec; print "."; 1.times{1.times{1.times{rec}}}; end; Fiber.new{rec}.resume'
+
+    vm_stack_size = 1024 * 1024
+    size_default = invoke_ruby(script, vm_stack_size: vm_stack_size).bytesize
+
+    size_0 = invoke_ruby(script, vm_stack_size: vm_stack_size, machine_stack_size: 0).bytesize
+    assert_operator(size_default, :>=, size_0)
+
+    size_large = invoke_ruby(script, vm_stack_size: vm_stack_size, machine_stack_size: LARGE_MACHINE_STACK_SIZE).bytesize
+    assert_operator(size_default, :<=, size_large)
+  end
+end
diff --git a/test/scheduler/http.rb b/test/scheduler/http.rb
new file mode 100755
index 0000000000..e2a007bc84
--- /dev/null
+++ b/test/scheduler/http.rb
@@ -0,0 +1,53 @@
+
+require 'benchmark'
+
+TOPICS = ["cats", "dogs", "pigs", "skeletons", "zombies", "ocelots", "villagers", "pillagers"]
+
+require 'net/http'
+require 'uri'
+require 'json'
+
+require_relative 'scheduler'
+
+def fetch_topics(topics)
+  responses = {}
+
+  topics.each do |topic|
+    Fiber.new(blocking: Fiber.current.blocking?) do
+      uri = URI("https://www.google.com/search?q=#{topic}")
+      responses[topic] = Net::HTTP.get(uri).scan(topic).size
+    end.resume
+  end
+
+  Thread.scheduler&.run
+
+  return responses
+end
+
+def sweep(repeats: 3, **options)
+  times = (1..8).map do |i|
+    $stderr.puts "Measuring #{i} topic(s)..."
+    topics = TOPICS[0...i]
+
+    Thread.new do
+      Benchmark.realtime do
+        scheduler = Scheduler.new
+        Thread.current.scheduler = scheduler
+
+        repeats.times do
+          Fiber.new(**options) do
+            pp fetch_topics(topics)
+          end.resume
+
+          scheduler.run
+        end
+      end
+    end.value / repeats
+  end
+
+  puts options.inspect
+  puts JSON.dump(times.map{|value| value.round(3)})
+end
+
+sweep(blocking: true)
+sweep(blocking: false)
diff --git a/test/scheduler/scheduler.rb b/test/scheduler/scheduler.rb
new file mode 100644
index 0000000000..b2d36cc728
--- /dev/null
+++ b/test/scheduler/scheduler.rb
@@ -0,0 +1,163 @@
+# frozen_string_literal: true
+
+require 'fiber'
+
+begin
+  require 'io/nonblock'
+rescue LoadError
+  # Ignore.
+end
+
+class Scheduler
+  def initialize
+    @readable = {}
+    @writable = {}
+    @waiting = {}
+    @blocking = []
+
+    @ios = ObjectSpace::WeakMap.new
+  end
+
+  attr :fiber
+
+  attr :readable
+  attr :writable
+  attr :waiting
+  attr :blocking
+
+  def next_timeout
+    fiber, timeout = @waiting.min_by{|key, value| value}
+
+    if timeout
+      offset = timeout - current_time
+
+      if offset < 0
+        return 0
+      else
+        return offset
+      end
+    end
+  end
+
+  def run
+    while @readable.any? or @writable.any? or @waiting.any?
+      # Can only handle file descriptors up to 1024...
+      readable, writable = IO.select(@readable.keys, @writable.keys, [], next_timeout)
+
+      # puts "readable: #{readable}" if readable&.any?
+      # puts "writable: #{writable}" if writable&.any?
+
+      readable&.each do |io|
+        @readable[io]&.resume
+      end
+
+      writable&.each do |io|
+        @writable[io]&.resume
+      end
+
+      if @waiting.any?
+        time = current_time
+        waiting = @waiting
+        @waiting = {}
+
+        waiting.each do |fiber, timeout|
+          if timeout <= time
+            fiber.resume
+          else
+            @waiting[fiber] = timeout
+          end
+        end
+      end
+    end
+  end
+
+  def for_fd(fd)
+    @ios[fd] ||= ::IO.for_fd(fd, autoclose: false)
+  end
+
+  def wait_readable(io)
+    @readable[io] = Fiber.current
+
+    Fiber.yield
+
+    @readable.delete(io)
+
+    return true
+  end
+
+  def wait_readable_fd(fd)
+    wait_readable(
+      for_fd(fd)
+    )
+  end
+
+  def wait_writable(io)
+    @writable[io] = Fiber.current
+
+    Fiber.yield
+
+    @writable.delete(io)
+
+    return true
+  end
+
+  def wait_writable_fd(fd)
+    wait_writable(
+      for_fd(fd)
+    )
+  end
+
+  def current_time
+    Process.clock_gettime(Process::CLOCK_MONOTONIC)
+  end
+
+  def wait_sleep(duration = nil)
+    @waiting[Fiber.current] = current_time + duration
+
+    Fiber.yield
+
+    return true
+  end
+
+  def wait_any(io, events, duration)
+    unless (events & IO::WAIT_READABLE).zero?
+      @readable[io] = Fiber.current
+    end
+
+    unless (events & IO::WAIT_WRITABLE).zero?
+      @writable[io] = Fiber.current
+    end
+
+    Fiber.yield
+
+    @readable.delete(io)
+    @writable.delete(io)
+
+    return true
+  end
+
+  def wait_for_single_fd(fd, events, duration)
+    wait_any(
+      for_fd(fd),
+      events,
+      duration
+    )
+  end
+
+  def enter_blocking_region
+    # puts "Enter blocking region: #{caller.first}"
+  end
+
+  def exit_blocking_region
+    # puts "Exit blocking region: #{caller.first}"
+    @blocking << caller.first
+  end
+
+  def fiber(&block)
+    fiber = Fiber.new(blocking: false, &block)
+
+    fiber.resume
+
+    return fiber
+  end
+end
diff --git a/test/scheduler/test_enumerator.rb b/test/scheduler/test_enumerator.rb
new file mode 100644
index 0000000000..7c97382c52
--- /dev/null
+++ b/test/scheduler/test_enumerator.rb
@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+require 'test/unit'
+require 'socket'
+require_relative 'scheduler'
+
+class TestSchedulerEnumerator < Test::Unit::TestCase
+  MESSAGE = "Hello World"
+
+  def test_read_characters
+    skip unless defined?(UNIXSocket)
+
+    i, o = UNIXSocket.pair
+    skip unless i.nonblock? && o.nonblock?
+
+    message = String.new
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      e = i.to_enum(:each_char)
+
+      Fiber do
+        o.write("Hello World")
+        o.close
+      end
+
+      Fiber do
+        begin
+          while c = e.next
+            message << c
+          end
+        rescue StopIteration
+          # Ignore.
+        end
+
+        i.close
+      end
+    end
+
+    thread.join
+
+    assert_equal(MESSAGE, message)
+  end
+end
diff --git a/test/scheduler/test_fiber.rb b/test/scheduler/test_fiber.rb
new file mode 100644
index 0000000000..3452591cd9
--- /dev/null
+++ b/test/scheduler/test_fiber.rb
@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+require 'test/unit'
+require_relative 'scheduler'
+
+class TestSchedulerFiber < Test::Unit::TestCase
+  def test_fiber_without_scheduler
+    # Cannot create fiber without scheduler.
+    assert_raise RuntimeError do
+      Fiber do
+      end
+    end
+  end
+
+  def test_fiber_blocking
+    scheduler = Scheduler.new
+
+    thread = Thread.new do
+      Thread.current.scheduler = scheduler
+
+      # Close is always a blocking operation.
+      IO.pipe.each(&:close)
+    end
+
+    thread.join
+
+    assert_not_empty scheduler.blocking
+    assert_match /test_fiber.rb:\d+:in `close'/, scheduler.blocking.last
+  end
+end
diff --git a/test/scheduler/test_http.rb b/test/scheduler/test_http.rb
new file mode 100644
index 0000000000..82aa73ca35
--- /dev/null
+++ b/test/scheduler/test_http.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+require 'net/http'
+require 'uri'
+require 'openssl'
+
+require 'test/unit'
+require_relative 'scheduler'
+
+class TestSchedulerHTTP < Test::Unit::TestCase
+  def test_get
+    Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      Fiber do
+        uri = URI("https://www.ruby-lang.org/en/")
+
+        http = Net::HTTP.new uri.host, uri.port
+        http.use_ssl = true
+        http.verify_mode = OpenSSL::SSL::VERIFY_NONE
+        body = http.get(uri.path).body
+
+        assert !body.empty?
+      end
+    end.join
+  end
+end
diff --git a/test/scheduler/test_io.rb b/test/scheduler/test_io.rb
new file mode 100644
index 0000000000..ef46d1ac2c
--- /dev/null
+++ b/test/scheduler/test_io.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+require 'test/unit'
+require_relative 'scheduler'
+
+class TestSchedulerIO < Test::Unit::TestCase
+  MESSAGE = "Hello World"
+
+  def test_read
+    skip unless defined?(UNIXSocket)
+
+    i, o = UNIXSocket.pair
+    skip unless i.nonblock? && o.nonblock?
+
+    message = nil
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      Fiber do
+        message = i.read(20)
+        i.close
+      end
+
+      Fiber do
+        o.write("Hello World")
+        o.close
+      end
+    end
+
+    thread.join
+
+    assert_equal MESSAGE, message
+  end
+end
diff --git a/test/scheduler/test_mutex.rb b/test/scheduler/test_mutex.rb
new file mode 100644
index 0000000000..8395e5522f
--- /dev/null
+++ b/test/scheduler/test_mutex.rb
@@ -0,0 +1,47 @@
+# frozen_string_literal: true
+require 'test/unit'
+require_relative 'scheduler'
+
+class TestSchedulerMutex < Test::Unit::TestCase
+  def test_mutex_synchronize
+    mutex = Mutex.new
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      Fiber do
+        assert_equal Thread.scheduler, scheduler
+
+        mutex.synchronize do
+          assert_nil Thread.scheduler
+        end
+      end
+    end
+
+    thread.join
+  end
+
+  def test_mutex_deadlock
+    mutex = Mutex.new
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      Fiber do
+        assert_equal Thread.scheduler, scheduler
+
+        mutex.synchronize do
+          Fiber.yield
+        end
+      end
+      
+      assert_raise ThreadError do
+        mutex.lock
+      end
+    end
+
+    thread.join
+  end
+end
diff --git a/test/scheduler/test_sleep.rb b/test/scheduler/test_sleep.rb
new file mode 100644
index 0000000000..0be760341e
--- /dev/null
+++ b/test/scheduler/test_sleep.rb
@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+require 'test/unit'
+require_relative 'scheduler'
+
+class TestSchedulerSleep < Test::Unit::TestCase
+  ITEMS = [0, 1, 2, 3, 4]
+
+  def test_sleep
+    items = []
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      5.times do |i|
+        Fiber do
+          sleep(i/100.0)
+          items << i
+        end
+      end
+
+      # Should be 5 fibers waiting:
+      assert_equal scheduler.waiting.size, 5
+    end
+
+    thread.join
+
+    assert_equal ITEMS, items
+  end
+end
diff --git a/test/socket/test_basicsocket.rb b/test/socket/test_basicsocket.rb
index c8e9b23f83..7b1c9b4a06 100644
--- a/test/socket/test_basicsocket.rb
+++ b/test/socket/test_basicsocket.rb
@@ -159,8 +159,6 @@ class TestSocket_BasicSocket < Test::Unit::TestCase
       set_nb = true
       buf = String.new
       if ssock.respond_to?(:nonblock?)
-        assert_not_predicate(ssock, :nonblock?)
-        assert_not_predicate(csock, :nonblock?)
         csock.nonblock = ssock.nonblock = false
 
         # Linux may use MSG_DONTWAIT to avoid setting O_NONBLOCK