Add load manangement.

Author: ioquatix

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

@@ -5,19 +5,20 @@
 
 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: nil, **options)
+	def Barrier(parent: Async::Idler.new, **options)
 		Sync(**options) do |task|
-			parent ||= task
-			
 			barrier = ::Async::Barrier.new(parent: parent)
 
 			yield barrier