From 81a32a03fec965e38dd77fc46c20af1c7ccea6a6 Mon Sep 17 00:00:00 2001 From: Andrew Fader Date: Thu, 22 Oct 2026 23:23:36 -0400 Subject: [PATCH 1/2] Add reactor thread workers Thread workers run jobs without an Async reactor, so a client built on Async, like async-http, starts one per call and pays for a new connection and TLS handshake each time. Fiber workers keep a reactor, but run every job on one thread, where a job that blocks without yielding (a C extension decoding an image, CPU-heavy Ruby) stalls the rest. `reactor: true` on a thread worker keeps the thread pool and gives each thread a reactor that lives as long as the thread, running its jobs one at a time. Clients then reuse their connections across calls and jobs. Like the fiber pool, threads start lazily so the pool can be built before forking, and a cancelled reactor marks the pool as fatal. It requires the async gem but not fiber-scoped isolation, and database pool sizing is the same as for threads. Co-Authored-By: Claude Opus 5.5 --- README.md | 15 ++++ lib/solid_queue/configuration.rb | 18 ++++ lib/solid_queue/reactor_thread_pool.rb | 113 ++++++++++++++++++++++++ lib/solid_queue/worker.rb | 8 +- test/unit/configuration_test.rb | 32 +++++++ test/unit/reactor_thread_pool_test.rb | 116 +++++++++++++++++++++++++ test/unit/worker_test.rb | 4 + 7 files changed, 305 insertions(+), 1 deletion(-) create mode 100644 lib/solid_queue/reactor_thread_pool.rb create mode 100644 test/unit/reactor_thread_pool_test.rb diff --git a/README.md b/README.md index 4367c6b07..39e5d67c3 100644 --- a/README.md +++ b/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. diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb index 18f833ab4..c9d26c780 100644 --- a/lib/solid_queue/configuration.rb +++ b/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) diff --git a/lib/solid_queue/reactor_thread_pool.rb b/lib/solid_queue/reactor_thread_pool.rb new file mode 100644 index 000000000..a6426bf11 --- /dev/null +++ b/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 diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index 3947b05d2..ae36f8aa9 100644 --- a/lib/solid_queue/worker.rb +++ b/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] diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb index 2a969a93c..95a612112 100644 --- a/test/unit/configuration_test.rb +++ b/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? diff --git a/test/unit/reactor_thread_pool_test.rb b/test/unit/reactor_thread_pool_test.rb new file mode 100644 index 000000000..4d2863498 --- /dev/null +++ b/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 diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 77f41edf5..a74875a48 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -13,6 +13,10 @@ class WorkerTest < ActiveSupport::TestCase { name: "fiber", options: { fibers: 3 } + }, + { + name: "reactor_thread", + options: { threads: 3, reactor: true } } ].freeze From a549ba2ed59774362bf229312d7c936516190581 Mon Sep 17 00:00:00 2001 From: Andrew Fader Date: Sat, 24 Oct 2026 00:28:08 -0400 Subject: [PATCH 2/2] Pin json < 3 for Rails 7.1 and 8.0 appraisals --- Appraisals | 2 ++ gemfiles/rails_7_1.gemfile | 1 + gemfiles/rails_8_0.gemfile | 1 + 3 files changed, 4 insertions(+) diff --git a/Appraisals b/Appraisals index 303c1cc76..74f71ad7a 100644 --- a/Appraisals +++ b/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 diff --git a/gemfiles/rails_7_1.gemfile b/gemfiles/rails_7_1.gemfile index 036d98368..5fd12a49c 100644 --- a/gemfiles/rails_7_1.gemfile +++ b/gemfiles/rails_7_1.gemfile @@ -3,5 +3,6 @@ source "https://rubygems.org" gem "railties", "~> 7.1.0" +gem "json", "< 3" gemspec path: "../" diff --git a/gemfiles/rails_8_0.gemfile b/gemfiles/rails_8_0.gemfile index 28bd9f00e..297d3f85c 100644 --- a/gemfiles/rails_8_0.gemfile +++ b/gemfiles/rails_8_0.gemfile @@ -3,5 +3,6 @@ source "https://rubygems.org" gem "railties", "~> 8.0.0" +gem "json", "< 3" gemspec path: "../"