Start extracting forking behaviour from Supervisor

Author: rosa

lib/solid_queue/log_subscriber.rb

@@ -118,7 +118,7 @@ def thread_error(event)
   end
 
   def graceful_termination(event)
-    attributes = event.payload.slice(:supervisor_pid, :supervised_pids)
+    attributes = event.payload.slice(:supervisor_pid, :supervised_processes)
 
     if event.payload[:shutdown_timeout_exceeded]
       warn formatted_event(event, action: "Supervisor wasn't terminated gracefully - shutdown timeout exceeded", **attributes)
@@ -128,7 +128,7 @@ def graceful_termination(event)
   end
 
   def immediate_termination(event)
-    info formatted_event(event, action: "Supervisor terminated immediately", **event.payload.slice(:supervisor_pid, :supervised_pids))
+    info formatted_event(event, action: "Supervisor terminated immediately", **event.payload.slice(:supervisor_pid, :supervised_processes))
   end
 
   def unhandled_signal_error(event)

lib/solid_queue/processes/registrable.rb

@@ -16,7 +16,7 @@ module Registrable
 
       def register
         @process = SolidQueue::Process.register \
-          kind: self.class.name.demodulize,
+          kind: kind,
           pid: pid,
           hostname: hostname,
           supervisor: try(:supervisor),

lib/solid_queue/supervisor.rb

@@ -2,26 +2,26 @@
 
 module SolidQueue
   class Supervisor < Processes::Base
-    include Signals
+    include Signals, Maintenance
 
     class << self
       def start(mode: :fork, load_configuration_from: nil)
         SolidQueue.supervisor = true
         configuration = Configuration.new(load_from: load_configuration_from)
 
-        new(*configuration.processes).start
+        Forks.new(*configuration.processes).start
       end
     end
 
-    def initialize(*configured_processes)
-      @configured_processes = Array(configured_processes)
-      @forks = {}
+    def initialize(*configuration)
+      @configuration = Array(configuration)
+      @processes = {}
     end
 
     def start
       run_callbacks(:boot) { boot }
 
-      start_forks
+      start_processes
       launch_maintenance_task
 
       supervise
@@ -34,141 +34,76 @@ def start
     end
 
     private
-      attr_reader :configured_processes, :forks
+      attr_reader :configuration, :processes
 
       def boot
         sync_std_streams
         setup_pidfile
         register_signal_handlers
       end
 
+      def start_processes
+        configuration.each { |configured_process| start_process(configured_process) }
+      end
+
       def supervise
         loop do
-          procline "supervising #{forks.keys.join(", ")}"
+          procline "supervising #{processes.keys.join(", ")}"
 
           process_signal_queue
-          reap_and_replace_terminated_forks
+          reap_and_replace_terminated_processes
           interruptible_sleep(1.second)
         end
       end
 
+
       def sync_std_streams
         STDOUT.sync = STDERR.sync = true
       end
 
       def setup_pidfile
-        @pidfile = if SolidQueue.supervisor_pidfile
-          Processes::Pidfile.new(SolidQueue.supervisor_pidfile).tap(&:setup)
+        if path = SolidQueue.supervisor_pidfile
+          @pidfile = Pidfile.new(path).tap(&:setup)
         end
       end
 
-      def start_forks
-        configured_processes.each { |configured_process| start_fork(configured_process) }
-      end
 
-      def launch_maintenance_task
-        @maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: SolidQueue.process_alive_threshold) do
-          prune_dead_processes
-          release_orphaned_executions
-        end
-        @maintenance_task.execute
+      def start_process(configured_process)
+        raise NotImplementedError
       end
 
+
       def shutdown
-        stop_process_prune
+        stop_maintenance_task
         restore_default_signal_handlers
         delete_pidfile
       end
 
       def graceful_termination
-        SolidQueue.instrument(:graceful_termination, supervisor_pid: ::Process.pid, supervised_pids: forks.keys) do |payload|
-          term_forks
+        SolidQueue.instrument(:graceful_termination, supervisor_pid: ::Process.pid, supervised_processes: processes.keys) do |payload|
+          term_processes
 
-          wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
-            reap_terminated_forks
+          wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? }) do
+            reap_terminated_processes
           end
 
-          unless all_forks_terminated?
+          unless all_processes_terminated?
             payload[:shutdown_timeout_exceeded] = true
             immediate_termination
           end
         end
       end
 
       def immediate_termination
-        SolidQueue.instrument(:immediate_termination, supervisor_pid: ::Process.pid, supervised_pids: forks.keys) do
-          quit_forks
+        SolidQueue.instrument(:immediate_termination, supervisor_pid: ::Process.pid, supervised_processes: processes.keys) do
+          quit_processes
         end
       end
 
-      def term_forks
-        signal_processes(forks.keys, :TERM)
-      end
-
-      def quit_forks
-        signal_processes(forks.keys, :QUIT)
-      end
-
-      def stop_process_prune
-        @maintenance_task&.shutdown
-      end
-
       def delete_pidfile
         @pidfile&.delete
       end
 
-      def prune_dead_processes
-        wrap_in_app_executor { SolidQueue::Process.prune }
-      end
-
-      def release_orphaned_executions
-        wrap_in_app_executor { SolidQueue::ClaimedExecution.orphaned.release_all }
-      end
-
-      def start_fork(configured_process)
-        configured_process.supervised_by process
-        configured_process.mode = :fork
-
-        pid = fork do
-          configured_process.start
-        end
-
-        forks[pid] = configured_process
-      end
-
-      def reap_and_replace_terminated_forks
-        loop do
-          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
-          break unless pid
-
-          replace_fork(pid, status)
-        end
-      end
-
-      def reap_terminated_forks
-        loop do
-          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
-          break unless pid
-
-          forks.delete(pid)
-        end
-      rescue SystemCallError
-        # All children already reaped
-      end
-
-      def replace_fork(pid, status)
-        SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
-          if supervised_fork = forks.delete(pid)
-            payload[:fork] = supervised_fork
-            start_fork(supervised_fork)
-          end
-        end
-      end
-
-      def all_forks_terminated?
-        forks.empty?
-      end
-
       def wait_until(timeout, condition, &block)
         if timeout > 0
           deadline = monotonic_time_now + timeout

lib/solid_queue/supervisor/forks.rb

@@ -0,0 +1,62 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Supervisor::Forks < Supervisor
+    def kind
+      "Supervisor(forks)"
+    end
+
+    private
+      def start_process(configured_process)
+        configured_process.supervised_by process
+        configured_process.mode = :fork
+
+        pid = fork do
+          configured_process.start
+        end
+
+        processes[pid] = configured_process
+      end
+
+      def term_processes
+        signal_processes(processes.keys, :TERM)
+      end
+
+      def quit_processes
+        signal_processes(processes.keys, :QUIT)
+      end
+
+      def reap_and_replace_terminated_processes
+        loop do
+          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
+          break unless pid
+
+          replace_fork(pid, status)
+        end
+      end
+
+      def reap_terminated_processes
+        loop do
+          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
+          break unless pid
+
+          processes.delete(pid)
+        end
+      rescue SystemCallError
+        # All children already reaped
+      end
+
+      def replace_fork(pid, status)
+        SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
+          if supervised_fork = processes.delete(pid)
+            payload[:fork] = supervised_fork
+            start_process(supervised_fork)
+          end
+        end
+      end
+
+      def all_processes_terminated?
+        processes.empty?
+      end
+  end
+end

lib/solid_queue/supervisor/maintenance.rb

@@ -0,0 +1,26 @@
+module SolidQueue
+  module Supervisor::Maintenance
+    extend ActiveSupport::Concern
+
+      private
+        def launch_maintenance_task
+          @maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: SolidQueue.process_alive_threshold) do
+            prune_dead_processes
+            release_orphaned_executions
+          end
+          @maintenance_task.execute
+        end
+
+        def stop_maintenance_task
+          @maintenance_task&.shutdown
+        end
+
+        def prune_dead_processes
+          wrap_in_app_executor { SolidQueue::Process.prune }
+        end
+
+        def release_orphaned_executions
+          wrap_in_app_executor { SolidQueue::ClaimedExecution.orphaned.release_all }
+        end
+    end
+end

lib/solid_queue/processes/pidfile.rb renamed to lib/solid_queue/supervisor/pidfile.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
-module SolidQueue::Processes
-  class Pidfile
+module SolidQueue
+  class Supervisor::Pidfile
     def initialize(path)
       @path = path
       @pid = ::Process.pid

lib/solid_queue/supervisor/signals.rb

@@ -6,8 +6,6 @@ class GracefulTerminationRequested < Interrupt; end
     class ImmediateTerminationRequested < Interrupt; end
 
     module Signals
-      extend ActiveSupport::Concern
-
       private
         SIGNALS = %i[ QUIT INT TERM ]
 

test/integration/processes_lifecycle_test.rb renamed to test/integration/forked_processes_lifecycle_test.rb

@@ -2,7 +2,7 @@
 
 require "test_helper"
 
-class ProcessLifecycleTest < ActiveSupport::TestCase
+class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
@@ -211,7 +211,7 @@ def assert_registered_workers_for(*queues)
     end
 
     def assert_registered_supervisor
-      processes = find_processes_registered_as("Supervisor")
+      processes = find_processes_registered_as("Supervisor(forks)")
       assert_equal 1, processes.count
       assert_equal @pid, processes.first.pid
     end