summaryrefslogtreecommitdiff
path: root/test/fiber
diff options
context:
space:
mode:
Diffstat (limited to 'test/fiber')
-rw-r--r--test/fiber/scheduler.rb138
-rw-r--r--test/fiber/test_scheduler.rb157
-rw-r--r--test/fiber/test_thread.rb6
3 files changed, 253 insertions, 48 deletions
diff --git a/test/fiber/scheduler.rb b/test/fiber/scheduler.rb
index 2401cb30d3..8f1ce4376b 100644
--- a/test/fiber/scheduler.rb
+++ b/test/fiber/scheduler.rb
@@ -65,69 +65,79 @@ class Scheduler
end
end
- def run
- # $stderr.puts [__method__, Fiber.current].inspect
-
+ def run_once
readable = writable = nil
- while @readable.any? or @writable.any? or @waiting.any? or @blocking.any?
- # May only handle file descriptors up to 1024...
- begin
- readable, writable = IO.select(@readable.keys + [@urgent.first], @writable.keys, [], next_timeout)
- rescue IOError
- # Ignore - this can happen if the IO is closed while we are waiting.
- end
+ begin
+ readable, writable = IO.select(@readable.keys + [@urgent.first], @writable.keys, [], next_timeout)
+ rescue IOError
+ # Ignore - this can happen if the IO is closed while we are waiting.
+ end
- # puts "readable: #{readable}" if readable&.any?
- # puts "writable: #{writable}" if writable&.any?
+ # puts "readable: #{readable}" if readable&.any?
+ # puts "writable: #{writable}" if writable&.any?
- selected = {}
+ selected = {}
- readable&.each do |io|
- if fiber = @readable.delete(io)
- @writable.delete(io) if @writable[io] == fiber
- selected[fiber] = IO::READABLE
- elsif io == @urgent.first
- @urgent.first.read_nonblock(1024)
- end
+ readable&.each do |io|
+ if fiber = @readable.delete(io)
+ @writable.delete(io) if @writable[io] == fiber
+ selected[fiber] = IO::READABLE
+ elsif io == @urgent.first
+ @urgent.first.read_nonblock(1024)
end
+ end
- writable&.each do |io|
- if fiber = @writable.delete(io)
- @readable.delete(io) if @readable[io] == fiber
- selected[fiber] = selected.fetch(fiber, 0) | IO::WRITABLE
- end
+ writable&.each do |io|
+ if fiber = @writable.delete(io)
+ @readable.delete(io) if @readable[io] == fiber
+ selected[fiber] = selected.fetch(fiber, 0) | IO::WRITABLE
end
+ end
- selected.each do |fiber, events|
- fiber.transfer(events)
- end
+ selected.each do |fiber, events|
+ fiber.transfer(events)
+ end
+
+ if @waiting.any?
+ time = current_time
+ waiting, @waiting = @waiting, {}
- if @waiting.any?
- time = current_time
- waiting, @waiting = @waiting, {}
-
- waiting.each do |fiber, timeout|
- if fiber.alive?
- if timeout <= time
- fiber.transfer
- else
- @waiting[fiber] = timeout
- end
+ waiting.each do |fiber, timeout|
+ if fiber.alive?
+ if timeout <= time
+ fiber.transfer
+ else
+ @waiting[fiber] = timeout
end
end
end
+ end
- if @ready.any?
- ready = nil
+ if @ready.any?
+ ready = nil
- @lock.synchronize do
- ready, @ready = @ready, []
- end
+ @lock.synchronize do
+ ready, @ready = @ready, []
+ end
- ready.each do |fiber|
- fiber.transfer if fiber.alive?
- end
+ ready.each do |fiber|
+ fiber.transfer if fiber.alive?
+ end
+ end
+ end
+
+ def run
+ # $stderr.puts [__method__, Fiber.current].inspect
+
+ # Use Thread.handle_interrupt like Async::Scheduler does
+ # This defers signal processing, which is the root cause of the gRPC bug
+ # See: https://github.com/socketry/async/blob/main/lib/async/scheduler.rb
+ Thread.handle_interrupt(::SignalException => :never) do
+ while @readable.any? or @writable.any? or @waiting.any? or @blocking.any?
+ run_once
+
+ break if Thread.pending_interrupt?
end
end
end
@@ -245,6 +255,13 @@ class Scheduler
end.value
end
+ # This hook is invoked by `IO#close`. Using a separate IO object
+ # demonstrates that the close operation is asynchronous.
+ def io_close(descriptor)
+ Fiber.blocking{IO.for_fd(descriptor.to_i).close}
+ return true
+ end
+
# This hook is invoked by `Kernel#sleep` and `Thread::Mutex#sleep`.
def kernel_sleep(duration = nil)
# $stderr.puts [__method__, duration, Fiber.current].inspect
@@ -471,6 +488,33 @@ class IOBufferScheduler < Scheduler
end
end
+class IOScheduler < Scheduler
+ def operations
+ @operations ||= []
+ end
+
+ def io_write(io, buffer, length, offset)
+ descriptor = io.fileno
+ string = buffer.get_string
+
+ self.operations << [:io_write, descriptor, string]
+
+ Fiber.blocking do
+ buffer.write(io, 0, offset)
+ end
+ end
+end
+
+class IOErrorScheduler < Scheduler
+ def io_read(io, buffer, length, offset)
+ return -Errno::EBADF::Errno
+ end
+
+ def io_write(io, buffer, length, offset)
+ return -Errno::EINVAL::Errno
+ end
+end
+
# This scheduler has a broken implementation of `unblock`` in the sense that it
# raises an exception. This is used to test the behavior of the scheduler when
# unblock raises an exception.
diff --git a/test/fiber/test_scheduler.rb b/test/fiber/test_scheduler.rb
index 7c77bd8cf0..d3696267f7 100644
--- a/test/fiber/test_scheduler.rb
+++ b/test/fiber/test_scheduler.rb
@@ -1,5 +1,7 @@
# frozen_string_literal: true
require 'test/unit'
+require 'securerandom'
+require 'fileutils'
require_relative 'scheduler'
class TestFiberScheduler < Test::Unit::TestCase
@@ -226,4 +228,159 @@ class TestFiberScheduler < Test::Unit::TestCase
thread.join
assert_kind_of RuntimeError, error
end
+
+ def test_post_fork_scheduler_reset
+ omit 'fork not supported' unless Process.respond_to?(:fork)
+
+ forked_scheduler_state = nil
+ thread = Thread.new do
+ r, w = IO.pipe
+ scheduler = Scheduler.new
+ Fiber.set_scheduler scheduler
+
+ forked_pid = fork do
+ r.close
+ w << (Fiber.scheduler ? 'set' : 'reset')
+ w.close
+ end
+ w.close
+ forked_scheduler_state = r.read
+ Process.wait(forked_pid)
+ ensure
+ r.close rescue nil
+ w.close rescue nil
+ end
+ thread.join
+ assert_equal 'reset', forked_scheduler_state
+ ensure
+ thread.kill rescue nil
+ end
+
+ def test_post_fork_fiber_blocking
+ omit 'fork not supported' unless Process.respond_to?(:fork)
+
+ fiber_blocking_state = nil
+ thread = Thread.new do
+ r, w = IO.pipe
+ scheduler = Scheduler.new
+ Fiber.set_scheduler scheduler
+
+ forked_pid = nil
+ Fiber.schedule do
+ forked_pid = fork do
+ r.close
+ w << (Fiber.current.blocking? ? 'blocking' : 'nonblocking')
+ w.close
+ end
+ end
+ w.close
+ fiber_blocking_state = r.read
+ Process.wait(forked_pid)
+ ensure
+ r.close rescue nil
+ w.close rescue nil
+ end
+ thread.join
+ assert_equal 'blocking', fiber_blocking_state
+ ensure
+ thread.kill rescue nil
+ end
+
+ def test_io_write_on_flush
+ begin
+ path = File.join(Dir.tmpdir, "ruby_test_io_write_on_flush_#{SecureRandom.hex}")
+ descriptor = nil
+ operations = nil
+
+ thread = Thread.new do
+ scheduler = IOScheduler.new
+ Fiber.set_scheduler scheduler
+
+ Fiber.schedule do
+ File.open(path, 'w+') do |file|
+ descriptor = file.fileno
+ file << 'foo'
+ file.flush
+ file << 'bar'
+ end
+ end
+
+ operations = scheduler.operations
+ end
+
+ thread.join
+ assert_equal [
+ [:io_write, descriptor, 'foo'],
+ [:io_write, descriptor, 'bar']
+ ], operations
+
+ assert_equal 'foobar', IO.read(path)
+ ensure
+ thread.kill rescue nil
+ FileUtils.rm_f(path)
+ end
+ end
+
+ def test_io_read_error
+ path = File.join(Dir.tmpdir, "ruby_test_io_read_error_#{SecureRandom.hex}")
+ error = nil
+
+ thread = Thread.new do
+ scheduler = IOErrorScheduler.new
+ Fiber.set_scheduler scheduler
+ Fiber.schedule do
+ File.open(path, 'w+') { it.read }
+ rescue => error
+ # Ignore.
+ end
+ end
+
+ thread.join
+ assert_kind_of Errno::EBADF, error
+ ensure
+ thread.kill rescue nil
+ FileUtils.rm_f(path)
+ end
+
+ def test_io_write_error
+ path = File.join(Dir.tmpdir, "ruby_test_io_write_error_#{SecureRandom.hex}")
+ error = nil
+
+ thread = Thread.new do
+ scheduler = IOErrorScheduler.new
+ Fiber.set_scheduler scheduler
+ Fiber.schedule do
+ File.open(path, 'w+') { it.sync = true; it << 'foo' }
+ rescue => error
+ # Ignore.
+ end
+ end
+
+ thread.join
+ assert_kind_of Errno::EINVAL, error
+ ensure
+ thread.kill rescue nil
+ FileUtils.rm_f(path)
+ end
+
+ def test_io_write_flush_error
+ path = File.join(Dir.tmpdir, "ruby_test_io_write_flush_error_#{SecureRandom.hex}")
+ error = nil
+
+ thread = Thread.new do
+ scheduler = IOErrorScheduler.new
+ Fiber.set_scheduler scheduler
+ Fiber.schedule do
+ File.open(path, 'w+') { it << 'foo' }
+ rescue => error
+ # Ignore.
+ end
+ end
+
+ thread.join
+ assert_kind_of Errno::EINVAL, error
+ ensure
+ thread.kill rescue nil
+ FileUtils.rm_f(path)
+ end
end
diff --git a/test/fiber/test_thread.rb b/test/fiber/test_thread.rb
index 0247f330d9..4d2fbde9ed 100644
--- a/test/fiber/test_thread.rb
+++ b/test/fiber/test_thread.rb
@@ -156,16 +156,20 @@ class TestFiberThread < Test::Unit::TestCase
end
def test_thread_join_hang
+ inner = nil
thread = Thread.new do
scheduler = SleepingUnblockScheduler.new
Fiber.set_scheduler scheduler
Fiber.schedule do
- Thread.new{sleep(0.01)}.value
+ inner = Thread.new{sleep(0.01)}
+ inner.value
end
end
thread.join
+ ensure
+ inner&.join
end
end
generated by cgit v1.2.3 (git 2.25.1) at 2026-06-07 04:02:03 +0000