Barrier waits in order of completion/failure.

benchmark/async/barrier.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+# Copyright, 2025, by Samuel Williams.
+
+require "async/barrier"
+
+require "sus/fixtures/async/scheduler_context"
+require "sus/fixtures/benchmark"
+
+describe Async::Barrier do
+	include Sus::Fixtures::Async::SchedulerContext
+	include Sus::Fixtures::Benchmark
+
+	measure "can schedule several tasks quickly" do |repeats|
+		barrier = Async::Barrier.new
+
+		repeats.times do |i|
+			barrier.async{}
+		end
+
+		barrier.wait
+	end
+end

fixtures/async/a_queue.rb

@@ -20,6 +20,18 @@ module Async
 				expect(queue.size).to be == 1
 				expect(queue.dequeue).to be == :item
 			end
+
+			it "can't add an item to a closed queue" do
+				queue.push(:item)
+				expect(queue).to have_attributes(size: be == 1)
+				expect(queue.dequeue).to be == :item
+
+				queue.close
+
+				expect do
+					queue.push(:item)
+				end.to raise_exception(Async::Queue::ClosedError)
+			end
 		end
 
 		with "#pop" do
@@ -118,6 +130,19 @@ module Async
 				expect(queue.wait).to be == :item
 			end
 		end
+
+		with "#close" do
+			it "signals waiting tasks when closed" do
+				waiting_task = reactor.async do
+					queue.dequeue
+				end
+
+				queue.close
+
+				waiting_task.wait
+				expect(waiting_task).to be(:finished?)
+			end
+		end
 
 		with "an empty queue" do
 			it "is expected to be empty" do
@@ -163,6 +188,42 @@ module Async
 				expect(count).to be == repeats
 			end
 		end
+
+		with "a closed queue" do
+			before do
+				queue.close
+			end
+
+			it "prevents push after close" do
+				expect{queue.push(:item)}.to raise_exception(Async::Queue::ClosedError)
+			end
+
+			it "prevents enqueue after close" do
+				expect{queue.enqueue(:item)}.to raise_exception(Async::Queue::ClosedError)
+			end
+
+			it "prevents << after close" do
+				expect{queue << :item}.to raise_exception(Async::Queue::ClosedError)
+			end
+
+			it "returns nil from dequeue when closed and empty" do
+				expect(queue.dequeue).to be_nil
+			end
+
+			it "returns nil from pop when closed and empty" do
+				expect(queue.pop).to be_nil
+			end
+
+			it "signals waiting tasks when closed" do
+				waiting_task = reactor.async do
+					queue.dequeue
+				end
+
+				# Already closed, so just check if the task finishes:
+				waiting_task.wait
+				expect(waiting_task).to be(:finished?)
+			end
+		end
 
 		it_behaves_like Async::ChainableAsync do
 			def before

gems.rb

@@ -33,6 +33,7 @@
 	gem "sus-fixtures-async"
 	gem "sus-fixtures-console"
 	gem "sus-fixtures-time"
+	gem "sus-fixtures-benchmark"
 
 	gem "bake-test"
 	gem "bake-test-external"

lib/async/barrier.rb

@@ -5,6 +5,7 @@
 
 require_relative "list"
 require_relative "task"
+require_relative "queue"
 
 module Async
 	# A general purpose synchronisation primitive, which allows one task to wait for a number of other tasks to complete. It can be used in conjunction with {Semaphore}.
@@ -16,6 +17,7 @@ class Barrier
 		# @public Since *Async v1*.
 		def initialize(parent: nil)
 			@tasks = List.new
+			@finished = Queue.new
 
 			@parent = parent
 		end
@@ -41,11 +43,15 @@ def size
 		# Execute a child task and add it to the barrier.
 		# @asynchronous Executes the given block concurrently.
 		def async(*arguments, parent: (@parent or Task.current), **options, &block)
-			task = parent.async(*arguments, **options, &block)
+			waiting = nil
 
-			@tasks.append(TaskNode.new(task))
-
-			return task
+			parent.async(*arguments, **options) do |task, *arguments|
+				waiting = TaskNode.new(task)
+				@tasks.append(waiting)
+				block.call(task, *arguments)
+			ensure
+				@finished.signal(waiting)
+			end
 		end
 
 		# Whether there are any tasks being held by the barrier.
@@ -55,14 +61,27 @@ def empty?
 		end
 
 		# Wait for all tasks to complete by invoking {Task#wait} on each waiting task, which may raise an error. As long as the task has completed, it will be removed from the barrier.
+		#
+		# @yields {|task| ...} If a block is given, the unwaited task is yielded. You must invoke {Task#wait} yourself. In addition, you may `break` if you have captured enough results.
+		#
 		# @asynchronous Will wait for tasks to finish executing.
 		def wait
-			@tasks.each do |waiting|
+			while [email protected]?
+				# Wait for a task to finish (we get the task node):
+				return unless waiting = @finished.wait
+
+				# Remove the task as it is now finishing:
+				@tasks.remove?(waiting)
+
+				# Get the task:
 				task = waiting.task
-				begin
+
+				# If a block is given, the user can implement their own behaviour:
+				if block_given?
+					yield task
+				else
+					# Wait for it to either complete or raise an error:
 					task.wait
-				ensure
-					@tasks.remove?(waiting) unless task.alive?
 				end
 			end
 		end
@@ -73,6 +92,8 @@ def stop
 			@tasks.each do |waiting|
 				waiting.task.stop
 			end
+
+			@finished.close
 		end
 	end
 end

lib/async/list.rb

@@ -134,7 +134,7 @@ def remove(node)
 			return removed(node)
 		end
 
-		# @returns [Boolean] Returns true if the list is empty.
+		# @returns [Boolean] True if the list is empty.
 		def empty?
 			@size == 0
 		end

lib/async/notification.rb

@@ -10,12 +10,14 @@ module Async
 	# @public Since *Async v1*.
 	class Notification < Condition
 		# Signal to a given task that it should resume operations.
+		#
+		# @returns [Boolean] if a task was signalled.
 		def signal(value = nil, task: Task.current)
-			return if @waiting.empty?
+			return false if @waiting.empty?
 
 			Fiber.scheduler.push Signal.new(self.exchange, value)
 
-			return nil
+			return true
 		end
 
 		Signal = Struct.new(:waiting, :value) do

lib/async/queue.rb

@@ -15,16 +15,31 @@ module Async
 	#
 	# @public Since *Async v1*.
 	class Queue
+		# An error raised when trying to enqueue items to a closed queue.
+		# @public Since *Async v2.24*.
+		class ClosedError < RuntimeError
+		end
+
 		# Create a new queue.
 		#
 		# @parameter parent [Interface(:async) | Nil] The parent task to use for async operations.
 		# @parameter available [Notification] The notification to use for signaling when items are available.
 		def initialize(parent: nil, available: Notification.new)
 			@items = []
+			@closed = false
 			@parent = parent
 			@available = available
 		end
 
+		# Close the queue, causing all waiting tasks to return `nil`. Any subsequent calls to {enqueue} will raise an exception.
+		def close
+			@closed = true
+
+			while @available.waiting?
+				@available.signal(nil)
+			end
+		end
+
 		# @attribute [Array] The items in the queue.
 		attr :items
 
@@ -40,6 +55,10 @@ def empty?
 
 		# Add an item to the queue.
 		def push(item)
+			if @closed
+				raise ClosedError, "Cannot push items to a closed queue."
+			end
+
 			@items << item
 
 			@available.signal unless self.empty?
@@ -52,6 +71,10 @@ def <<(item)
 
 		# Add multiple items to the queue.
 		def enqueue(*items)
+			if @closed
+				raise ClosedError, "Cannot enqueue items to a closed queue."
+			end
+
 			@items.concat(items)
 
 			@available.signal unless self.empty?
@@ -60,6 +83,10 @@ def enqueue(*items)
 		# Remove and return the next item from the queue.
 		def dequeue
 			while @items.empty?
+				if @closed
+					return nil
+				end
+
 				@available.wait
 			end
 
@@ -120,9 +147,17 @@ def initialize(limit = 1, full: Notification.new, **options)
 		# @attribute [Integer] The maximum number of items that can be enqueued.
 		attr :limit
 
+		def close
+			super
+
+			while @full.waiting?
+				@full.signal(nil)
+			end
+		end
+
 		# @returns [Boolean] Whether trying to enqueue an item would block.
 		def limited?
-			@items.size >= @limit
+			!@closed && @items.size >= @limit
 		end
 
 		# Add an item to the queue.
@@ -149,6 +184,10 @@ def enqueue(*items)
 					@full.wait
 				end
 
+				if @closed
+					raise ClosedError, "Cannot enqueue items to a closed queue."
+				end
+
 				available = @limit - @items.size
 				@items.concat(items.shift(available))