Refactor async mode and restore tests for it

The tests were removed in 4b3483a

This also restores support for "standalone" async mode, which means the
Solid Queue's supervisor running on its own process, via `bin/jobs`, but
having all other processes (workers, dispatchers, schedulers) run as
threads instead of processes. When Solid Queue is run via the Puma
plugin, it won't be considered standalone, and it won't need to handle
signals and act on them.

Author: rosa

lib/puma/plugin/solid_queue.rb

@@ -36,8 +36,8 @@ def start_forked(launcher)
           end
         end
 
-        launcher.events.on_stopped { stop_solid_queue }
-        launcher.events.on_restart { stop_solid_queue }
+        launcher.events.on_stopped { stop_solid_queue_fork }
+        launcher.events.on_restart { stop_solid_queue_fork }
       else
         launcher.events.after_booted do
           @solid_queue_pid = fork do
@@ -46,24 +46,38 @@ def start_forked(launcher)
           end
         end
 
-        launcher.events.after_stopped { stop_solid_queue }
-        launcher.events.before_restart { stop_solid_queue }
+        launcher.events.after_stopped { stop_solid_queue_fork }
+        launcher.events.before_restart { stop_solid_queue_fork }
       end
     end
 
     def start_async(launcher)
       if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
-        launcher.events.on_booted { @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async) }
-        launcher.events.on_stopped { solid_queue_supervisor&.stop }
-        launcher.events.on_restart { solid_queue_supervisor&.stop; @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async) }
+        launcher.events.on_booted do
+          @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async, standalone: false)
+        end
+
+        launcher.events.on_stopped { @solid_queue_supervisor&.stop }
+
+        launcher.events.on_restart do
+          solid_queue_supervisor&.stop
+          @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async, standalone: false)
+        end
       else
-        launcher.events.after_booted { @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async) }
-        launcher.events.after_stopped { solid_queue_supervisor&.stop }
-        launcher.events.before_restart { solid_queue_supervisor&.stop; @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async) }
+        launcher.events.after_booted do
+          @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async, standalone: false)
+        end
+
+        launcher.events.after_stopped { @solid_queue_supervisor&.stop }
+
+        launcher.events.before_restart do
+          solid_queue_supervisor&.stop
+          @solid_queue_supervisor = SolidQueue::Supervisor.start(mode: :async, standalone: false)
+        end
       end
     end
 
-    def stop_solid_queue
+    def stop_solid_queue_fork
       Process.waitpid(solid_queue_pid, Process::WNOHANG)
       log "Stopping Solid Queue..."
       Process.kill(:INT, solid_queue_pid) if solid_queue_pid
@@ -76,7 +90,7 @@ def monitor_puma
     end
 
     def monitor_solid_queue
-      monitor(:solid_queue_dead?, "Detected Solid Queue has gone away, stopping Puma...")
+      monitor(:solid_queue_fork_dead?, "Detected Solid Queue has gone away, stopping Puma...")
     end
 
     def monitor(process_dead, message)
@@ -90,7 +104,7 @@ def monitor(process_dead, message)
       end
     end
 
-    def solid_queue_dead?
+    def solid_queue_fork_dead?
       if solid_queue_started?
         Process.waitpid(solid_queue_pid, Process::WNOHANG)
       end

lib/solid_queue/async_supervisor.rb

@@ -3,96 +3,43 @@
 module SolidQueue
   class AsyncSupervisor < Supervisor
     private
-      attr_reader :threads
-
-      def start_processes
-        @threads = {}
-
-        configuration.configured_processes.each { |configured_process| start_process(configured_process) }
+      def check_and_replace_terminated_processes
+        terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
+        terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
       end
 
-      def start_process(configured_process)
-        process_instance = configured_process.instantiate.tap do |instance|
-          instance.supervised_by process
-          instance.mode = :async
-        end
+      def replace_thread(thread_id, instance)
+        SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
+          payload[:thread] = instance
+          handle_claimed_jobs_by(terminated_instance, thread)
 
-        thread = Thread.new do
-          begin
-            process_instance.start
-          rescue Exception => e
-            puts "Error in thread: #{e.message}"
-            puts e.backtrace
-          end
+          start_process(configured_processes.delete(thread_id))
         end
-        threads[thread] = [ process_instance, configured_process ]
       end
 
-      def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do |payload|
-          processes.each(&:stop)
-
-          Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_threads_terminated? }) do
-            # No-op, we just wait
-          end
-
-          unless all_threads_terminated?
-            payload[:shutdown_timeout_exceeded] = true
-            terminate_immediately
-          end
-        end
-      end
+      def perform_graceful_termination
+        process_instances.values.each(&:stop)
 
-      def terminate_immediately
-        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
-          threads.keys.each(&:kill)
-        end
+        Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
       end
 
-      def supervised_processes
-        processes.map(&:to_s)
+      def perform_immediate_termination
+        exit!
       end
 
-      def reap_and_replace_terminated_forks
-        # No-op in async mode, we'll check for dead threads in the supervise loop
+      def all_processes_terminated?
+        process_instances.values.none?(&:alive?)
       end
 
-      def all_threads_terminated?
-        threads.keys.all? { |thread| !thread.alive? }
-      end
-
-      def supervise
-        loop do
-          break if stopped?
-
-          set_procline
-          process_signal_queue
-
-          unless stopped?
-            check_and_replace_terminated_threads
-            interruptible_sleep(1.second)
+      # When a supervised thread terminates unexpectedly, mark all executions
+      # it had claimed as failed so they can be retried by another worker.
+      def handle_claimed_jobs_by(terminated_instance, thread)
+        wrap_in_app_executor do
+          if registered_process = SolidQueue::Process.find_by(name: terminated_instance.name)
+            error = Processes::ThreadTerminatedError.new(terminated_instance.name)
+            registered_process.fail_all_claimed_executions_with(error)
           end
         end
-      ensure
-        shutdown
-      end
-
-      def check_and_replace_terminated_threads
-        terminated_threads = {}
-        threads.each do |thread, (process, configured_process)|
-          unless thread.alive?
-            terminated_threads[thread] = configured_process
-          end
-        end
-
-        terminated_threads.each do |thread, configured_process|
-          threads.delete(thread)
-          start_process(configured_process)
-        end
-      end
-
-      def processes
-        threads.values.map(&:first)
       end
   end
-end
+end

lib/solid_queue/configuration.rb

@@ -31,11 +31,8 @@ def instantiate
     DEFAULT_CONFIG_FILE_PATH = "config/queue.yml"
     DEFAULT_RECURRING_SCHEDULE_FILE_PATH = "config/recurring.yml"
 
-    attr_reader :mode
-
     def initialize(**options)
       @options = options.with_defaults(default_options)
-      @mode = @options[:mode].to_s.inquiry
     end
 
     def configured_processes
@@ -59,6 +56,14 @@ def error_messages
       end
     end
 
+    def mode
+      @options[:mode].to_s.inquiry
+    end
+
+    def standalone?
+      mode.fork? || @options[:standalone]
+    end
+
     private
       attr_reader :options
 
@@ -88,6 +93,7 @@ def ensure_correctly_sized_thread_pool
       def default_options
         {
           mode: :fork,
+          standalone: true,
           config_file: Rails.root.join(ENV["SOLID_QUEUE_CONFIG"] || DEFAULT_CONFIG_FILE_PATH),
           recurring_schedule_file: Rails.root.join(ENV["SOLID_QUEUE_RECURRING_SCHEDULE"] || DEFAULT_RECURRING_SCHEDULE_FILE_PATH),
           only_work: false,

lib/solid_queue/fork_supervisor.rb

@@ -3,111 +3,76 @@
 module SolidQueue
   class ForkSupervisor < Supervisor
     private
-      attr_reader :forks, :configured_processes
 
-      def start_processes
-        @forks = {}
-        @configured_processes = {}
+    def perform_graceful_termination
+      term_forks
 
-        configuration.configured_processes.each { |configured_process| start_process(configured_process) }
+      Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? }) do
+        reap_terminated_forks
       end
+    end
 
-      def start_process(configured_process)
-        process_instance = configured_process.instantiate.tap do |instance|
-          instance.supervised_by process
-          instance.mode = :fork
-        end
-
-        pid = fork do
-          process_instance.start
-        end
-
-        configured_processes[pid] = configured_process
-        forks[pid] = process_instance
-      end
-
-      def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) 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 perform_immediate_termination
+      quit_forks
+    end
 
-      def terminate_immediately
-        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
-          quit_forks
-        end
-      end
+    def term_forks
+      signal_processes(process_instances.keys, :TERM)
+    end
 
-      def supervised_processes
-        forks.keys
-      end
+    def quit_forks
+      signal_processes(process_instances.keys, :QUIT)
+    end
 
-      def term_forks
-        signal_processes(forks.keys, :TERM)
-      end
+    def check_and_replace_terminated_processes
+      loop do
+        pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
+        break unless pid
 
-      def quit_forks
-        signal_processes(forks.keys, :QUIT)
+        replace_fork(pid, status)
       end
+    end
 
-      def reap_and_replace_terminated_forks
-        loop do
-          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
-          break unless pid
+    def reap_terminated_forks
+      loop do
+        pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
+        break unless pid
 
-          replace_fork(pid, status)
+        if (terminated_fork = process_instances.delete(pid)) && !status.exited? || status.exitstatus > 0
+          handle_claimed_jobs_by(terminated_fork, status)
         end
-      end
 
-      def reap_terminated_forks
-        loop do
-          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
-          break unless pid
-
-          if (terminated_fork = forks.delete(pid)) && (!status.exited? || status.exitstatus > 0)
-            handle_claimed_jobs_by(terminated_fork, status)
-          end
-
-          configured_processes.delete(pid)
-        end
-      rescue SystemCallError
-        # All children already reaped
+        configured_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 terminated_fork = forks.delete(pid)
-            payload[:fork] = terminated_fork
-            handle_claimed_jobs_by(terminated_fork, status)
+    def replace_fork(pid, status)
+      SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
+        if terminated_fork = process_instances.delete(pid)
+          payload[:fork] = terminated_fork
+          handle_claimed_jobs_by(terminated_fork, status)
 
-            start_process(configured_processes.delete(pid))
-          end
+          start_process(configured_processes.delete(pid))
         end
       end
-
-      # When a supervised fork crashes or exits we need to mark all the
-      # executions it had claimed as failed so that they can be retried
-      # by some other worker.
-      def handle_claimed_jobs_by(terminated_fork, status)
-        wrap_in_app_executor do
-          if registered_process = SolidQueue::Process.find_by(name: terminated_fork.name)
-            error = Processes::ProcessExitError.new(status)
-            registered_process.fail_all_claimed_executions_with(error)
-          end
+    end
+
+    # When a supervised fork crashes or exits we need to mark all the
+    # executions it had claimed as failed so that they can be retried
+    # by some other worker.
+    def handle_claimed_jobs_by(terminated_fork, status)
+      wrap_in_app_executor do
+        if registered_process = SolidQueue::Process.find_by(name: terminated_fork.name)
+          error = Processes::ProcessExitError.new(status)
+          registered_process.fail_all_claimed_executions_with(error)
         end
       end
+    end
 
-      def all_forks_terminated?
-        forks.empty?
-      end
+    def all_processes_terminated?
+      process_instances.empty?
+    end
   end
 end