Introduce Kernel#Barrier as a top level scheduling operation. (#426)

Author: ioquatix

lib/async.rb

@@ -9,6 +9,7 @@
 
 require_relative "kernel/async"
 require_relative "kernel/sync"
+require_relative "kernel/barrier"
 
 # Asynchronous programming framework.
 module Async

lib/async/idler.rb

@@ -13,10 +13,13 @@ class Idler
 		# @parameter maximum_load [Numeric] The maximum load before we start shedding work.
 		# @parameter backoff [Numeric] The initial backoff time, used for delaying work.
 		# @parameter parent [Interface(:async) | Nil] The parent task to use for async operations.
-		def initialize(maximum_load = 0.8, backoff: 0.01, parent: nil)
+		def initialize(maximum_load = 0.8, backoff: 0.001, parent: nil)
 			@maximum_load = maximum_load
 			@backoff = backoff
+			@current = backoff
+			
 			@parent = parent
+			@mutex = Mutex.new
 		end
 
 		# Wait until the system is idle, then execute the given block in a new task.
@@ -38,20 +41,29 @@ def async(*arguments, parent: (@parent or Task.current), **options, &block)
 		#
 		# If the scheduler is overloaded, this method will sleep for an exponentially increasing amount of time.
 		def wait
-			scheduler = Fiber.scheduler
-			backoff = nil
-			
-			while true
-				load = scheduler.load
-				
-				break if load < @maximum_load
+			@mutex.synchronize do
+				scheduler = Fiber.scheduler
 
-				if backoff
-					sleep(backoff)
-					backoff *= 2.0
-				else
-					scheduler.yield
-					backoff = @backoff
+				while true
+					load = scheduler.load
+					
+					if load <= @maximum_load
+						# Even though load is okay, if @current is high, we were recently overloaded. Sleep proportionally to prevent burst after load drop:
+						if @current > @backoff
+							# Sleep a fraction of @current to rate limit:
+							sleep(@current - @backoff)
+							
+							# Decay @current gently towards @backoff:
+							alpha = 0.99
+							@current *= alpha + (1.0 - alpha) * (load / @maximum_load)
+						end
+						
+						break
+					else
+						# We're overloaded, so increase backoff:
+						@current *= (load / @maximum_load)
+						sleep(@current)
+					end
 				end
 			end
 		end

lib/kernel/barrier.rb

@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require_relative "sync"
+require_relative "../async/barrier"
+require_relative "../async/idler"
+
+module Kernel
+	# Create a barrier, yield it to the block, and then wait for all tasks to complete.
+	#
+	# If no scheduler is running, one will be created automatically for the duration of the block.
+	#
+	# By default, the barrier uses an `Async::Idler` to manage load, but this can be overridden by providing a different parent or `nil` to disable load management.
+	#
+	# @parameter parent [Task | Semaphore | Nil] The parent for holding any children tasks.
+	# @parameter **options [Hash] Additional options passed to {Kernel::Sync}.
+	# @public Since *Async v2.34*.
+	def Barrier(parent: Async::Idler.new, **options)
+		Sync(**options) do |task|
+			barrier = ::Async::Barrier.new(parent: parent)
+			
+			yield barrier
+			
+			barrier.wait
+		ensure
+			barrier&.stop
+		end
+	end
+end

releases.md

@@ -1,5 +1,28 @@
 # Releases
 
+## Unreleased
+
+### `Kernel::Barrier` Convenience Interface
+
+Starting multiple concurrent tasks and waiting for them to finish is a common pattern. This change introduces a small ergonomic helper, `Barrier`, defined in `Kernel`, that encapsulates this behavior: it creates an `Async::Barrier`, yields it to a block, waits for completion (using `Sync` to run a reactor if needed), and ensures remaining tasks are stopped on exit.
+
+``` ruby
+require 'async'
+
+Barrier do |barrier|
+	3.times do |i|
+		barrier.async do |task|
+			sleep(rand * 0.1)  # Simulate work
+			puts "Task #{i} completed"
+		end
+	end
+end
+
+# All tasks are guaranteed to complete or be stopped when the block exits.
+```
+
+If an exception is raised by a task, it will be propagated to the caller, and any remaining tasks will be stopped. The `parent:` parameter can be used to specify a parent task for the barrier, otherwise it will use the current task if available, or create a new reactor if not.
+
 ## v2.33.0
 
   - Introduce `Async::Promise.fulfill` for optional promise resolution.

test/async/idler.rb

@@ -27,7 +27,7 @@
 		end
 
 		# This test must be longer than the idle calculation window (1s)...
-		sleep 1.1
+		sleep 2.0
 
 		# Verify that the load is within the desired range:
 		expect(Fiber.scheduler.load).to be_within(0.2).of(0.5)

test/kernel/barrier.rb

@@ -0,0 +1,179 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "kernel/barrier"
+
+describe Kernel do
+	with "#Barrier" do
+		it "should create a barrier, yield it, wait, and stop automatically" do
+			finished = 0
+			results = []
+			
+			Barrier do |barrier|
+				3.times do |i|
+					barrier.async do |task|
+						sleep(0.01 * i)  # Different completion times
+						results << "task_#{i}"
+						finished += 1
+					end
+				end
+			end
+			
+			expect(finished).to be == 3
+			expect(results.size).to be == 3
+		end
+		
+		it "should handle exceptions and still clean up properly" do
+			exception_raised = false
+			
+			expect do
+				Barrier do |barrier|
+					barrier.async do |task|
+						raise "Test exception"
+					end
+					
+					barrier.async do |task|
+						sleep(0.1)  # This should be stopped
+					end
+				end
+			end.to raise_exception(RuntimeError, message: be =~ /Test exception/)
+			
+			# The barrier should have been cleaned up despite the exception.
+		end
+		
+		it "should support parent parameter" do
+			parent_task = nil
+			child_task = nil
+			
+			Sync do |task|
+				parent_task = task
+				
+				Barrier(parent: task) do |barrier|
+					barrier.async do |async_task|
+						child_task = async_task
+						# While the child task is running, parent should be set:
+						expect(child_task.parent).to be == parent_task
+						sleep(0.01)
+					end
+				end
+			end
+			
+			expect(child_task).not.to be_nil
+		end
+		
+		it "should wait for all tasks to complete before returning" do
+			completion_order = []
+			
+			Barrier do |barrier|
+				3.times do |i|
+					barrier.async do |task|
+						sleep(0.01 * (3 - i))  # Reverse order completion
+						completion_order << i
+					end
+				end
+			end
+			
+			# All tasks should have completed
+			expect(completion_order.size).to be == 3
+			expect(completion_order.sort).to be == [0, 1, 2]
+		end
+		
+		it "should stop remaining tasks when block exits early" do
+			tasks = []
+			
+			Sync do |parent|
+				begin
+					Barrier do |barrier|
+						3.times do |i|
+							tasks << barrier.async do |task|
+								sleep(1)  # Long running task
+							end
+						end
+						
+						# Simulate early exit due to some condition
+						raise "Early exit"
+					end
+				rescue => e
+					# Expected exception
+				end
+				
+				# Wait for tasks to finish/stopped deterministically
+				tasks.each do |t|
+					begin
+						t.wait
+					rescue => e
+						# ignore errors from waiting on stopped tasks
+					end
+				end
+			end
+			
+			# All three tasks should have been stopped
+			expect(tasks.map(&:stopped?).all?).to be == true
+		end
+		
+		it "should handle empty barriers gracefully" do
+			result = nil
+			
+			expect do
+				result = Async::Barrier() do |barrier|
+					# No tasks added
+				end
+			end.not.to raise_exception
+			
+			# Should complete successfully
+		end
+		
+		it "should create a barrier, yield it, wait, and stop automatically" do
+			finished = 0
+			results = []
+			
+			Barrier do |barrier|
+				3.times do |i|
+					barrier.async do |task|
+						sleep(0.01 * i)
+						results << "task_#{i}"
+						finished += 1
+					end
+				end
+			end
+			
+			expect(finished).to be == 3
+			expect(results.size).to be == 3
+		end
+		
+		it "should handle exceptions and still clean up properly" do
+			expect do
+				Barrier do |barrier|
+					barrier.async do |task|
+						raise "Kernel helper exception"
+					end
+					barrier.async do |task|
+						sleep(0.1)
+					end
+				end
+			end.to raise_exception(RuntimeError, message: be =~ /Kernel helper exception/)
+		end
+		
+		it "should support parent parameter" do
+			parent_task = nil
+			child_task = nil
+			
+			Sync do |task|
+				parent_task = task
+				
+				Barrier(parent: task) do |barrier|
+					barrier.async do |async_task|
+						child_task = async_task
+						# While the child task is running, parent should be set:
+						expect(child_task.parent).to be == parent_task
+						sleep(0.01)
+					end
+				end
+			end
+			
+			expect(child_task).not.to be_nil
+		end
+	end
+end