Retry interrupted waits in io_wait and process_wait

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

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.

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.

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