Extract rest of forking behaviour from Supervisor to another subclass

Author: rosa

lib/solid_queue/supervisor.rb

@@ -2,7 +2,10 @@
 
 module SolidQueue
   class Supervisor < Processes::Base
-    include Signals, Maintenance
+    include Maintenance
+
+    class GracefulTerminationRequested < Interrupt; end
+    class ImmediateTerminationRequested < Interrupt; end
 
     class << self
       def start(mode: :fork, load_configuration_from: nil)
@@ -15,7 +18,6 @@ def start(mode: :fork, load_configuration_from: nil)
 
     def initialize(configuration)
       @configuration = configuration
-      @processes = {}
     end
 
     def start
@@ -28,80 +30,42 @@ def start
     rescue GracefulTerminationRequested
       terminate_gracefully
     rescue ImmediateTerminationRequested
-      immediate_termination
+      terminate_immediately
     ensure
       run_callbacks(:shutdown) { shutdown }
     end
 
     private
-      attr_reader :configuration, :processes
+      attr_reader :configuration
 
       def boot
         sync_std_streams
-        setup_pidfile
-        register_signal_handlers
       end
 
       def start_processes
         configuration.processes.each { |configured_process| start_process(configured_process) }
       end
 
       def supervise
-        loop do
-          procline "supervising #{processes.keys.join(", ")}"
-
-          process_signal_queue
-          reap_and_replace_terminated_processes
-          interruptible_sleep(1.second)
-        end
-      end
-
-
-      def sync_std_streams
-        STDOUT.sync = STDERR.sync = true
-      end
-
-      def setup_pidfile
-        if path = SolidQueue.supervisor_pidfile
-          @pidfile = Pidfile.new(path).tap(&:setup)
-        end
+        raise NotImplementedError
       end
 
-
       def start_process(configured_process)
         raise NotImplementedError
       end
 
-
-      def shutdown
-        stop_maintenance_task
-        restore_default_signal_handlers
-        delete_pidfile
+      def terminate_gracefully
       end
 
-      def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, supervisor_pid: ::Process.pid, supervised_processes: processes.keys) do |payload|
-          term_processes
-
-          Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? }) do
-            reap_terminated_processes
-          end
-
-          unless all_processes_terminated?
-            payload[:shutdown_timeout_exceeded] = true
-            immediate_termination
-          end
-        end
+      def terminate_immediately
       end
 
-      def immediate_termination
-        SolidQueue.instrument(:immediate_termination, supervisor_pid: ::Process.pid, supervised_processes: processes.keys) do
-          quit_processes
-        end
+      def shutdown
+        stop_maintenance_task
       end
 
-      def delete_pidfile
-        @pidfile&.delete
+      def sync_std_streams
+        STDOUT.sync = STDERR.sync = true
       end
   end
 end

lib/solid_queue/supervisor/forks.rb

@@ -2,11 +2,33 @@
 
 module SolidQueue
   class Supervisor::Forks < Supervisor
+    include Signals
+
+    before_boot :setup_pidfile
+    after_shutdown :delete_pidfile
+
+    def initialize(*)
+      super
+      @forks = {}
+    end
+
     def kind
       "Supervisor(fork)"
     end
 
     private
+      attr_reader :forks
+
+      def supervise
+        loop do
+          procline "supervising #{forks.keys.join(", ")}"
+
+          process_signal_queue
+          reap_and_replace_terminated_forks
+          interruptible_sleep(1.second)
+        end
+      end
+
       def start_process(configured_process)
         configured_process.supervised_by process
         configured_process.mode = :fork
@@ -15,18 +37,39 @@ def start_process(configured_process)
           configured_process.start
         end
 
-        processes[pid] = configured_process
+        forks[pid] = configured_process
       end
 
-      def term_processes
-        signal_processes(processes.keys, :TERM)
+      def terminate_gracefully
+        SolidQueue.instrument(:graceful_termination, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do |payload|
+          term_forks
+
+          Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
+            reap_terminated_forks
+          end
+
+          unless all_forks_terminated?
+            payload[:shutdown_timeout_exceeded] = true
+            terminate_immediately
+          end
+        end
+      end
+
+      def terminate_immediately
+        SolidQueue.instrument(:immediate_termination, supervisor_pid: ::Process.pid, supervised_processes: forks.keys) do
+          quit_forks
+        end
+      end
+
+      def term_forks
+        signal_processes(forks.keys, :TERM)
       end
 
-      def quit_processes
-        signal_processes(processes.keys, :QUIT)
+      def quit_forks
+        signal_processes(forks.keys, :QUIT)
       end
 
-      def reap_and_replace_terminated_processes
+      def reap_and_replace_terminated_forks
         loop do
           pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
           break unless pid
@@ -35,28 +78,38 @@ def reap_and_replace_terminated_processes
         end
       end
 
-      def reap_terminated_processes
+      def reap_terminated_forks
         loop do
           pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
           break unless pid
 
-          processes.delete(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 = processes.delete(pid)
+          if supervised_fork = forks.delete(pid)
             payload[:fork] = supervised_fork
             start_process(supervised_fork)
           end
         end
       end
 
-      def all_processes_terminated?
-        processes.empty?
+      def setup_pidfile
+        if path = SolidQueue.supervisor_pidfile
+          @pidfile = Pidfile.new(path).tap(&:setup)
+        end
+      end
+
+      def delete_pidfile
+        @pidfile&.delete
+      end
+
+      def all_forks_terminated?
+        forks.empty?
       end
   end
 end

lib/solid_queue/supervisor/maintenance.rb

@@ -2,26 +2,27 @@ module SolidQueue
   module Supervisor::Maintenance
     extend ActiveSupport::Concern
 
-    private
+    included do
+      after_boot :release_orphaned_executions
+    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
+    private
+      def launch_maintenance_task
+        @maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: SolidQueue.process_alive_threshold) do
+          prune_dead_processes
+        end.tap(&:execute)
       end
-      @maintenance_task.execute
-    end
 
-    def stop_maintenance_task
-      @maintenance_task&.shutdown
-    end
+      def stop_maintenance_task
+        @maintenance_task&.shutdown
+      end
 
-    def prune_dead_processes
-      wrap_in_app_executor { SolidQueue::Process.prune }
-    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 release_orphaned_executions
+        wrap_in_app_executor { SolidQueue::ClaimedExecution.orphaned.release_all }
+      end
   end
 end

lib/solid_queue/supervisor/signals.rb

@@ -2,10 +2,14 @@
 
 module SolidQueue
   class Supervisor
-    class GracefulTerminationRequested < Interrupt; end
-    class ImmediateTerminationRequested < Interrupt; end
-
     module Signals
+      extend ActiveSupport::Concern
+
+      included do
+        before_boot :register_signal_handlers
+        after_shutdown :restore_default_signal_handlers
+      end
+
       private
         SIGNALS = %i[ QUIT INT TERM ]
 

test/integration/concurrency_controls_test.rb

@@ -169,15 +169,15 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "don't block claimed executions that get released" do
-    SequentialUpdateResultJob.perform_later(@result, name: "I'll be released to ready", pause: SolidQueue.shutdown_timeout + 3.seconds)
+    SequentialUpdateResultJob.perform_later(@result, name: "I'll be released to ready", pause: SolidQueue.shutdown_timeout + 10.seconds)
     job = SolidQueue::Job.last
 
     sleep(0.2)
     assert job.claimed?
 
     # This won't leave time to the job to finish
     signal_process(@pid, :TERM, wait: 0.1.second)
-    sleep(SolidQueue.shutdown_timeout + 0.2.seconds)
+    sleep(SolidQueue.shutdown_timeout + 0.6.seconds)
 
     assert_not job.reload.finished?
     assert job.reload.ready?