diff --git a/lib/async/scheduler.rb b/lib/async/scheduler.rb index ffdd9e17..75cac7f7 100644 --- a/lib/async/scheduler.rb +++ b/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 diff --git a/releases.md b/releases.md index 0510ff3d..a778a3c6 100644 --- a/releases.md +++ b/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`. diff --git a/test/async/scheduler.rb b/test/async/scheduler.rb index 6870b356..76e1926d 100644 --- a/test/async/scheduler.rb +++ b/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