diff --git a/async.gemspec b/async.gemspec index 5b999667..37c64c03 100644 --- a/async.gemspec +++ b/async.gemspec @@ -29,5 +29,5 @@ Gem::Specification.new do |spec| spec.add_dependency "console", "~> 1.29" spec.add_dependency "fiber-annotation" - spec.add_dependency "io-event", "~> 1.11" + spec.add_dependency "io-event", "~> 1.21" end diff --git a/lib/async/scheduler.rb b/lib/async/scheduler.rb index 46610357..3bfd4bbc 100644 --- a/lib/async/scheduler.rb +++ b/lib/async/scheduler.rb @@ -309,10 +309,12 @@ def address_resolve(hostname) # @parameter timeout [Float | Nil] The maximum time to wait, or if nil, indefinitely. def io_wait(io, events, timeout = nil) fiber = Fiber.current + expired = false if timeout # If an explicit timeout is specified, we expect that the user will handle it themselves: timer = @timers.after(timeout) do + expired = true fiber.transfer end elsif timeout = io.timeout @@ -322,7 +324,13 @@ def io_wait(io, events, timeout = nil) end end - return @selector.io_wait(fiber, io, events) + # A selector wait may return a falsy result when the fiber is resumed without the requested IO becoming ready. For example, a deferred unblock from a previous blocking operation may arrive after the fiber has moved on to this wait. Retry these stale or spurious wake-ups without resetting the original timer. + until result = @selector.io_wait(fiber, io, events) + # If the original timer resumed the fiber, the falsy result represents the timeout rather than a spurious wake-up: + return nil if expired + end + + return result ensure timer&.cancel! end @@ -415,7 +423,14 @@ def fiber_interrupt(fiber, exception) # @returns [Process::Status] A process status instance. # @asynchronous May be non-blocking.. def process_wait(pid, flags) - return @selector.process_wait(Fiber.current, pid, flags) + fiber = Fiber.current + + # A native process wait may be interrupted before the child exits. io-event reports this as `false` and leaves the retry policy to the scheduler. Retry only `false`, since `nil` is a legitimate result for `Process::WNOHANG`. + while true + status = @selector.process_wait(fiber, pid, flags) + + return status unless status == false + end end # Wait for the specified IOs to become ready for the specified events. diff --git a/releases.md b/releases.md index 1ecb3dea..8e095419 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Fixed scheduler I/O and process waits returning prematurely after stale or interrupted wake-ups. I/O waits now preserve their original timeout, while blocking process waits retry and non-blocking `Process::WNOHANG` waits still return `nil`. + ## v2.44.0 - Fixed scheduler cleanup after forking while other fibers are blocked. diff --git a/test/async/scheduler.rb b/test/async/scheduler.rb index c8b9b80c..221d6964 100644 --- a/test/async/scheduler.rb +++ b/test/async/scheduler.rb @@ -268,7 +268,78 @@ end end + with "#process_wait" do + it "retries interrupted process waits" do + status = Object.new + results = [false, status] + selector = Object.new + + selector.define_singleton_method(:process_wait) do |fiber, pid, flags| + results.shift + end + + scheduler = Async::Scheduler.new(selector: selector) + + expect(scheduler.process_wait(123, 0)).to be_equal(status) + end + + it "returns nil from non-blocking waits" do + selector = Object.new + + selector.define_singleton_method(:process_wait) do |fiber, pid, flags| + nil + end + + scheduler = Async::Scheduler.new(selector: selector) + + expect(scheduler.process_wait(123, Process::WNOHANG)).to be_nil + end + + end + with "#block" do + it "ignores stale wake-ups from previous blocking operations" do + input, output = IO.pipe + duration = nil + result = nil + + Sync do |parent| + queue = Thread::Queue.new + + child = parent.async do |task| + begin + task.with_timeout(0.02) do + queue.pop + end + rescue Async::TimeoutError + # Expected - the item was pushed after the timeout already expired. + end + + # The deferred wake-up from `queue.push` must not spuriously interrupt a subsequent IO operation: + duration = Async::Clock.measure do + result = input.wait_readable(0.02) + end + end + + producer = parent.async do + sleep(0.01) + queue.push(:wakeup) + end + + # Prevent the event loop from running until both the producer's sleep and the child's timeout are overdue, so that the wake-up from `queue.push` is still pending when the timeout fires: + Fiber.blocking{sleep(0.03)} + + child.wait + producer.wait + end + + expect(result).to be_nil + expect(duration).to be >= 0.02 + ensure + input&.close + output&.close + end + it "can block and unblock the scheduler after closing" do scheduler = Async::Scheduler.new