diff --git a/fixtures/async/a_condition.rb b/fixtures/async/a_condition.rb index 56fa3f4f..59201dcc 100644 --- a/fixtures/async/a_condition.rb +++ b/fixtures/async/a_condition.rb @@ -84,6 +84,25 @@ module Async expect(condition.waiting_count).to be == 0 end + it "can timeout while waiting" do + expect do + condition.wait(timeout: 0.01) + end.to raise_exception(Async::TimeoutError) + + expect(condition.waiting_count).to be == 0 + end + + it "can receive nil before the timeout" do + task = reactor.async do + condition.wait(timeout: 1) + end + + expect(condition.waiting_count).to be == 1 + condition.signal + + expect(task.wait).to be_nil + end + with "timeout" do let(:ready) {Async::Variable.new(condition)} let(:waiting) {Async::Variable.new(subject.new)} diff --git a/lib/async/condition.rb b/lib/async/condition.rb index 92c09a60..20aadf28 100644 --- a/lib/async/condition.rb +++ b/lib/async/condition.rb @@ -6,21 +6,31 @@ # Copyright, 2026, by Robert Mosolgo. require "fiber" +require_relative "error" require_relative "list" module Async # A synchronization primitive, which allows fibers to wait until a particular condition is (edge) triggered. # @public Since *Async v1*. class Condition + Entry = Struct.new(:value) + private_constant :Entry + # Create a new condition. def initialize @ready = ::Thread::Queue.new end - # Queue up the current fiber and wait on yielding the task. + # Queue up the current fiber and wait until the condition is signalled. + # @parameter timeout [Numeric | Nil] The maximum time to wait, or `nil` to wait indefinitely. # @returns [Object] - def wait - @ready.pop + # @raises [Async::TimeoutError] If the timeout expires before the condition is signalled. + def wait(timeout: nil) + if entry = @ready.pop(timeout: timeout) + return entry.value + else + raise TimeoutError, "Timeout while waiting for condition!" + end end # @returns [Boolean] If there are no fibers waiting on this condition. @@ -44,9 +54,10 @@ def signal(value = nil) return if empty? ready = self.exchange + entry = Entry.new(value) ready.num_waiting.times do - ready.push(value) + ready.push(entry) end ready.close diff --git a/lib/async/notification.rb b/lib/async/notification.rb index 5c4df367..f727dbc8 100644 --- a/lib/async/notification.rb +++ b/lib/async/notification.rb @@ -26,8 +26,10 @@ def alive? end def transfer + entry = Entry.new(value) + ready.num_waiting.times do - ready.push(value) + ready.push(entry) end ready.close diff --git a/releases.md b/releases.md index 0510ff3d..03c175f8 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Added `Async::Condition#wait(timeout:)`, inherited by `Async::Notification`, which raises `Async::TimeoutError` if the condition is not signalled before the timeout expires. + ## v2.45.1 - Fixed `Scheduler#io_wait` returning `nil` instead of `false` when an explicit timeout expired. Native callers such as `Socket#connect` with `connect_timeout:` distinguish a timeout by checking for `false`, so the `nil` caused `TypeError: no implicit conversion from nil to integer` instead of the intended `IO::TimeoutError`.