Implement UNIX-style load averages

lib/async/scheduler.rb

@@ -32,6 +32,13 @@ class Scheduler < Node
 			value == "true" ? true : nil
 		end
 
+		# The periods over which exponentially weighted load averages are computed, in seconds.
+		LOAD_AVERAGE_PERIODS = [60.0, 5 * 60.0, 15 * 60.0].freeze
+
+		# How much run loop time to accumulate before updating the load averages.
+		LOAD_AVERAGE_UPDATE_INTERVAL = 1.0
+		private_constant :LOAD_AVERAGE_UPDATE_INTERVAL
+
 		# Raised when an operation is attempted on a closed scheduler.
 		class ClosedError < RuntimeError
 			# Create a new error.
@@ -85,8 +92,10 @@ def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_
 
 			@blocked = 0
 
-			@busy_time = 0.0
-			@idle_time = 0.0
+			@load = 0.0
+			@load_average = Array.new(LOAD_AVERAGE_PERIODS.size, 0.0).freeze
+			@load_busy_time = 0.0
+			@load_total_time = 0.0
 
 			@timers = ::IO::Event::Timers.new
 
@@ -105,24 +114,44 @@ def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_
 		#
 		# @returns [Float] The load of the scheduler. 0.0 means no load, 1.0 means fully loaded or over-loaded.
 		def load
-			total_time = @busy_time + @idle_time
+			# Weight the last completed window by the portion of the update interval that the current window has not yet covered:
+			remaining = LOAD_AVERAGE_UPDATE_INTERVAL - @load_total_time
 
-			# If the total time is zero, then the load is zero:
-			return 0.0 if total_time.zero?
-
-			# We normalize to a 1 second window:
-			if total_time > 1.0
-				ratio = 1.0 / total_time
-				@busy_time *= ratio
-				@idle_time *= ratio
-
-				# We don't need to divide here as we've already normalised it to a 1s window:
-				return @busy_time
+			if remaining > 0.0
+				return (@load * remaining + @load_busy_time) / LOAD_AVERAGE_UPDATE_INTERVAL
 			else
-				return @busy_time / total_time
+				return @load_busy_time / @load_total_time
 			end
 		end
 
+		# The scheduler load averaged over one, five, and fifteen minutes.
+		#
+		# @public Since *Async v2.46*.
+		# @returns [Array(Float)] The scheduler load averages.
+		def load_averages
+			return @load_average
+		end
+
+		private def update_load_average(busy_duration, total_duration)
+			return if total_duration <= 0.0
+
+			@load_busy_time += busy_duration
+			@load_total_time += total_duration
+			return if @load_total_time < LOAD_AVERAGE_UPDATE_INTERVAL
+
+			total_time = @load_total_time
+			load = @load_busy_time / total_time
+			@load = load
+
+			@load_average = LOAD_AVERAGE_PERIODS.map.with_index do |period, index|
+				decay = Math.exp(-total_time / period)
+				@load_average[index] * decay + load * (1.0 - decay)
+			end.freeze
+
+			@load_busy_time = 0.0
+			@load_total_time = 0.0
+		end
+
 		# Invoked when the fiber scheduler is being closed.
 		#
 		# Executes the run loop until all tasks are finished, then closes the scheduler.
@@ -483,8 +512,7 @@ def io_select(...)
 			idle_duration = @selector.idle_duration
 			busy_duration = total_duration - idle_duration
 
-			@busy_time += busy_duration
-			@idle_time += idle_duration
+			update_load_average(busy_duration, total_duration)
 
 			# The reactor still has work to do:
 			return true

releases.md

@@ -1,5 +1,10 @@
 # Releases
 
+## Unreleased
+
+  - Added `Async::Scheduler#load_averages`, which returns one, five, and fifteen minute exponentially weighted moving averages of the scheduler load, suitable for monitoring and autoscaling signals.
+  - `Async::Scheduler#load` now reports the load over a sliding one-second window and no longer mutates internal state, so reading it has no side effects.
+
 ## 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`.

test/async/scheduler.rb

@@ -72,16 +72,83 @@
 	end
 
 	with "#load" do
-		it "normalizes load over a one second window" do
+		it "returns the load over the last update window" do
 			scheduler = Async::Scheduler.new
 
-			scheduler.instance_variable_set(:@busy_time, 2.0)
-			scheduler.instance_variable_set(:@idle_time, 2.0)
+			scheduler.send(:update_load_average, 30.0, 60.0)
 
 			expect(scheduler.load).to be == 0.5
 		ensure
 			scheduler&.close
 		end
+
+		it "blends the current window into the last completed window" do
+			scheduler = Async::Scheduler.new
+
+			# Complete a fully loaded window:
+			scheduler.send(:update_load_average, 1.0, 1.0)
+			expect(scheduler.load).to be == 1.0
+
+			# Half of the next window elapses fully idle:
+			scheduler.send(:update_load_average, 0.0, 0.5)
+			expect(scheduler.load).to be == 0.5
+		ensure
+			scheduler&.close
+		end
+
+		it "reflects a partial window before the first update" do
+			scheduler = Async::Scheduler.new
+
+			scheduler.send(:update_load_average, 0.25, 0.5)
+
+			expect(scheduler.load).to be == 0.25
+		ensure
+			scheduler&.close
+		end
+
+		it "does not change based on how often it is read" do
+			scheduler = Async::Scheduler.new
+			scheduler.send(:update_load_average, 30.0, 60.0)
+
+			load = scheduler.load
+			3.times do
+				expect(scheduler.load).to be == load
+			end
+		ensure
+			scheduler&.close
+		end
+	end
+
+	with "#load_averages" do
+		it "batches updates over one second" do
+			scheduler = Async::Scheduler.new
+			scheduler.send(:update_load_average, 0.25, 0.5)
+
+			expect(scheduler.load_averages).to be == [0.0, 0.0, 0.0]
+
+			scheduler.send(:update_load_average, 0.5, 0.5)
+
+			expected = Async::Scheduler::LOAD_AVERAGE_PERIODS.map do |period|
+				0.75 * (1.0 - Math.exp(-1.0 / period))
+			end
+
+			expect(scheduler.load_averages).to be == expected
+		ensure
+			scheduler&.close
+		end
+
+		it "tracks one, five, and fifteen minute exponentially weighted averages" do
+			scheduler = Async::Scheduler.new
+			scheduler.send(:update_load_average, 60.0, 60.0)
+
+			expected = Async::Scheduler::LOAD_AVERAGE_PERIODS.map do |period|
+				1.0 - Math.exp(-60.0 / period)
+			end
+
+			expect(scheduler.load_averages).to be == expected
+		ensure
+			scheduler&.close
+		end
 	end
 
 	with "#close" do