Add reactor thread workers

Appraisals

@@ -2,6 +2,7 @@
 
 appraise "rails-7-1" do
   gem "railties", "~> 7.1.0"
+  gem "json", "< 3"
 end
 
 appraise "rails-7-2" do
@@ -10,6 +11,7 @@ end
 
 appraise "rails-8-0" do
   gem "railties", "~> 8.0.0"
+  gem "json", "< 3"
 end
 
 appraise "rails-8-1" do

README.md

@@ -289,6 +289,7 @@ Here's an overview of the different options:
 
 - `threads`: configures a worker to execute jobs in a thread pool of this size. By default, workers use `threads: 3`. Only workers have this setting, and it can't be combined with `fibers`.
 It is recommended to set this value less than or equal to the queue database's connection pool size minus 2, as each worker uses connections for polling and heartbeat and thread mode may use additional connections for job execution.
+- `reactor`: when `true`, a thread worker runs each of its threads' jobs inside an [Async](https://github.com/socketry/async) reactor that lives as long as the thread, one job at a time per thread. See [reactor thread workers](#reactor-thread-workers). Requires the `async` gem, and can't be combined with `fibers`.
 - `fibers`: configures a worker to execute jobs as fibers on a single fiber reactor thread, with this value as the maximum number of in-flight jobs. It can't be combined with `threads`.
   Fiber workers require fiber-scoped isolated execution state. In Rails apps, set `config.active_support.isolation_level = :fiber` before using `fibers`. Solid Queue refuses to boot fiber workers when isolation remains thread-scoped. On Rails 7.2 and later, a practical starting point is usually `3-5` queue database connections per worker process rather than matching the `fibers` value, because ordinary Active Record query paths can release connections between non-blocking waits. On Rails 7.1, size the queue database pool more conservatively, as in-flight fiber jobs may still retain connections roughly in proportion to `fibers`.
 - `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. This works with both `threads` and `fibers` workers as long as the supervisor is running in the default `fork` mode. **Note**: this option is ignored only when the supervisor itself is [running in `async` mode](#fork-vs-async-mode).
@@ -412,6 +413,20 @@ This means finished and failed jobs still follow the normal Solid Queue lifecycl
 In a similar way, if a worker is terminated in any other way not initiated by the above signals (e.g. a worker is sent a `KILL` signal), jobs in progress will be marked as failed so that they can be inspected, with a `SolidQueue::Processes::ProcessExitError`. Sometimes a job in particular is responsible for this, for example, if it has a memory leak and you have a mechanism to kill processes over a certain memory threshold, so this will help identifying this kind of situation.
 
 
+#### Reactor thread workers
+
+Jobs on a thread worker run without a reactor, so a client built on Async, such as `async-http`, starts one for every call and pays for a new connection and TLS handshake each time. Fiber workers avoid that, but they run every job on one thread, so a job that blocks without yielding, such as a C extension decoding an image, stalls the others. Setting `reactor: true` on a thread worker keeps the thread pool and gives each thread a reactor that lives as long as the thread, so those clients reuse their connections across calls and across jobs:
+
+```yml
+workers:
+  - queues: [ api_sync, webhooks ]
+    threads: 5
+    reactor: true
+```
+
+Each thread still runs one job at a time, so database pool sizing is the same as for `threads`, and fiber-scoped isolated execution state isn't required.
+
+
 ### Database configuration
 
 You can configure the database used by Solid Queue via the `config.solid_queue.connects_to` option in the `config/application.rb` or `config/environments/production.rb` config files. By default, a single database is used for both writing and reading called `queue` to match the database configuration you set up during the install.

gemfiles/rails_7_1.gemfile

@@ -3,5 +3,6 @@
 source "https://rubygems.org"
 
 gem "railties", "~> 7.1.0"
+gem "json", "< 3"
 
 gemspec path: "../"

gemfiles/rails_8_0.gemfile

@@ -3,5 +3,6 @@
 source "https://rubygems.org"
 
 gem "railties", "~> 8.0.0"
+gem "json", "< 3"
 
 gemspec path: "../"

lib/solid_queue/configuration.rb

@@ -8,6 +8,7 @@ class Configuration
     validate :ensure_configured_processes, :ensure_valid_recurring_tasks
     validate :ensure_valid_worker_execution_options
     validate :ensure_fiber_workers_have_required_dependency, :ensure_fiber_workers_use_supported_isolation_level
+    validate :ensure_reactor_workers_have_required_dependency
     validate :warn_about_incorrectly_sized_database_pool, :warn_about_missing_config_files
 
     before_validation { warnings.clear }
@@ -130,6 +131,10 @@ def ensure_valid_worker_execution_options
           if options.key?(:threads) && options.key?(:fibers)
             errors.add(:base, "Workers can specify either `threads` or `fibers`, but not both.")
           end
+
+          if options[:reactor] && fiber_worker?(options)
+            errors.add(:base, "`reactor` applies to thread workers. Fiber workers already run their jobs on a reactor.")
+          end
         end
       end
 
@@ -143,6 +148,15 @@ def ensure_fiber_workers_have_required_dependency
           "Add `gem \"async\"` to your Gemfile to configure workers with `fibers`.")
       end
 
+      def ensure_reactor_workers_have_required_dependency
+        return unless workers_options.any? { |options| reactor_thread_worker?(options) }
+
+        require "async"
+      rescue LoadError
+        errors.add(:base, "Reactor thread workers require the `async` gem. " \
+          "Add `gem \"async\"` to your Gemfile to configure workers with `reactor: true`.")
+      end
+
       def ensure_fiber_workers_use_supported_isolation_level
         return unless workers_options.any? { |options| fiber_worker?(options) }
 
@@ -319,6 +333,10 @@ def fiber_worker?(options)
         options.key?(:fibers)
       end
 
+      def reactor_thread_worker?(options)
+        options[:reactor] && !fiber_worker?(options)
+      end
+
       def worker_defaults_for(options)
         if fiber_worker?(options)
           WORKER_DEFAULTS.except(:threads)

lib/solid_queue/reactor_thread_pool.rb

@@ -0,0 +1,113 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class ReactorThreadPool < Pool
+    def initialize(size, on_idle: nil, on_unrecoverable_error: nil)
+      super
+
+      @state_mutex = Mutex.new
+      @shutdown = false
+      @fatal_error = nil
+      @pending_executions = Thread::Queue.new
+      @reactor_threads = []
+    end
+
+    def post(execution)
+      raise_if_fatal_error!
+      raise RuntimeError, "Execution pool is shutting down" if shutdown?
+
+      super
+    end
+
+    def available_capacity
+      raise_if_fatal_error!
+      super
+    end
+
+    def shutdown
+      state_mutex.synchronize do
+        next false if @shutdown
+
+        @shutdown = true
+      end.tap do |shut_down|
+        # Each thread drains the executions already queued, then its blocked
+        # pop returns nil and its reactor exits
+        pending_executions.close if shut_down
+      end
+    end
+
+    def shutdown?
+      state_mutex.synchronize { @shutdown }
+    end
+
+    def wait_for_termination(timeout)
+      deadline = Timer.monotonic_time_now + timeout.to_f
+      threads = state_mutex.synchronize { reactor_threads.dup }
+
+      threads.all? { |thread| thread.join([ deadline - Timer.monotonic_time_now, 0 ].max) }
+    end
+
+    private
+      attr_reader :pending_executions, :reactor_threads, :state_mutex
+
+      def name
+        @name ||= "solid_queue-reactor-thread-pool-#{object_id}"
+      end
+
+      def schedule(execution)
+        start_reactor_threads_if_needed
+        pending_executions << execution
+      end
+
+      # Threads are started lazily, like the fiber pool's reactor, so the pool
+      # can be built in the supervisor before it forks.
+      def start_reactor_threads_if_needed
+        state_mutex.synchronize do
+          next if reactor_threads.any?
+
+          require "async"
+          size.times { reactor_threads << start_reactor_thread }
+        end
+      end
+
+      # Each thread runs its jobs one at a time on a reactor that lives as long
+      # as the thread, so async clients keep their connections across jobs
+      def start_reactor_thread
+        create_thread do
+          Sync do
+            while execution = pending_executions.pop
+              perform_execution(execution)
+            end
+          end
+        rescue Exception => error
+          register_fatal_error(error)
+          raise
+        end
+      end
+
+      def perform_execution(execution)
+        wrap_in_app_executor { execution.perform }
+      rescue Async::Stop => error
+        handle_thread_error(error)
+        register_fatal_error(error)
+      rescue Exception => error
+        handle_unrecoverable_error(error)
+        handle_thread_error(error)
+      ensure
+        restore_capacity
+      end
+
+      def register_fatal_error(error)
+        state_mutex.synchronize do
+          @fatal_error ||= error
+        end
+
+        on_idle&.call
+      end
+
+      def raise_if_fatal_error!
+        error = state_mutex.synchronize { @fatal_error }
+        raise error if error
+      end
+  end
+end

lib/solid_queue/worker.rb

@@ -11,7 +11,13 @@ class Worker < Processes::Poller
     attr_reader :queues, :pool
 
     def initialize(**options)
-      execution_pool_type = options.key?(:fibers) ? :fiber : :thread
+      execution_pool_type = if options.key?(:fibers)
+        :fiber
+      elsif options[:reactor]
+        :reactor_thread
+      else
+        :thread
+      end
 
       options = options.dup.with_defaults(SolidQueue::Configuration::WORKER_DEFAULTS)
       execution_pool_size = execution_pool_type == :fiber ? options[:fibers] : options[:threads]

test/unit/configuration_test.rb

@@ -91,6 +91,38 @@ class ConfigurationTest < ActiveSupport::TestCase
     end
   end
 
+  test "reactor configures thread workers to run each thread's jobs on a reactor" do
+    configuration = SolidQueue::Configuration.new(
+      workers: [ { queues: "api*", threads: 3, reactor: true } ],
+      dispatchers: [],
+      skip_recurring: true
+    )
+
+    assert configuration.valid?
+    assert_processes configuration, :worker, 1, threads: 3, reactor: true
+  end
+
+  test "workers reject reactor with fibers" do
+    with_execution_isolation(:fiber) do
+      configuration = SolidQueue::Configuration.new(
+        workers: [ { queues: "llm*", fibers: 10, reactor: true } ],
+        dispatchers: [],
+        skip_recurring: true
+      )
+
+      assert_not configuration.valid?
+      assert_match /`reactor` applies to thread workers/, configuration.errors.full_messages.first
+    end
+  end
+
+  test "reactor thread workers require the async gem" do
+    configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { threads: 3, reactor: true } ])
+    configuration.expects(:require).with("async").raises(LoadError.new("cannot load such file -- async"))
+
+    assert_not configuration.valid?
+    assert_match /Reactor thread workers require the `async` gem/, configuration.errors.full_messages.first
+  end
+
   test "fiber worker size inflates required database pool size on Rails 7.1" do
     skip if fiber_workers_release_connections_between_queries?
 

test/unit/reactor_thread_pool_test.rb

@@ -0,0 +1,116 @@
+require "test_helper"
+
+class ReactorThreadPoolTest < Minitest::Test
+  Execution = Struct.new(:started, :results, :pause) do
+    def perform
+      started << true if started
+      sleep(pause) if pause
+      results << [ Thread.current.object_id, Async::Task.current.reactor.object_id ] if results
+    end
+  end
+
+  CancelledExecution = Struct.new(:started) do
+    def perform
+      started << true if started
+      raise Async::Stop.new
+    end
+  end
+
+  def test_builds_a_reactor_thread_pool
+    pool = mock
+
+    SolidQueue::ReactorThreadPool.expects(:new).with(5, on_idle: nil, on_unrecoverable_error: nil).returns(pool)
+
+    assert_equal pool, SolidQueue::Pool.build(type: :reactor_thread, size: 5)
+  end
+
+  def test_executes_jobs_concurrently_on_separate_threads_each_inside_a_reactor
+    pool = SolidQueue::ReactorThreadPool.new(2)
+    started = Thread::Queue.new
+    results = Thread::Queue.new
+
+    pool.post Execution.new(started, results, 0.1)
+    pool.post Execution.new(started, results, 0.1)
+
+    2.times { Timeout.timeout(1.second) { started.pop } }
+    entries = 2.times.map { Timeout.timeout(1.second) { results.pop } }
+
+    assert_equal 2, entries.map(&:first).uniq.count
+    assert_equal 2, entries.map(&:last).uniq.count
+    assert_equal 2, pool.available_capacity
+  ensure
+    pool&.shutdown
+    pool&.wait_for_termination(1.second)
+  end
+
+  def test_keeps_the_same_reactor_across_jobs_on_a_thread
+    pool = SolidQueue::ReactorThreadPool.new(1)
+    results = Thread::Queue.new
+
+    pool.post Execution.new(nil, results, nil)
+    first = Timeout.timeout(1.second) { results.pop }
+    pool.post Execution.new(nil, results, nil)
+    second = Timeout.timeout(1.second) { results.pop }
+
+    assert_equal first, second
+  ensure
+    pool&.shutdown
+    pool&.wait_for_termination(1.second)
+  end
+
+  def test_waits_for_in_flight_executions_during_shutdown
+    pool = SolidQueue::ReactorThreadPool.new(1)
+    started = Thread::Queue.new
+
+    pool.post Execution.new(started, nil, 0.1)
+    Timeout.timeout(1.second) { started.pop }
+
+    pool.shutdown
+
+    assert_equal false, pool.wait_for_termination(0.01)
+    assert pool.wait_for_termination(1.second)
+  ensure
+    pool&.shutdown
+    pool&.wait_for_termination(1.second)
+  end
+
+  def test_starts_threads_lazily_so_the_pool_can_be_built_before_forking
+    pool = SolidQueue::ReactorThreadPool.new(1)
+
+    pid = fork do
+      results = Thread::Queue.new
+      pool.post Execution.new(nil, results, nil)
+      Timeout.timeout(1.second) { results.pop }
+      pool.shutdown
+      pool.wait_for_termination(1.second)
+      exit!(0)
+    end
+
+    _, status = Process.waitpid2(pid)
+    assert_equal 0, status.exitstatus
+  ensure
+    pool&.shutdown
+    pool&.wait_for_termination(1.second)
+  end
+
+  def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled
+    notifications = Thread::Queue.new
+    started = Thread::Queue.new
+    reported_errors = []
+    original_on_thread_error = SolidQueue.on_thread_error
+    SolidQueue.on_thread_error = ->(error) { reported_errors << error.class.name }
+
+    pool = SolidQueue::ReactorThreadPool.new(1, on_idle: -> { notifications << :changed })
+
+    pool.post CancelledExecution.new(started)
+    Timeout.timeout(1.second) { started.pop }
+    Timeout.timeout(1.second) { notifications.pop }
+
+    error = assert_raises(Async::Stop) { pool.available_capacity }
+    assert_equal [ error.class.name ], reported_errors
+  ensure
+    SolidQueue.on_thread_error = original_on_thread_error
+    pool&.shutdown
+    pool&.wait_for_termination(1.second)
+  end
+end