Remove the new async mode

It's been short-lived ^_^U

Author: rosa

README.md

@@ -80,7 +80,7 @@ We have three types of actors in Solid Queue:
 - _Dispatchers_ are in charge of selecting jobs scheduled to run in the future that are due and _dispatching_ them, which is simply moving them from the `solid_queue_scheduled_executions` table over to the `solid_queue_ready_executions` table so that workers can pick them up. They're also in charge of managing [recurring tasks](#recurring-tasks), dispatching jobs to process them according to their schedule. On top of that, they do some maintenance work related to [concurrency controls](#concurrency-controls).
 - The _supervisor_ runs workers and dispatchers according to the configuration, controls their heartbeats, and stops and starts them when needed.
 
-By default, Solid Queue runs in `fork` mode. This means the supervisor will fork a separate process for each supervised worker/dispatcher. There's also an `async` mode where each worker and dispatcher will be run as a thread of the supervisor process. This can be used with [the provided Puma plugin](#puma-plugin)
+Solid Queue's supervisor will fork a separate process for each supervised worker/dispatcher.
 
 By default, Solid Queue will try to find your configuration under `config/solid_queue.yml`, but you can set a different path using the environment variable `SOLID_QUEUE_CONFIG`. This is what this configuration looks like:
 
@@ -131,7 +131,7 @@ Here's an overview of the different options:
 
   Finally, you can combine prefixes with exact names, like `[ staging*, background ]`, and the behaviour with respect to order will be the same as with only exact names.
 - `threads`: this is the max size of the thread pool that each worker will have to run jobs. Each worker will fetch this number of jobs from their queue(s), at most and will post them to the thread pool to be run. By default, this is `3`. Only workers have this setting.
-- `processes`: this is the number of worker processes that will be forked by the supervisor with the settings given. By default, this is `1`, just a single process. This setting is useful if you want to dedicate more than one CPU core to a queue or queues with the same configuration. Only workers have this setting. **Note**: this option will be ignored if [running in `async` mode](#running-as-a-fork-or-asynchronously).
+- `processes`: this is the number of worker processes that will be forked by the supervisor with the settings given. By default, this is `1`, just a single process. This setting is useful if you want to dedicate more than one CPU core to a queue or queues with the same configuration. Only workers have this setting.
 - `concurrency_maintenance`: whether the dispatcher will perform the concurrency maintenance work. This is `true` by default, and it's useful if you don't use any [concurrency controls](#concurrency-controls) and want to disable it or if you run multiple dispatchers and want some of them to just dispatch jobs without doing anything else.
 - `recurring_tasks`: a list of recurring tasks the dispatcher will manage. Read more details about this one in the [Recurring tasks](#recurring-tasks) section.
 
@@ -159,16 +159,6 @@ When receiving a `QUIT` signal, if workers still have jobs in-flight, these will
 If processes have no chance of cleaning up before exiting (e.g. if someone pulls a cable somewhere), in-flight jobs might remain claimed by the processes executing them. Processes send heartbeats, and the supervisor checks and prunes processes with expired heartbeats, which will release any claimed jobs back to their queues. You can configure both the frequency of heartbeats and the threshold to consider a process dead. See the section below for this.
 
 
-### Fork vs. async mode
-
-By default, the supervisor will fork additional processes for each worker and dispatcher so that they run in different processes. This provides the best isolation and performance, but can have additional memory usage. Alternatively, you can run all workers and dispatchers in the same process as the supervisor by passing `--mode async` when starting it:
-```
-bin/jobs --mode async
-```
-
-If you use the `async` mode, the `processes` option in the configuration will be ignored.
-
-
 ### Dedicated database configuration
 
 Solid Queue can be configured to run on a different database than the main application.
@@ -315,18 +305,6 @@ plugin :solid_queue
 ```
 to your `puma.rb` configuration.
 
-### Running as a fork or asynchronously
-
-By default, the Puma plugin will fork additional processes for each worker and dispatcher so that they run in different processes. This provides the best isolation and performance, but can have additional memory usage.
-
-Alternatively, workers and dispatchers can be run within the same Puma process(s). To do so just configure the plugin as:
-
-```ruby
-plugin :solid_queue
-solid_queue_mode :async
-```
-
-Note that in this case, the `processes` configuration option will be ignored.
 
 ## Jobs and transactional integrity
 :warning: Having your jobs in the same ACID-compliant database as your application data enables a powerful yet sharp tool: taking advantage of transactional integrity to ensure some action in your app is not committed unless your job is also committed. This can be very powerful and useful, but it can also backfire if you base some of your logic on this behaviour, and in the future, you move to another active job backend, or if you simply move Solid Queue to its own database, and suddenly the behaviour changes under you.

UPGRADING.md

@@ -1,3 +1,10 @@
+# Upgrading to version 0.7.x
+
+This version removed the new async mode introduced in version 0.4.0 and introduced a new binstub that can be used to start Solid Queue's supervisor. To install it, you can just run
+```
+bin/rails generate solid_queue:install
+```
+
 # Upgrading to version 0.6.x
 
 ## New migration in 3 steps
@@ -44,7 +51,7 @@ And then run the migrations.
 
 
 # Upgrading to version 0.4.x
-This version introduced an _async_ mode to run the supervisor and have all workers and dispatchers run as part of the same process as the supervisor, instead of separate, forked, processes. Together with this, we introduced some changes in how the supervisor is started. Prior this change, you could choose whether you wanted to run workers, dispatchers or both, by starting Solid Queue as `solid_queue:work` or `solid_queue:dispatch`. From version 0.4.0, the only option available is:
+This version introduced an _async_ mode (this mode has been removed in version 0.7.0) to run the supervisor and have all workers and dispatchers run as part of the same process as the supervisor, instead of separate, forked, processes. Together with this, we introduced some changes in how the supervisor is started. Prior this change, you could choose whether you wanted to run workers, dispatchers or both, by starting Solid Queue as `solid_queue:work` or `solid_queue:dispatch`. From version 0.4.0, the only option available is:
 
 ```
 $ bundle exec rake solid_queue:start

lib/puma/plugin/solid_queue.rb

@@ -1,50 +1,28 @@
 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 = $$
 
-    if launcher.options[:solid_queue_mode] == :async
-      start_async(launcher)
-    else
-      start_forked(launcher)
+    in_background do
+      monitor_solid_queue
     end
-  end
 
-  private
-    def start_forked(launcher)
-      in_background do
-        monitor_solid_queue
+    launcher.events.on_booted do
+      @solid_queue_pid = fork do
+        Thread.new { monitor_puma }
+        SolidQueue::Supervisor.start
       end
-
-      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 }
     end
 
-    def start_async(launcher)
-      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
+    launcher.events.on_stopped { stop_solid_queue }
+    launcher.events.on_restart { stop_solid_queue }
+  end
 
+  private
     def stop_solid_queue
       Process.waitpid(solid_queue_pid, Process::WNOHANG)
       log "Stopping Solid Queue..."

lib/solid_queue/cli.rb

@@ -5,7 +5,6 @@
 module SolidQueue
   class Cli < Thor
     class_option :config_file, type: :string, aliases: "-c", default: Configuration::DEFAULT_CONFIG_FILE_PATH, desc: "Path to config file"
-    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)"
 
     def self.exit_on_failure?
       true
@@ -15,7 +14,7 @@ def self.exit_on_failure?
     default_command :start
 
     def start
-      SolidQueue::Supervisor.start(mode: options["mode"], load_configuration_from: options["config_file"])
+      SolidQueue::Supervisor.start(load_configuration_from: options["config_file"])
     end
   end
 end

lib/solid_queue/configuration.rb

@@ -28,8 +28,7 @@ def instantiate
       dispatchers: [ DISPATCHER_DEFAULTS ]
     }
 
-    def initialize(mode: :fork, load_from: nil)
-      @mode = mode.to_s.inquiry
+    def initialize(load_from: nil)
       @raw_config = config_from(load_from)
     end
 
@@ -43,17 +42,13 @@ def max_number_of_threads
     end
 
     private
-      attr_reader :raw_config, :mode
+      attr_reader :raw_config
 
       DEFAULT_CONFIG_FILE_PATH = "config/solid_queue.yml"
 
       def workers
         workers_options.flat_map do |worker_options|
-          processes = if mode.fork?
-            worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
-          else
-            WORKER_DEFAULTS[:processes]
-          end
+          processes = worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
           processes.times.map { Process.new(:worker, worker_options.with_defaults(WORKER_DEFAULTS)) }
         end
       end

lib/solid_queue/supervisor.rb

@@ -5,21 +5,23 @@ class Supervisor < Processes::Base
     include Maintenance, Signals, Pidfiled
 
     class << self
-      def start(mode: "fork", load_configuration_from: nil, **options)
+      def start(load_configuration_from: nil)
         SolidQueue.supervisor = true
-        configuration = Configuration.new(mode: mode, load_from: load_configuration_from)
+        configuration = Configuration.new(load_from: load_configuration_from)
 
         if configuration.configured_processes.any?
-          klass = mode.to_s.inquiry.fork? ? ForkSupervisor : AsyncSupervisor
-          klass.new(configuration, **options).tap(&:start)
+          new(configuration).tap(&:start)
         else
           abort "No workers or processed configured. Exiting..."
         end
       end
     end
 
-    def initialize(configuration, **options)
+    def initialize(configuration)
       @configuration = configuration
+      @forks = {}
+      @configured_processes = {}
+
       super
     end
 
@@ -37,7 +39,7 @@ def stop
     end
 
     private
-      attr_reader :configuration
+      attr_reader :configuration, :forks, :configured_processes
 
       def boot
         SolidQueue.instrument(:start_process, process: self) do
@@ -52,6 +54,36 @@ def start_processes
         configuration.configured_processes.each { |configured_process| start_process(configured_process) }
       end
 
+      def supervise
+        loop do
+          break if stopped?
+
+          set_procline
+          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
+          instance.mode = :fork
+        end
+
+        pid = fork do
+          process_instance.start
+        end
+
+        configured_processes[pid] = configured_process
+        forks[pid] = process_instance
+      end
+
       def stopped?
         @stopped
       end
@@ -60,15 +92,15 @@ def set_procline
         procline "supervising #{supervised_processes.join(", ")}"
       end
 
-      def start_process(configured_process)
-        raise NotImplementedError
-      end
-
       def terminate_gracefully
         SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do |payload|
-          perform_graceful_termination
+          term_forks
+
+          Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
+            reap_terminated_forks
+          end
 
-          unless all_processes_terminated?
+          unless all_forks_terminated?
             payload[:shutdown_timeout_exceeded] = true
             terminate_immediately
           end
@@ -77,36 +109,78 @@ def terminate_gracefully
 
       def terminate_immediately
         SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
-          perform_immediate_termination
+          quit_forks
         end
       end
 
-      def perform_graceful_termination
-        raise NotImplementedError
+      def shutdown
+        SolidQueue.instrument(:shutdown_process, process: self) do
+          run_callbacks(:shutdown) do
+            stop_maintenance_task
+          end
+        end
       end
 
-      def perform_immediate_termination
-        raise NotImplementedError
+      def sync_std_streams
+        STDOUT.sync = STDERR.sync = true
       end
 
       def supervised_processes
-        raise NotImplementedError
+        forks.keys
       end
 
-      def all_processes_terminated?
-        raise NotImplementedError
+      def term_forks
+        signal_processes(forks.keys, :TERM)
       end
 
-      def shutdown
-        SolidQueue.instrument(:shutdown_process, process: self) do
-          run_callbacks(:shutdown) do
-            stop_maintenance_task
+      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 sync_std_streams
-        STDOUT.sync = STDERR.sync = true
+      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
+
+      def handle_claimed_jobs_by(terminated_fork, status)
+        if registered_process = process.supervisees.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