Extract more termination logic to Supervisor class

And rename sidecar to standalone, inverting the meaning.

Author: rosa

lib/puma/plugin/solid_queue.rb

@@ -40,7 +40,7 @@ def start_forked(launcher)
     end
 
     def start_async(launcher)
-      launcher.events.on_booted { @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async, sidecar: true) }
+      launcher.events.on_booted { @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async, standalone: false) }
       launcher.events.on_stopped { solid_queue_supervisor.stop }
       launcher.events.on_restart { solid_queue_supervisor.stop; solid_queue_supervisor.start }
     end

lib/solid_queue/dispatcher.rb

@@ -55,7 +55,7 @@ def all_work_completed?
       end
 
       def set_procline
-        procline "waiting"
+        procline "dispatching every #{polling_interval.seconds} seconds"
       end
   end
 end

lib/solid_queue/supervisor.rb

@@ -56,15 +56,45 @@ def stopped?
         @stopped
       end
 
-      def supervise
+      def set_procline
+        procline "supervising #{supervised_processes.join(", ")}"
       end
 
       def start_process(configured_process)
         raise NotImplementedError
       end
 
-      def instrument_termination(type, &block)
-        SolidQueue.instrument("#{type}_termination".to_sym, process_id: process_id, supervisor_pid: ::Process.pid, &block)
+      def terminate_gracefully
+        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do |payload|
+          perform_graceful_termination
+
+          unless all_processes_terminated?
+            payload[:shutdown_timeout_exceeded] = true
+            terminate_immediately
+          end
+        end
+      end
+
+      def terminate_immediately
+        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
+          perform_immediate_termination
+        end
+      end
+
+      def perform_graceful_termination
+        raise NotImplementedError
+      end
+
+      def perform_immediate_termination
+        raise NotImplementedError
+      end
+
+      def supervised_processes
+        raise NotImplementedError
+      end
+
+      def all_processes_terminated?
+        raise NotImplementedError
       end
 
       def shutdown

lib/solid_queue/supervisor/async_supervisor.rb

@@ -2,12 +2,12 @@
 
 module SolidQueue
   class Supervisor::AsyncSupervisor < Supervisor
-    skip_callback :boot, :before, :register_signal_handlers, if: :sidecar?
+    skip_callback :boot, :before, :register_signal_handlers, unless: :standalone?
 
-    def initialize(*, sidecar: false)
+    def initialize(*, standalone: true)
       super
 
-      @sidecar = sidecar
+      @standalone = standalone
       @threads = Concurrent::Map.new
     end
 
@@ -26,8 +26,8 @@ def stop
     private
       attr_reader :threads
 
-      def sidecar?
-        @sidecar
+      def standalone?
+        @standalone
       end
 
       def start_process(configured_process)
@@ -36,16 +36,19 @@ def start_process(configured_process)
         end
 
         process_instance.start
-
         threads[process_instance.name] = process_instance
       end
 
+      def supervised_processes
+        threads.keys
+      end
+
       def supervise
-        unless sidecar?
+        if standalone?
           loop do
             break if stopped?
 
-            procline "supervising #{threads.keys.join(", ")}"
+            set_procline
             process_signal_queue
 
             interruptible_sleep(10.second) unless stopped?
@@ -61,26 +64,15 @@ def stop_threads
         stop_threads.each { |thr| thr.join(SolidQueue.shutdown_timeout) }
       end
 
-      def terminate_gracefully
-        instrument_termination(:graceful) do |payload|
-          payload[:supervised_processes] = threads.keys
-
-          unless all_threads_terminated?
-            payload[:shutdown_timeout_exceeded] = true
-            terminate_immediately
-          end
-        end
+      def perform_graceful_termination
+        # All done when stopping
       end
 
-      def terminate_immediately
-        instrument_termination(:immediate) do |payload|
-          payload[:supervised_processes] = threads.keys
-
-          exit!
-        end
+      def perform_immediate_termination
+        exit!
       end
 
-      def all_threads_terminated?
+      def all_processes_terminated?
         threads.values.none?(&:alive?)
       end
   end

lib/solid_queue/supervisor/fork_supervisor.rb

@@ -16,22 +16,6 @@ def kind
     private
       attr_reader :forks, :configured_processes
 
-      def supervise
-        loop do
-          break if stopped?
-
-          procline "supervising #{forks.keys.join(", ")}"
-          process_signal_queue
-
-          unless stopped?
-            reap_and_replace_terminated_forks
-            interruptible_sleep(1.second)
-          end
-        end
-      ensure
-        shutdown
-      end
-
       def start_process(configured_process)
         process_instance = configured_process.instantiate.tap do |instance|
           instance.supervised_by process
@@ -46,31 +30,38 @@ def start_process(configured_process)
         forks[pid] = process_instance
       end
 
-      def terminate_gracefully
-        instrument_termination(:graceful) do |payload|
-          payload[:supervised_processes] = forks.keys
+      def supervised_processes
+        forks.keys
+      end
 
-          term_forks
+      def supervise
+        loop do
+          break if stopped?
 
-          Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
-            reap_terminated_forks
-          end
+          set_procline
+          process_signal_queue
 
-          unless all_forks_terminated?
-            payload[:shutdown_timeout_exceeded] = true
-            terminate_immediately
+          unless stopped?
+            reap_and_replace_terminated_forks
+            interruptible_sleep(1.second)
           end
         end
+      ensure
+        shutdown
       end
 
-      def terminate_immediately
-        instrument_termination(:immediate) do |payload|
-          payload[:supervised_processes] = forks.keys
+      def perform_graceful_termination
+        term_forks
 
-          quit_forks
+        Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? }) do
+          reap_terminated_forks
         end
       end
 
+      def perform_immediate_termination
+        quit_forks
+      end
+
       def term_forks
         signal_processes(forks.keys, :TERM)
       end
@@ -121,7 +112,7 @@ def handle_claimed_jobs_by(terminated_fork, status)
         end
       end
 
-      def all_forks_terminated?
+      def all_processes_terminated?
         forks.empty?
       end
   end

test/unit/async_supervisor_test.rb

@@ -3,8 +3,8 @@
 class AsyncSupervisorTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
-  test "start as sidecar" do
-    supervisor = run_supervisor_as_sidecar
+  test "start as non-standalone" do
+    supervisor = run_non_standalone_supervisor
     wait_for_registered_processes(4)
 
     assert_registered_processes(kind: "Supervisor(async)")
@@ -16,7 +16,7 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
     assert_no_registered_processes
   end
 
-  test "start as separate process" do
+  test "start standalone" do
     pid = run_supervisor_as_fork(mode: :async)
     wait_for_registered_processes(4)
 
@@ -28,9 +28,9 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
     assert_no_registered_processes
   end
 
-  test "start as sidecar with provided configuration" do
+  test "start as non-standalone with provided configuration" do
     config_as_hash = { workers: [], dispatchers: [ { batch_size: 100 } ] }
-    supervisor = run_supervisor_as_sidecar(load_configuration_from: config_as_hash)
+    supervisor = run_non_standalone_supervisor(load_configuration_from: config_as_hash)
     wait_for_registered_processes(2) # supervisor + dispatcher
 
     assert_registered_processes(kind: "Supervisor(async)")
@@ -61,7 +61,7 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
       dispatchers: []
     }
 
-    supervisor = run_supervisor_as_sidecar(load_configuration_from: config_as_hash)
+    supervisor = run_non_standalone_supervisor(load_configuration_from: config_as_hash)
     wait_for_registered_processes(3)
     assert_registered_processes(kind: "Supervisor(async)")
 
@@ -72,7 +72,7 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
   end
 
   private
-    def run_supervisor_as_sidecar(**kwargs)
-      SolidQueue::Supervisor.start(mode: :async, sidecar: true, **kwargs)
+    def run_non_standalone_supervisor(**kwargs)
+      SolidQueue::Supervisor.start(mode: :async, standalone: false, **kwargs)
     end
 end