diff --git a/fixtures/io/event/test_scheduler.rb b/fixtures/io/event/test_scheduler.rb index 5dcc6b74..6c71315d 100644 --- a/fixtures/io/event/test_scheduler.rb +++ b/fixtures/io/event/test_scheduler.rb @@ -159,6 +159,13 @@ def io_wait(io, events, timeout = nil) timer&.cancel! end + def process_wait(pid, flags) + @blocked += 1 + @selector.process_wait(Fiber.current, pid, flags) + ensure + @blocked -= 1 + end + def kernel_sleep(duration = nil) if duration self.block(nil, duration) diff --git a/lib/io/event/interrupt.rb b/lib/io/event/interrupt.rb index b49985db..9b8bcbd8 100644 --- a/lib/io/event/interrupt.rb +++ b/lib/io/event/interrupt.rb @@ -14,6 +14,8 @@ def initialize(selector) @selector = selector @input, @output = ::IO.pipe + @output.sync = true + @fiber = Fiber.new do while true if @selector.io_wait(@fiber, @input, IO::READABLE) @@ -27,12 +29,10 @@ def initialize(selector) @fiber.transfer end - # Send a sigle byte interrupt. + # Send a single byte interrupt. def signal - @output.write(".") - @output.flush - rescue IOError - # Ignore. + # This must not block or enter blocking operations or raise an exception. + @output.write_nonblock(".", exception: false) end def close diff --git a/releases.md b/releases.md index ab758607..92d9139a 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Correctly implement `Interrupt#signal` so that it is robust enough to be called by `Scheduler#unblock`. + ## v1.16.3 - Handle `IOError` raised while shutting down the pure Ruby interrupt pipe, so `IO::Event::Interrupt#close` does not leak expected shutdown errors from the interrupt fiber. diff --git a/test/io/event/interrupt.rb b/test/io/event/interrupt.rb index 8c0e8f9d..97c1ebd8 100644 --- a/test/io/event/interrupt.rb +++ b/test/io/event/interrupt.rb @@ -3,7 +3,10 @@ # Released under the MIT License. # Copyright, 2026, by Samuel Williams. +require "io/event" require "io/event/interrupt" +require "io/event/test_scheduler" +require "io/nonblock" describe IO::Event.const_get(:Interrupt) do let(:loop) {Fiber.current} @@ -34,4 +37,35 @@ expect(fiber).not.to be(:alive?) end end + + with "test scheduler" do + it "can be used to wake up a fiber blocked in `Thread#join`" do + skip_unless_method_defined(:fork, Process.singleton_class) + + 10.times do + r, w = IO.pipe + + Thread.new do + selector = IO::Event::Selector::Select.new(Fiber.current) + scheduler = IO::Event::TestScheduler.new(selector: selector) + Fiber.set_scheduler(scheduler) + + Fiber.schedule do + selector.dump_state($stderr, label: "interrupt fork before fork") if ENV["IO_EVENT_DIAGNOSTICS"] + + pid = Process.fork do + # Child process: + w.write("hello") + end + + # Parent process: + w.close + expect(r.read).to be == "hello" + ensure + Process.waitpid(pid) if pid + end + end.join + end + end + end end