Move promotion logic inside workers

Fix: #32

This remove a lot of busy work from the server, and allow to promote
as soon as the condition is reached.

The small downside is that we need a lock between processes to ensure
two workers won't try to promote at the same time.

This also means that the mold selector is removed. It wasn't that
interesting anyway, because the Unicorn/Pitchfork architecute introduce
a bias in request distribution, worker#0 will almost always get requests
first so will almost always be the best candidate for promotion.

Dockerfile

@@ -1,4 +1,4 @@
-FROM ruby:3.1
+FROM ruby:3.2
 RUN apt-get update -y && apt-get install -y ragel socat netcat smem apache2-utils
 WORKDIR /app
 CMD [ "bash" ]

docs/CONFIGURATION.md

@@ -334,30 +334,6 @@ By default automatic reforking isn't enabled.
 
 Make sure to read the [fork safety guide](FORK_SAFETY.md) before enabling reforking.
 
-### `mold_selector`
-
-Sets the mold selector implementation.
-
-```ruby
-mold_selector do |server|
-  candidate = server.children.workers.sample # return an random worker
-  server.logger.info("worker=#{worker.nr} pid=#{worker.pid} selected as new mold")
-  candidate
-end
-```
-
-The has access to `server.children` a `Pitchfork::Children` instance.
-This object can be used to introspect the state of the cluster and select the most
-appropriate worker to be used as the new mold from which workers will be reforked.
-
-The default implementation selects the worker with the least
-amount of shared memory. This heuristic aim to select the most
-warmed up worker.
-
-This should be considered a very advanced API and it is discouraged
-to use it unless you are confident you have a clear understanding
-of pitchfork's architecture.
-
 ## Rack Features
 
 ### `default_middleware`

lib/pitchfork.rb

@@ -171,7 +171,7 @@ def self.clean_fork(&block)
 
 %w(
   const socket_helper stream_input tee_input mem_info children message http_parser
-  refork_condition mold_selector configurator tmpio http_response worker http_server
+  refork_condition configurator tmpio http_response worker http_server
 ).each do |s|
   require_relative "pitchfork/#{s}"
 end

lib/pitchfork/configurator.rb

@@ -51,10 +51,9 @@ class Configurator
         end
       },
       :after_worker_ready => lambda { |server, worker|
-        server.logger.info("worker=#{worker.nr} ready")
+        server.logger.info("worker=#{worker.nr} gen=#{worker.generation} ready")
       },
       :early_hints => false,
-      :mold_selector => MoldSelector::LeastSharedMemory.new,
       :refork_condition => nil,
       :check_client_connection => false,
       :rewindable_input => true,
@@ -139,10 +138,6 @@ def after_worker_exit(*args, &block)
       set_hook(:after_worker_exit, block_given? ? block : args[0], 3)
     end
 
-    def mold_selector(*args, &block)
-      set_hook(:mold_selector, block_given? ? block : args[0], 3)
-    end
-
     def timeout(seconds)
       set_int(:timeout, seconds, 3)
       # POSIX says 31 days is the smallest allowed maximum timeout for select()

lib/pitchfork/flock.rb

@@ -0,0 +1,51 @@
+require 'tempfile'
+
+module Pitchfork
+  class Flock
+    Error = Class.new(StandardError)
+
+    def initialize(name)
+      @name = name
+      @file = Tempfile.create([name, '.lock'])
+      @file.write("#{Process.pid}\n")
+      @file.flush
+      @owned = false
+    end
+
+    def at_fork
+      @owned = false
+      @file.close
+      @file = File.open(@file.path, "w")
+      nil
+    end
+
+    def unlink
+      File.unlink(@file.path)
+    rescue Errno::ENOENT
+      false
+    end
+
+    def try_lock
+      raise Error, "Pitchfork::Flock(#{@name}) trying to lock an already owned lock" if @owned
+
+      if @file.flock(File::LOCK_EX | File::LOCK_NB)
+        @owned = true
+      else
+        false
+      end
+    end
+
+    def unlock
+      raise Error, "Pitchfork::Flock(#{@name}) trying to unlock a non-owned lock" unless @owned
+
+      begin
+        if @file.flock(File::LOCK_UN)
+          @owned = false
+          true
+        else
+          false
+        end
+      end
+    end
+  end
+end

lib/pitchfork/http_server.rb

@@ -1,5 +1,6 @@
 # -*- encoding: binary -*-
 require 'pitchfork/pitchfork_http'
+require 'pitchfork/flock'
 
 module Pitchfork
   # This is the process manager of Pitchfork. This manages worker
@@ -13,7 +14,7 @@ class HttpServer
                   :listener_opts, :children,
                   :orig_app, :config, :ready_pipe,
                   :default_middleware, :early_hints
-    attr_writer   :after_worker_exit, :after_worker_ready, :refork_condition, :mold_selector
+    attr_writer   :after_worker_exit, :after_worker_ready, :refork_condition
 
     attr_reader :logger
     include Pitchfork::SocketHelper
@@ -65,6 +66,8 @@ def initialize(app, options = {})
       @respawn = false
       @last_check = time_now
       @default_middleware = true
+      @promotion_lock = Flock.new("pitchfork-promotion")
+
       options = options.dup
       @ready_pipe = options.delete(:ready_pipe)
       @init_listeners = options[:listeners].dup || []
@@ -266,7 +269,7 @@ def monitor_loop(sleep = true)
         end
         if @respawn
           maintain_worker_count
-          automatically_refork_workers if REFORKING_AVAILABLE
+          restart_outdated_workers if REFORKING_AVAILABLE
         end
 
         master_sleep(sleep_time) if sleep
@@ -318,6 +321,7 @@ def stop(graceful = true)
         reap_all_workers
       end
       kill_each_child(:KILL)
+      @promotion_lock.unlink
     end
 
     def rewindable_input
@@ -429,17 +433,17 @@ def trigger_refork
       end
 
       unless @children.pending_promotion?
-        @children.refresh
-        if new_mold = @mold_selector.call(self)
+        if new_mold = @children.fresh_workers.first
           @children.promote(new_mold)
         else
-          logger.error("The mold select didn't return a candidate")
+          logger.error("No children at all???")
         end
       else
       end
     end
 
     def after_fork_internal
+      @promotion_lock.at_fork
       @control_socket[0].close_write # this is master-only, now
       @ready_pipe.close if @ready_pipe
       Pitchfork::Configurator::RACKUP.clear
@@ -484,6 +488,7 @@ def spawn_initial_mold
       mold = Worker.new(nil)
       mold.create_socketpair!
       mold.pid = Pitchfork.clean_fork do
+        @promotion_lock.try_lock
         mold.after_fork_in_child
         build_app!
         bind_listeners!
@@ -533,34 +538,21 @@ def maintain_worker_count
       @children.each_worker { |w| w.nr >= worker_processes and w.soft_kill(:QUIT) }
     end
 
-    def automatically_refork_workers
+    def restart_outdated_workers
       # If we're already in the middle of forking a new generation, we just continue
-      if @children.mold
-        # We don't shutdown any outdated worker if any worker is already being spawned
-        # or a worker is exiting. Workers are only reforked one by one to minimize the
-        # impact on capacity.
-        # In the future we may want to use a dynamic limit, e.g. 10% of workers may be down at
-        # a time.
-        return if @children.pending_workers?
-        return if @children.workers.any?(&:exiting?)
-
-        if outdated_worker = @children.workers.find { |w| w.generation < @children.mold.generation }
-          logger.info("worker=#{outdated_worker.nr} pid=#{outdated_worker.pid} restarting")
-          outdated_worker.soft_kill(:QUIT)
-          return # That's all folks
-        end
-      end
+      return unless @children.mold
 
-      # If all workers are alive and well, we can consider reforking a new generation
-      if @refork_condition
-        @children.refresh
-        if @refork_condition.met?(@children, logger)
-          logger.info("Refork condition met, scheduling a promotion")
-          unless @sig_queue.include?(:USR2)
-            @sig_queue << :USR2
-            awaken_master
-          end
-        end
+      # We don't shutdown any outdated worker if any worker is already being spawned
+      # or a worker is exiting. Workers are only reforked one by one to minimize the
+      # impact on capacity.
+      # In the future we may want to use a dynamic limit, e.g. 10% of workers may be down at
+      # a time.
+      return if @children.pending_workers?
+      return if @children.workers.any?(&:exiting?)
+
+      if outdated_worker = @children.workers.find { |w| w.generation < @children.mold.generation }
+        logger.info("worker=#{outdated_worker.nr} pid=#{outdated_worker.pid} restarting")
+        outdated_worker.soft_kill(:QUIT)
       end
     end
 
@@ -721,6 +713,12 @@ def worker_loop(worker)
           client = false if client == :wait_readable
           if client
             case client
+            when Message::PromoteWorker
+              if @promotion_lock.try_lock
+                logger.info("Refork asked by master, promoting ourselves")
+                worker.tick = time_now.to_i
+                return worker.promoted!
+              end
             when Message
               worker.update(client)
             else
@@ -729,11 +727,21 @@ def worker_loop(worker)
             end
             worker.tick = time_now.to_i
           end
-          return if worker.mold? # We've been promoted we can exit the loop
         end
 
         # timeout so we can .tick and keep parent from SIGKILL-ing us
         worker.tick = time_now.to_i
+        if @refork_condition && !worker.outdated?
+          if @refork_condition.met?(worker, logger)
+            if @promotion_lock.try_lock
+              logger.info("Refork condition met, promoting ourselves")
+              return worker.promote! # We've been promoted we can exit the loop
+            else
+              # TODO: if we couldn't acquire the lock, we should backoff the refork_condition to avoid hammering the lock
+            end
+          end
+        end
+
         waiter.get_readers(ready, readers, @timeout * 500) # to milliseconds, but halved
       rescue => e
         Pitchfork.log_error(@logger, "listen loop error", e) if readers[0]
@@ -743,7 +751,8 @@ def worker_loop(worker)
     def mold_loop(mold)
       readers = init_mold_process(mold)
       waiter = prep_readers(readers)
-      mold.acknowlege_promotion(@control_socket[1])
+      mold.declare_promotion(@control_socket[1])
+      @promotion_lock.unlock
       ready = readers.dup
 
       begin

lib/pitchfork/refork_condition.rb

@@ -7,9 +7,9 @@ def initialize(request_counts)
         @limits = request_counts
       end
 
-      def met?(children, logger)
-        if limit = @limits[children.last_generation]
-          if worker = children.fresh_workers.find { |w| w.requests_count >= limit }
+      def met?(worker, logger)
+        if limit = @limits[worker.generation]
+          if worker.requests_count >= limit
             logger.info("worker=#{worker.nr} pid=#{worker.pid} processed #{worker.requests_count} requests, triggering a refork")
             return true
           end