Re-institute async mode

lib/puma/plugin/solid_queue.rb

@@ -1,40 +1,68 @@
 require "puma/plugin"
 
+module Puma
+ class DSL
+ def solid_queue_mode(mode = :fork)
+ @options[:solid_queue_mode] = mode.to_sym
+ end
+ end
+end
+
 Puma::Plugin.create do
  attr_reader :puma_pid, :solid_queue_pid, :log_writer, :solid_queue_supervisor
 
  def start(launcher)
  @log_writer = launcher.log_writer
  @puma_pid = $$
 
- in_background do
- monitor_solid_queue
+ if launcher.options[:solid_queue_mode] == :async
+ start_async(launcher)
+ else
+ start_forked(launcher)
  end
+ end
 
- if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
- launcher.events.on_booted do
- @solid_queue_pid = fork do
- Thread.new { monitor_puma }
- SolidQueue::Supervisor.start
- end
+ private
+ def start_forked(launcher)
+ in_background do
+ monitor_solid_queue
  end
 
- launcher.events.on_stopped { stop_solid_queue }
- launcher.events.on_restart { stop_solid_queue }
- else
- launcher.events.after_booted do
- @solid_queue_pid = fork do
- Thread.new { monitor_puma }
- SolidQueue::Supervisor.start
+ if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
+ launcher.events.on_booted do
+ @solid_queue_pid = fork do
+ Thread.new { monitor_puma }
+ SolidQueue::Supervisor.start(mode: :fork)
+ end
+ end
+
+ launcher.events.on_stopped { stop_solid_queue }
+ launcher.events.on_restart { stop_solid_queue }
+ else
+ launcher.events.after_booted do
+ @solid_queue_pid = fork do
+ Thread.new { monitor_puma }
+ SolidQueue::Supervisor.start(mode: :fork)
+ end
  end
+
+ launcher.events.after_stopped { stop_solid_queue }
+ launcher.events.before_restart { stop_solid_queue }
  end
+ end
 
- launcher.events.after_stopped { stop_solid_queue }
- launcher.events.before_restart { stop_solid_queue }
+ 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) }
+ 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) }
+ end
  end
- end
 
- private
  def stop_solid_queue
  Process.waitpid(solid_queue_pid, Process::WNOHANG)
  log "Stopping Solid Queue..."

lib/solid_queue/async_supervisor.rb

@@ -0,0 +1,98 @@
+# frozen_string_literal: true
+
+module SolidQueue
+ class AsyncSupervisor < Supervisor
+ private
+ attr_reader :threads
+
+ def start_processes
+ @threads = {}
+
+ configuration.configured_processes.each { |configured_process| start_process(configured_process) }
+ end
+
+ def start_process(configured_process)
+ process_instance = configured_process.instantiate.tap do |instance|
+ instance.supervised_by process
+ instance.mode = :async
+ end
+
+ thread = Thread.new do
+ begin
+ process_instance.start
+ rescue Exception => e
+ puts "Error in thread: #{e.message}"
+ puts e.backtrace
+ end
+ 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 terminate_immediately
+ SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
+ threads.keys.each(&:kill)
+ end
+ end
+
+ def supervised_processes
+ processes.map(&:to_s)
+ end
+
+ def reap_and_replace_terminated_forks
+ # No-op in async mode, we'll check for dead threads in the supervise loop
+ 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)
+ 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

lib/solid_queue/cli.rb

@@ -8,6 +8,8 @@ class Cli < Thor
  desc: "Path to config file (default: #{Configuration::DEFAULT_CONFIG_FILE_PATH}).",
  banner: "SOLID_QUEUE_CONFIG"
 
+ class_option :mode, type: :string, default: "fork", enum: %w[ fork async ], desc: "Whether to fork processes for workers and dispatchers (fork) or to run these in the same process as the supervisor (async)"
+
  class_option :recurring_schedule_file, type: :string,
  desc: "Path to recurring schedule definition (default: #{Configuration::DEFAULT_RECURRING_SCHEDULE_FILE_PATH}).",
  banner: "SOLID_QUEUE_RECURRING_SCHEDULE"

lib/solid_queue/configuration.rb

@@ -31,8 +31,11 @@ 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
@@ -84,6 +87,7 @@ def ensure_correctly_sized_thread_pool
 
  def default_options
  {
+ mode: :fork,
  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,
@@ -110,7 +114,12 @@ def skip_recurring_tasks?
 
  def workers
  workers_options.flat_map do |worker_options|
- processes = worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
+ processes = if mode.fork?
+ worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
+ else
+ 1
+ end
+
  processes.times.map { Process.new(:worker, worker_options.with_defaults(WORKER_DEFAULTS)) }
  end
  end

lib/solid_queue/fork_supervisor.rb

@@ -0,0 +1,111 @@
+# frozen_string_literal: true
+
+module SolidQueue
+ class ForkSupervisor < Supervisor
+ private
+ attr_reader :forks, :configured_processes
+
+ def start_processes
+ @forks = {}
+ @configured_processes = {}
+
+ configuration.configured_processes.each { |configured_process| start_process(configured_process) }
+ 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 terminate_immediately
+ SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
+ quit_forks
+ end
+ end
+
+ def supervised_processes
+ forks.keys
+ end
+
+ def term_forks
+ signal_processes(forks.keys, :TERM)
+ end
+
+ def quit_forks
+ signal_processes(forks.keys, :QUIT)
+ 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
+
+ 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
+ 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)
+
+ start_process(configured_processes.delete(pid))
+ end
+ 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)
+ 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
+
+ def all_forks_terminated?
+ forks.empty?
+ end
+ end
+end

lib/solid_queue/processes/runnable.rb

@@ -9,11 +9,7 @@ module Runnable
  def start
  boot
 
- if running_async?
- @thread = create_thread { run }
- else
- run
- end
+ run
  end
 
  def stop