Move sleep cluster logic to its own class (#3746)

* Move cluster logic to its own class

Simplify the calculation. This is an alternative to #3723.

Puma 6.6.1 behavior documented here https://gist.github.com/schneems/cef38e9448dfb72943d13050a7da0869

This changes the puma 7 behavior to:

- Starts sleeping the event loop before being overloaded (the prior `pool.busy_threads` included in the ready @todo queue) and would only fire. This is closer to puma 6.6.1 behavior
- Sleeps a consistent proportional amount, closer to puma 7 behavior, versus puma 6 used a static value.
- Sleeping behavior is restricted to 2 or more workers, verus puma 7 was accidentally enabled for clusters with only 1 worker

Fixes #3740. This `wait_for_less_busy_worker` value is now treated as a maximum value for the sleep calculation. It also now respects a value of 0 to mean "don't sleep at all" which was the prior behavior of that value.

* Whitespace

* Remove unused method

The busy_thread call is now accessed directly through the thread pool and only used in one place.

* Add docs about server <-> reactor relationship

The code doesn't make it clear who is calling whom when looking at it in isolation.

* Make maximum 25x thread count and allow for overage

Puma can take in more requests than it has threads. This change preserves the same logic as before, but it doesn't hit maximum sleep until it is at 25x the number of max threads. That means that if a server was using 5 threads,  before it would hit 0.005 sleep if those five threads were busy, now if (only) five threads are busy it will sleep 0.0002 seconds.

* Move order of comparison

If busy threads is zero we don't need to calculate percentage.

* Refactor out clamp

Because we're clamped at the numerator, the division will naturally tend towards 1.0. We don't need a second clamp on the result. This allows us to remove an intermediate variable and multiply directly on the return calculation.

* Lazy max_threads

There are proposals to make thread counts dynamic RE #3658. If this happens we need to pull in the current value, and cannot rely on it being set once.

* Avoid re-checking the `max_delay` number every iteration

* Refactor out branching

The equation already holds up the tested properties without needing to special-case when busy threads is zero. Removes a branching conditional.

* Push worker logic into delay class

The logic of whether or not the calculation should be performed based on worker count can be calculated once and not repeated on every iteration.

* Optimize case where no threads are busy

Moves the condition from checking if the delay is zero to checking if the input is zero earlier. This is prevents some calculations and preserves the same number of conditionals.

Author: schneems

lib/puma/cluster_accept_loop_delay.rb

@@ -0,0 +1,92 @@
+# frozen_string_literal: true
+
+module Puma
+  # Calculate a delay value for sleeping when running in clustered mode
+  #
+  # The main reason this is a class is so it can be unit tested independently.
+  # This makes modification easier in the future if we can encode properties of the
+  # delay into a test instead of relying on end-to-end testing only.
+  #
+  # This is an imprecise mechanism to address specific goals:
+  #
+  # - Evenly distribute requests across all workers at start
+  # - Evenly distribute CPU resources across all workers
+  #
+  # ## Goal: Distribute requests across workers at start
+  #
+  # There was a perf bug in Puma where one worker would wake up slightly before the rest and accept
+  # all the requests on the socket even though it didn't have enough resources to process all of them.
+  # This was originally fixed by never calling accept when a worker had more requests than threads
+  # already https://github.com/puma/puma/pull/3678/files/2736ebddb3fc8528e5150b5913fba251c37a8bf7#diff-a95f46e7ce116caddc9b9a9aa81004246d5210d5da5f4df90a818c780630166bL251-L291
+  #
+  # With the introduction of true keepalive support, there are two ways a request can come in:
+  # - A new request from a new client comes into the socket and it must be "accept"-d
+  # - A keepalive request is served and the connection is retained. Another request is then accepted
+  #
+  # Ideally the server handles requests in the order they come in, and ideally it doesn't accept more requests than it can handle.
+  # These goals are contradictory, because when the server is at maximum capacity due to keepalive connections, it could mean we
+  # block all new requests, even if those came in before the new request on the older keepalive connection.
+  #
+  # ## Distribute CPU resources across all workers
+  #
+  # - This issue was opened https://github.com/puma/puma/issues/2078
+  #
+  # There are several entangled issues and it's not exactly clear the root cause, but the observable outcome
+  # was that performance was better with a small sleep, and that eventually became the default.
+  #
+  # An attempt to describe why this works is here: https://github.com/puma/puma/issues/2078#issuecomment-3287032470.
+  #
+  # Summarizing: The delay is for tuning the rate at which "accept" is called on the socket.
+  # Puma works by calling "accept" nonblock on the socket in a loop. When there are multiple workers,
+  # (processes) then they will "race" to accept a request at roughly the same rate. However if one
+  # worker has all threads busy processing requests, then accepting a new request might "steal" it from
+  # a less busy worker. If a worker has no work to do, it should loop as fast as possible.
+  #
+  # ## Solution(s): Distribute requests across workers at start
+  #
+  # For now, both goals are framed as "load balancing" across workers (processes) and achieved through
+  # the same mechanism of sleeping longer to delay busier workers. Rather than the prior Puma 6.x
+  # and earlier behavior of using a binary on/off sleep value, we increase it an amound proportional
+  # to the load the server is under. Capping the maximum delay to the scenario where all threads are busy
+  # and the todo list has reached a multiplier of the maximum number of threads.
+  #
+  # Private: API may change unexpectedly
+  class ClusterAcceptLoopDelay
+    attr_reader :max_threads, :max_delay
+
+    # Initialize happens once, `call` happens often. Push global calculations here
+    def initialize(
+        # Number of workers in the cluster
+        workers: ,
+        # Maximum delay in seconds i.e. 0.005 is 5 microseconds
+        max_delay: # In seconds i.e. 0.005 is 5 microseconds
+
+      )
+      @on = max_delay > 0 && workers >= 2
+      @max_delay = max_delay.to_f
+
+      # Reach maximum delay when `max_threads * overload_multiplier` is reached in the system
+      @overload_multiplier = 25.0
+    end
+
+    def on?
+      @on
+    end
+
+    # We want the extreme values of this delay to be known (minimum and maximum) as well as
+    # a predictable curve between the two. i.e. no step functions or hard cliffs.
+    #
+    # Return value is always numeric. Returns 0 if there should be no delay
+    def calculate(
+      # Number of threads working right now, plus number of requests in the todo list
+      busy_threads_plus_todo:,
+      # Maximum number of threads in the pool, note that the busy threads (alone) may go over this value at times
+      # if the pool needs to be reaped. The busy thread plus todo count may go over this value by a large amount
+      max_threads:
+    )
+      max_value = @overload_multiplier * max_threads
+      # Approaches max delay when `busy_threads_plus_todo` approaches `max_value`
+      return max_delay * busy_threads_plus_todo.clamp(0, max_value) / max_value
+    end
+  end
+end

lib/puma/server.rb

@@ -13,6 +13,7 @@
 require_relative 'util'
 require_relative 'request'
 require_relative 'configuration'
+require_relative 'cluster_accept_loop_delay'
 
 require 'socket'
 require 'io/wait' unless Puma::HAS_NATIVE_IO_WAIT
@@ -58,7 +59,6 @@ def handle_request(client, requests)
     attr_accessor :app
     attr_accessor :binder
 
-
     # Create a server for the rack app +app+.
     #
     # +log_writer+ is a Puma::LogWriter object used to log info and error messages.
@@ -110,6 +110,10 @@ def initialize(app, events = nil, options = {})
       @enable_keep_alives      &&= @queue_requests
       @io_selector_backend       = @options[:io_selector_backend]
       @http_content_length_limit = @options[:http_content_length_limit]
+      @cluster_accept_loop_delay = ClusterAcceptLoopDelay.new(
+        workers: @options[:workers],
+        max_delay: @options[:wait_for_less_busy_worker] || 0 # Real default is in Configuration::DEFAULTS, this is for unit testing
+      )
 
       if @options[:fiber_per_request]
         singleton_class.prepend(FiberPerRequest)
@@ -245,11 +249,6 @@ def pool_capacity
       @thread_pool&.pool_capacity
     end
 
-    # @!attribute [r] busy_threads
-    def busy_threads
-      @thread_pool&.busy_threads
-    end
-
     # Runs the server.
     #
     # If +background+ is true (the default) then a thread is spun
@@ -266,7 +265,11 @@ def run(background=true, thread_name: 'srv')
       @thread_pool = ThreadPool.new(thread_name, options) { |client| process_client client }
 
       if @queue_requests
-        @reactor = Reactor.new(@io_selector_backend) { |c| reactor_wakeup c }
+        @reactor = Reactor.new(@io_selector_backend) { |c|
+          # Inversion of control, the reactor is calling a method on the server when it
+          # is done buffering a request or receives a new request from a keepalive connection.
+          self.reactor_wakeup(c)
+        }
         @reactor.run
       end
 
@@ -291,6 +294,9 @@ def run(background=true, thread_name: 'srv')
     # This method is called from the Reactor thread when a queued Client receives data,
     # times out, or when the Reactor is shutting down.
     #
+    # While the code lives in the Server, the logic is executed on the reactor thread, independently
+    # from the server.
+    #
     # It is responsible for ensuring that a request has been completely received
     # before it starts to be processed by the ThreadPool. This may be known as read buffering.
     # If read buffering is not done, and no other read buffering is performed (such as by an application server
@@ -339,7 +345,6 @@ def handle_servers
         pool = @thread_pool
         queue_requests = @queue_requests
         drain = options[:drain_on_shutdown] ? 0 : nil
-        max_flt = @max_threads.to_f
 
         addr_send_name, addr_value = case options[:remote_address]
         when :value
@@ -384,15 +389,13 @@ def handle_servers
                 # clients until the code is finished.
                 pool.wait_while_out_of_band_running
 
-                # only use delay when clustered and busy
-                if pool.busy_threads >= @max_threads
-                  if @clustered
-                    delay = 0.0001 * ((@reactor&.reactor_size || 0) + pool.busy_threads * 1.5)/max_flt
-                    sleep delay
-                  else
-                    # use small sleep for busy single worker
-                    sleep 0.0001
-                  end
+                # A well rested herd (cluster) runs faster
+                if @cluster_accept_loop_delay.on? && (busy_threads_plus_todo = pool.busy_threads) > 0
+                  delay = @cluster_accept_loop_delay.calculate(
+                    max_threads: @max_threads,
+                    busy_threads_plus_todo: busy_threads_plus_todo
+                  )
+                  sleep(delay)
                 end
 
                 io = begin

test/test_cluster_accept_loop_delay.rb

@@ -0,0 +1,76 @@
+# frozen_string_literal: true
+
+require_relative "helper"
+require "puma/cluster_accept_loop_delay"
+
+class TestClusterAcceptLoopDelay < PumaTest
+  parallelize_me!
+
+  def test_off_when_fewer_than_two_workers
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 2,
+      max_delay: 1
+    )
+    assert_equal true, cal_delay.on?
+
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 1,
+      max_delay: 1
+    )
+    assert_equal false, cal_delay.on?
+  end
+
+  def test_zero_max_delay_always_returns_zero
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 2,
+      max_delay: 0
+    )
+    assert_equal false, cal_delay.on?
+    assert_equal 0, cal_delay.calculate(busy_threads_plus_todo: 0, max_threads: 16)
+    assert_equal 0, cal_delay.calculate(busy_threads_plus_todo: 42, max_threads: 16)
+    assert_equal 0, cal_delay.calculate(busy_threads_plus_todo: 42 * 42, max_threads: 16)
+  end
+
+  def test_zero_busy_threads_plus_todo_always_returns_zero
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 2,
+      max_delay: 0.005
+    )
+
+    assert_equal 0, cal_delay.calculate(busy_threads_plus_todo: 0, max_threads: 10)
+  end
+
+  def test_linear_increase_with_busy_threads_plus_todo
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 2,
+      max_delay: 0.05
+    )
+
+    assert_in_delta 0, cal_delay.calculate(busy_threads_plus_todo: 0, max_threads: 1), 0.001
+    assert_in_delta 0.002, cal_delay.calculate(busy_threads_plus_todo: 1, max_threads: 1), 0.001
+    assert_in_delta 0.05, cal_delay.calculate(busy_threads_plus_todo: 25, max_threads: 1), 0.001
+    assert_in_delta 0.05, cal_delay.calculate(busy_threads_plus_todo: 26, max_threads: 1), 0.001
+  end
+
+  def test_always_return_float_when_non_zero
+    # Dividing integers accidentally returns 0 so want to make sure we are correctly converting to float before division
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 2,
+      max_delay: Integer(5)
+    )
+
+    assert_in_delta 0, cal_delay.calculate(busy_threads_plus_todo: 0.to_f, max_threads: Integer(1)), 0.001
+    assert_equal Float, cal_delay.calculate(busy_threads_plus_todo: Integer(25), max_threads: Integer(1)).class
+    assert_in_delta 5, cal_delay.calculate(busy_threads_plus_todo: 25, max_threads: Integer(1)), 0.001
+  end
+
+  def test_extreme_busy_values_produce_sensible_delays
+    cal_delay = Puma::ClusterAcceptLoopDelay.new(
+      workers: 2,
+      max_delay: 0.05
+    )
+
+    assert_in_delta 0, cal_delay.calculate(busy_threads_plus_todo: -10, max_threads: 5), 0.001
+    assert_in_delta 0.05, cal_delay.calculate(busy_threads_plus_todo: Float::INFINITY, max_threads: 5), 0.001
+  end
+end