Add comprehensive timeout support to queue implementations. (#420)

- Add `closed?` method to `PriorityQueue` for full queue interface compatibility.

Author: samuel-williams-shopify

fixtures/async/a_queue_with_timeout.rb

@@ -0,0 +1,119 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "sus/fixtures/async"
+
+module Async
+	AQueueWithTimeout = Sus::Shared("a queue with timeout support") do
+		include Sus::Fixtures::Async::ReactorContext
+		
+		let(:queue) {subject.new}
+		
+		with "timeout support" do
+			it "supports timeout: 0 for non-blocking dequeue" do
+				# Empty queue should return nil immediately
+				result = queue.dequeue(timeout: 0)
+				expect(result).to be_nil
+				
+				# With item, should return immediately
+				queue.push("item")
+				result = queue.dequeue(timeout: 0)
+				expect(result).to be == "item"
+			end
+			
+			it "supports timeout: 0 for non-blocking pop" do
+				# Empty queue should return nil immediately
+				result = queue.pop(timeout: 0)
+				expect(result).to be_nil
+				
+				# With item, should return immediately
+				queue.push("item")
+				result = queue.pop(timeout: 0)
+				expect(result).to be == "item"
+			end
+			
+			it "supports positive timeout values" do
+				start_time = Time.now
+				
+				# Should timeout after specified time
+				result = queue.dequeue(timeout: 0.1)
+				elapsed = Time.now - start_time
+				
+				expect(result).to be_nil
+				expect(elapsed).to be >= 0.1
+				expect(elapsed).to be < 0.2  # Should not wait much longer
+			end
+			
+			it "returns item before timeout expires" do
+				result = nil
+				
+				# Start dequeue with timeout in background
+				task = reactor.async do
+					result = queue.dequeue(timeout: 1.0)
+				end
+				
+				# Add item quickly
+				reactor.sleep(0.05)
+				queue.push("quick_item")
+				
+				task.wait
+				expect(result).to be == "quick_item"
+			end
+			
+			it "handles concurrent timeouts" do
+				results = []
+				
+				# Start multiple consumers with different timeouts
+				task1 = reactor.async do
+					results << [:task1, queue.dequeue(timeout: 0.1)]
+				end
+				
+				task2 = reactor.async do
+					results << [:task2, queue.dequeue(timeout: 0.2)]
+				end
+				
+				task3 = reactor.async do
+					results << [:task3, queue.dequeue(timeout: 0.3)]
+				end
+				
+				# Wait for all to timeout
+				[task1, task2, task3].each(&:wait)
+				
+				# All should have timed out
+				expect(results).to be == [
+					[:task1, nil],
+					[:task2, nil],
+					[:task3, nil]
+				]
+			end
+			
+			it "preserves FIFO order when items arrive before timeout" do
+				results = []
+				
+				# Start multiple consumers with same timeout
+				tasks = 3.times.map do |i|
+					reactor.async do
+						results << [i, queue.dequeue(timeout: 1.0)]
+					end
+				end
+				
+				# Add items quickly
+				reactor.sleep(0.05)
+				queue.push("item1")
+				queue.push("item2")
+				queue.push("item3")
+				
+				tasks.each(&:wait)
+				
+				# Should maintain FIFO order
+				expect(results).to be == [
+					[0, "item1"],
+					[1, "item2"],
+					[2, "item3"]
+				]
+			end
+		end
+	end
+end

lib/async/priority_queue.rb

@@ -39,8 +39,8 @@ def signal(value)
 				condition.signal
 			end
 
-			def wait_for_value(mutex)
-				condition.wait(mutex)
+			def wait_for_value(mutex, timeout = nil)
+				condition.wait(mutex, timeout)
 				return self.value
 			end
 
@@ -55,6 +55,8 @@ def valid?
 			end
 		end
 
+		private_constant :Waiter
+		
 		# Create a new priority queue.
 		#
 		# @parameter parent [Interface(:async) | Nil] The parent task to use for async operations.
@@ -81,6 +83,11 @@ def close
 			end
 		end
 
+		# @returns [Boolean] Whether the queue is closed.
+		def closed?
+			@closed
+		end
+		
 		# @attribute [Array] The items in the queue.
 		attr :items
 
@@ -153,13 +160,14 @@ def enqueue(*items)
 
 		# Remove and return the next item from the queue.
 		#
-		# If the queue is empty, this method will block until an item is available.
+		# If the queue is empty, this method will block until an item is available or timeout expires.
 		# Fibers are served in priority order, with higher priority fibers receiving
 		# items first.
 		#
 		# @parameter priority [Numeric] The priority of this consumer (higher = served first).
-		# @returns [Object] The next item in the queue.
-		def dequeue(priority: 0)
+		# @parameter timeout [Numeric, nil] Maximum time to wait for an item. If nil, waits indefinitely. If 0, returns immediately.
+		# @returns [Object, nil] The next item in the queue, or nil if timeout expires.
+		def dequeue(priority: 0, timeout: nil)
 			@mutex.synchronize do
 				# If queue is closed and empty, return nil immediately:
 				if @closed && @items.empty?
@@ -174,6 +182,9 @@ def dequeue(priority: 0)
 					end
 				end
 
+				# Handle immediate timeout (non-blocking)
+				return nil if timeout == 0
+				
 				# Need to wait - create our own condition variable and add to waiting queue:
 				sequence = @sequence
 				@sequence += 1
@@ -185,7 +196,7 @@ def dequeue(priority: 0)
 					@waiting.push(waiter)
 
 					# Wait for our specific condition variable to be signaled:
-					return waiter.wait_for_value(@mutex)
+					return waiter.wait_for_value(@mutex, timeout)
 				ensure
 					waiter&.invalidate!
 				end
@@ -195,8 +206,10 @@ def dequeue(priority: 0)
 		# Compatibility with {::Queue#pop}.
 		#
 		# @parameter priority [Numeric] The priority of this consumer.
-		def pop(priority: 0)
-			self.dequeue(priority: priority)
+		# @parameter timeout [Numeric, nil] Maximum time to wait for an item. If nil, waits indefinitely. If 0, returns immediately.
+		# @returns [Object, nil] The dequeued item, or nil if timeout expires.
+		def pop(priority: 0, timeout: nil)
+			self.dequeue(priority: priority, timeout: timeout)
 		end
 
 		# Process each item in the queue.

lib/async/queue.rb

@@ -73,13 +73,17 @@ def enqueue(*items)
 		end
 
 		# Remove and return the next item from the queue.
-		def dequeue
-			@delegate.pop
+		# @parameter timeout [Numeric, nil] Maximum time to wait for an item. If nil, waits indefinitely. If 0, returns immediately.
+		# @returns [Object, nil] The dequeued item, or nil if timeout expires.
+		def dequeue(timeout: nil)
+			@delegate.pop(timeout: timeout)
 		end
 
 		# Compatibility with {::Queue#pop}.
-		def pop(...)
-			@delegate.pop(...)
+		# @parameter timeout [Numeric, nil] Maximum time to wait for an item. If nil, waits indefinitely. If 0, returns immediately.
+		# @returns [Object, nil] The dequeued item, or nil if timeout expires.
+		def pop(timeout: nil)
+			@delegate.pop(timeout: timeout)
 		end
 
 		# Process each item in the queue.

releases.md

@@ -1,5 +1,12 @@
 # Releases
 
+## Unreleased
+
+- Add timeout support to `Async::Queue#dequeue` and `Async::Queue#pop` methods.
+- Add timeout support to `Async::PriorityQueue#dequeue` and `Async::PriorityQueue#pop` methods.
+- Add `closed?` method to `Async::PriorityQueue` for full queue interface compatibility.
+- Support non-blocking operations using `timeout: 0` parameter.
+
 ## v2.29.0
 
 This release introduces thread-safety as a core concept of Async. Many core classes now have thread-safe guarantees, allowing them to be used safely across multiple threads.

test/async/priority_queue.rb

@@ -6,12 +6,17 @@
 
 require "async/priority_queue"
 require "sus/fixtures/async"
+require "async/a_queue"
+require "async/a_queue_with_timeout"
 
 describe Async::PriorityQueue do
 	include Sus::Fixtures::Async::ReactorContext
 
 	let(:queue) {subject.new}
 
+	it_behaves_like Async::AQueue
+	it_behaves_like Async::AQueueWithTimeout
+	
 	with "#push" do
 		it "can push and pop items" do
 			queue.push("item")
@@ -608,21 +613,4 @@
 			expect(priorities_served).to be == [2, 0]
 		end
 	end
-	
-	describe Async::PriorityQueue::Waiter do
-		it "should invalidate correctly" do
-			condition = ConditionVariable.new
-			fiber = Fiber.current
-			waiter = Async::PriorityQueue::Waiter.new(fiber, 1, 1, condition, nil)
-			
-			expect(waiter).to be(:valid?)
-			expect(waiter.fiber).to be == fiber
-			expect(waiter.condition).to be == condition
-			
-			waiter.invalidate!
-			
-			expect(waiter).not.to be(:valid?)
-			expect(waiter.fiber).to be_nil
-		end
-	end
 end

test/async/queue.rb

@@ -10,9 +10,11 @@
 
 require "sus/fixtures/async"
 require "async/a_queue"
+require "async/a_queue_with_timeout"
 
 describe Async::Queue do
 	include Sus::Fixtures::Async::ReactorContext
 
 	it_behaves_like Async::AQueue
+	it_behaves_like Async::AQueueWithTimeout
 end