From f6a27fc0e85124e95e50a9765a0a652f48dbe2a0 Mon Sep 17 00:00:00 2001 From: Carmine Paolino Date: Wed, 22 Jul 2026 20:32:26 +0200 Subject: [PATCH 01/13] Add fiber execution mode for workers Workers can now be configured with `fibers: N` as an alternative to `threads: N` (specifying both is invalid). In fiber mode, claimed jobs execute as fibers on a single Async reactor thread, bounded by the configured fiber count. This suits mostly I/O-bound jobs, where cooperative scheduling allows much higher in-flight capacity with fewer database connections than a thread pool. The existing pool was extracted into ExecutionPools::ThreadPool (SolidQueue::Pool remains as an alias), with the new ExecutionPools::FiberPool sharing the same interface. Pools are built on worker boot, after forking, since a reactor thread wouldn't survive a fork. Fiber mode requires the async gem (>= 2.24), which is not a runtime dependency: workers configured with `fibers` raise a clear error if it isn't available. It also requires fiber-scoped isolated execution state (ActiveSupport::IsolatedExecutionState.isolation_level = :fiber), validated at boot and configuration-check time. The database pool size guidance is fiber-aware and Rails-version aware: on Rails 7.2+, fibers release connections between queries, so a fiber worker needs far fewer connections than its fiber count. --- Gemfile.lock | 18 ++ README.md | 34 ++- lib/solid_queue/configuration.rb | 80 +++++- lib/solid_queue/execution_pools.rb | 18 ++ lib/solid_queue/execution_pools/fiber_pool.rb | 239 ++++++++++++++++++ .../execution_pools/thread_pool.rb | 82 ++++++ lib/solid_queue/pool.rb | 51 +--- lib/solid_queue/worker.rb | 67 ++++- solid_queue.gemspec | 1 + .../async_processes_lifecycle_test.rb | 15 +- test/integration/instrumentation_test.rb | 2 +- test/test_helper.rb | 13 + test/unit/async_supervisor_test.rb | 2 +- test/unit/configuration_test.rb | 85 ++++++- test/unit/execution_pools/fiber_pool_test.rb | 135 ++++++++++ test/unit/worker_test.rb | 194 +++++++++++--- 16 files changed, 921 insertions(+), 115 deletions(-) create mode 100644 lib/solid_queue/execution_pools.rb create mode 100644 lib/solid_queue/execution_pools/fiber_pool.rb create mode 100644 lib/solid_queue/execution_pools/thread_pool.rb create mode 100644 test/unit/execution_pools/fiber_pool_test.rb diff --git a/Gemfile.lock b/Gemfile.lock index 7ef8b8519..8e6c625df 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -55,12 +55,22 @@ GEM rake thor (>= 0.14.0) ast (2.4.2) + async (2.38.1) + console (~> 1.29) + fiber-annotation + io-event (~> 1.11) + metrics (~> 0.12) + traces (~> 0.18) base64 (0.3.0) benchmark (0.4.1) bigdecimal (3.3.1) builder (3.3.0) concurrent-ruby (1.3.5) connection_pool (2.5.4) + console (1.34.3) + fiber-annotation + fiber-local (~> 1.1) + json crass (1.0.6) date (3.4.1) debug (1.9.2) @@ -70,6 +80,10 @@ GEM erubi (1.13.1) et-orbi (1.2.11) tzinfo + fiber-annotation (0.2.0) + fiber-local (1.1.0) + fiber-storage + fiber-storage (1.0.1) fugit (1.11.1) et-orbi (~> 1, >= 1.2.11) raabro (~> 1.4) @@ -78,6 +92,7 @@ GEM i18n (1.14.7) concurrent-ruby (~> 1.0) io-console (0.8.0) + io-event (1.14.5) irb (1.14.3) rdoc (>= 4.0.0) reline (>= 0.4.2) @@ -87,6 +102,7 @@ GEM loofah (2.23.1) crass (~> 1.0.2) nokogiri (>= 1.12.0) + metrics (0.15.0) minitest (5.26.0) mocha (2.1.0) ruby2_keywords (>= 0.0.5) @@ -181,6 +197,7 @@ GEM stringio (3.1.2) thor (1.3.2) timeout (0.4.3) + traces (0.18.2) tzinfo (2.0.6) concurrent-ruby (~> 1.0) unicode-display_width (3.1.3) @@ -199,6 +216,7 @@ PLATFORMS DEPENDENCIES appraisal + async (>= 2.24) debug (~> 1.9) logger minitest (~> 5.0) diff --git a/README.md b/README.md index 700dec3d2..24e7acfab 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,10 @@ Or you can also set the environment variable `SOLID_QUEUE_SUPERVISOR_MODE` to `a **The recommended and default mode is `fork`. Only use `async` if you know what you're doing and have strong reasons to** +This supervisor mode is separate from a worker's concurrency model. Supervisor mode decides whether supervised processes live in forks or threads. Worker configuration decides whether claimed jobs run in a thread pool (`threads: N`) or as fibers on a single fiber reactor thread (`fibers: N`). + +Because these are separate concerns, you can combine the default `fork` supervisor mode with fiber workers. In that setup, each worker process gets its own fiber reactor and bounded fiber count. + ## Configuration By default, Solid Queue will try to find your configuration under `config/queue.yml`, but you can set a different path using the environment variable `SOLID_QUEUE_CONFIG` or by using the `-c/--config_file` option with `bin/jobs`, like this: @@ -225,9 +229,9 @@ production: batch_size: 500 concurrency_maintenance_interval: 300 workers: - - queues: "*" - threads: 3 - polling_interval: 2 + - queues: "llm*" + fibers: 100 + polling_interval: 0.05 - queues: [ real_time, background ] threads: 5 polling_interval: 0.1 @@ -274,9 +278,11 @@ Here's an overview of the different options: Check the sections below on [how queue order behaves combined with priorities](#queue-order-and-priorities), and [how the way you specify the queues per worker might affect performance](#queues-specification-and-performance). -- `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. -It is recommended to set this value less than or equal to the queue database's connection pool size minus 2, as each worker thread uses one connection, and two additional connections are reserved for polling and heartbeat. -- `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](#fork-vs-async-mode). +- `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. +- `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). - `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. @@ -367,7 +373,15 @@ queues: back* ### Threads, processes, and signals -Workers in Solid Queue use a thread pool to run work in multiple threads, configurable via the `threads` parameter above. Besides this, parallelism can be achieved via multiple processes on one machine (configurable via different workers or the `processes` parameter above) or by horizontal scaling. +By default, workers in Solid Queue use a thread pool to run work in multiple threads, configurable via the `threads` parameter above. Workers can also be configured with `fibers`, in which case claimed jobs are executed as fibers on a single reactor thread and bounded by the worker's fiber count. Besides this, parallelism can be achieved via multiple processes on one machine (configurable via different workers or the `processes` parameter above) or by horizontal scaling. + +Fiber worker execution is best suited for cooperative, mostly I/O-bound jobs. Blocking or CPU-heavy work still blocks the single reactor thread, so it should not be expected to outperform thread mode for every workload. + +Because fiber workers run multiple fibers on a single thread, Rails must also isolate execution state per fiber rather than per thread. If your app keeps the default thread-scoped isolation level, Solid Queue will raise a boot-time error instead of running fiber workers with shared Active Record state. + +On Rails 7.2 and later, fiber workers can often use a much smaller queue database pool than an equivalent thread pool. A practical starting point is `3-5` queue database connections per worker process: one for job execution, one for polling, one for heartbeats, plus some headroom. In the default `fork` supervisor mode, that guidance applies per worker process. In supervisor `async` mode, all workers share one process, so add together the requirements for the workers running there. + +That lower-pool guidance depends on job code not holding connections open across non-blocking waits. APIs such as `ActiveRecord::Base.connection`, `lease_connection`, `connection_pool.checkout`, or long-lived `with_connection` / transaction blocks can pin connections and push fiber workers back toward thread-like pool usage. On Rails 7.1, plan conservatively and assume the configured fiber count can still grow queue database connection usage. The supervisor is in charge of managing these processes, and it responds to the following signals when running in its own process via `bin/jobs` or with [the Puma plugin](#puma-plugin) with the default `fork` mode: - `TERM`, `INT`: starts graceful termination. The supervisor will send a `TERM` signal to its supervised processes, and it'll wait up to `SolidQueue.shutdown_timeout` time until they're done. If any supervised processes are still around by then, it'll send a `QUIT` signal to them to indicate they must exit. @@ -379,6 +393,10 @@ On Windows, the `QUIT` signal can't be trapped, so the supervisor only responds 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. Jobs that were claimed by processes with an expired heartbeat will be marked as failed with a `SolidQueue::Processes::ProcessPrunedError`. You can configure both the frequency of heartbeats and the threshold to consider a process dead. See the section below for this. +Worker heartbeats are driven by a separate timer task, not by the worker execution backend itself. This means fiber workers do not rely on the reactor loop to prove liveness. However, liveness is still tracked at the worker-process level, not at the individual thread or fiber level. + +This means finished and failed jobs still follow the normal Solid Queue lifecycle, but a single stuck job can remain claimed if the worker process itself is still alive. If you need stronger stuck-job detection, that requires an explicit timeout or watchdog mechanism on top of process heartbeats. + 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. @@ -394,7 +412,7 @@ _Note_: The settings in this section should be set in your `config/application.r There are several settings that control how Solid Queue works that you can set as well: - `logger`: the logger you want Solid Queue to use. Defaults to the app logger. -- `app_executor`: the [Rails executor](https://guides.rubyonrails.org/threading_and_code_execution.html#executor) used to wrap asynchronous operations, defaults to the app executor +- `app_executor`: the [Rails executor](https://guides.rubyonrails.org/threading_and_code_execution.html#executor) used to wrap background operations, defaults to the app executor - `on_thread_error`: custom lambda/Proc to call when there's an error within a Solid Queue thread that takes the exception raised as argument. Defaults to ```ruby diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb index f88ce3ed7..9d538fbfd 100644 --- a/lib/solid_queue/configuration.rb +++ b/lib/solid_queue/configuration.rb @@ -6,7 +6,9 @@ class Configuration include ActiveModel::Validations::Callbacks validate :ensure_configured_processes, :ensure_valid_recurring_tasks - validate :warn_about_incorrectly_sized_thread_pool, :warn_about_missing_config_files + validate :ensure_valid_worker_execution_options + validate :ensure_fiber_workers_have_required_dependency, :ensure_fiber_workers_use_supported_isolation_level + validate :warn_about_incorrectly_sized_database_pool, :warn_about_missing_config_files before_validation { warnings.clear } @@ -37,6 +39,7 @@ def instantiate DEFAULT_CONFIG_FILE_PATH = "config/queue.yml" DEFAULT_RECURRING_SCHEDULE_FILE_PATH = "config/recurring.yml" + FIBER_QUERY_SCOPED_CONNECTIONS_VERSION = Gem::Version.new("7.2.0") def initialize(**options) @options = options.with_defaults(default_options) @@ -99,12 +102,12 @@ def ensure_valid_recurring_tasks end end - def warn_about_incorrectly_sized_thread_pool + def warn_about_incorrectly_sized_database_pool db_pool_size = SolidQueue::Record.connection_pool&.size - if db_pool_size && db_pool_size < estimated_number_of_threads - warnings.add(:base, "Warning: Solid Queue is configured to use #{estimated_number_of_threads} threads but the " \ - "database connection pool is #{db_pool_size}. Increase it in `config/database.yml`") + if db_pool_size && db_pool_size < estimated_database_pool_size + warnings.add(:base, "Warning: Solid Queue needs at least #{estimated_database_pool_size} database connections " \ + "for the configured workers but the database connection pool is #{db_pool_size}. Increase it in `config/database.yml`") end rescue ActiveRecord::ActiveRecordError # No usable database connection. Skip the pool-size warning in that case. @@ -121,6 +124,30 @@ def warn_about_missing_config_files end end + def ensure_valid_worker_execution_options + workers_options.each do |options| + if options.key?(:threads) && options.key?(:fibers) + errors.add(:base, "Workers can specify either `threads` or `fibers`, but not both.") + end + end + end + + def ensure_fiber_workers_have_required_dependency + return unless workers_options.any? { |options| fiber_worker?(options) } + + SolidQueue::ExecutionPools::FiberPool.ensure_dependency! + rescue LoadError => error + errors.add(:base, error.message) + end + + def ensure_fiber_workers_use_supported_isolation_level + return unless workers_options.any? { |options| fiber_worker?(options) } + + SolidQueue::ExecutionPools::FiberPool.ensure_supported_isolation_level! + rescue ArgumentError => error + errors.add(:base, error.message) + end + def default_options { mode: ENV["SOLID_QUEUE_SUPERVISOR_MODE"] || :fork, @@ -162,7 +189,8 @@ def workers 1 end - processes.times.map { Process.new(:worker, worker_options.with_defaults(WORKER_DEFAULTS)) } + defaults = worker_defaults_for(worker_options) + processes.times.map { Process.new(:worker, worker_options.with_defaults(defaults)) } end end @@ -256,10 +284,42 @@ def load_config_from_file(file) end end - def estimated_number_of_threads - # At most "threads" in each worker + 1 thread for the worker + 1 thread for the heartbeat task - thread_count = workers_options.map { |options| options.fetch(:threads, WORKER_DEFAULTS[:threads]) }.max - (thread_count || 1) + 2 + def estimated_database_pool_size + worker_pool_size = workers_options.map { |options| estimated_database_pool_size_for_worker(options) }.max + worker_pool_size || 1 + end + + def estimated_database_pool_size_for_worker(options) + # Connections used to execute jobs + 1 for the worker's polling thread + 1 for the heartbeat task + estimated_execution_connections_for_worker(options) + 2 + end + + def worker_capacity(options) + options[:fibers] || options[:threads] || WORKER_DEFAULTS[:threads] + end + + def estimated_execution_connections_for_worker(options) + fiber_worker?(options) ? fiber_execution_connections_for_worker(options) : worker_capacity(options) + end + + def fiber_execution_connections_for_worker(options) + fiber_jobs_release_connections_between_queries? ? 1 : worker_capacity(options) + end + + def fiber_jobs_release_connections_between_queries? + ActiveRecord.gem_version >= FIBER_QUERY_SCOPED_CONNECTIONS_VERSION + end + + def fiber_worker?(options) + options.key?(:fibers) + end + + def worker_defaults_for(options) + if fiber_worker?(options) + WORKER_DEFAULTS.except(:threads) + else + WORKER_DEFAULTS + end end end end diff --git a/lib/solid_queue/execution_pools.rb b/lib/solid_queue/execution_pools.rb new file mode 100644 index 000000000..766cc1894 --- /dev/null +++ b/lib/solid_queue/execution_pools.rb @@ -0,0 +1,18 @@ +# frozen_string_literal: true + +module SolidQueue + module ExecutionPools + class << self + def build(type:, size:, on_state_change: nil) + case type + when :thread + ThreadPool.new(size, on_state_change: on_state_change) + when :fiber + FiberPool.new(size, on_state_change: on_state_change) + else + raise ArgumentError, "Unknown execution pool type #{type.inspect}. Expected one of: :thread, :fiber" + end + end + end + end +end diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb new file mode 100644 index 000000000..3234e3430 --- /dev/null +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -0,0 +1,239 @@ +# frozen_string_literal: true + +module SolidQueue + module ExecutionPools + class FiberPool + include AppExecutor + + IDLE_WAIT_INTERVAL = 0.01 + + class MissingDependencyError < LoadError + def initialize(error) + super( + "Fiber worker execution requires the `async` gem. " \ + "Add `gem \"async\"` to your Gemfile to configure workers with `fibers`. " \ + "Original error: #{error.message}" + ) + end + end + + class UnsupportedIsolationLevelError < ArgumentError + def initialize(level) + super( + "Fiber worker execution requires fiber-scoped isolated execution state. " \ + "Set `ActiveSupport::IsolatedExecutionState.isolation_level = :fiber` " \ + "(or `config.active_support.isolation_level = :fiber` in Rails). " \ + "Current isolation level: #{level.inspect}" + ) + end + end + + class << self + def ensure_dependency! + ensure_io_timeout_compatibility! + + require "async" + require "async/semaphore" + rescue LoadError => error + raise MissingDependencyError.new(error) + end + + def ensure_supported_isolation_level! + return if supported_isolation_level? + + raise UnsupportedIsolationLevelError.new(ActiveSupport::IsolatedExecutionState.isolation_level) + end + + def supported_isolation_level? + ActiveSupport::IsolatedExecutionState.isolation_level == :fiber + end + + def ensure_io_timeout_compatibility!(io_class = IO) + unless io_class.method_defined?(:timeout) && io_class.method_defined?(:timeout=) + # Async 2.24, which Ruby 3.1 resolves to, expects Ruby's newer IO + # timeout API to exist on any socket it waits on. Older Rubies don't + # provide it, so give the gem the minimal accessor interface it needs. + io_class.class_eval do + def timeout + @timeout + end + + def timeout=(value) + @timeout = value + end + end + end + + return if io_class.const_defined?(:TimeoutError, false) + + io_class.const_set(:TimeoutError, Class.new(StandardError)) + end + end + + attr_reader :size + + def initialize(size, on_state_change: nil) + @size = size + @on_state_change = on_state_change + @available_capacity = size + @mutex = Mutex.new + @state_mutex = Mutex.new + @shutdown = false + @fatal_error = nil + @boot_queue = Thread::Queue.new + @pending_executions = Thread::Queue.new + + self.class.ensure_dependency! + self.class.ensure_supported_isolation_level! + + @reactor_thread = start_reactor + + boot_result = @boot_queue.pop + raise boot_result if boot_result.is_a?(Exception) + end + + def post(execution) + reserved = false + raise_if_fatal_error! + raise RuntimeError, "Execution pool is shutting down" if shutdown? + + reserve_capacity! + reserved = true + pending_executions << execution + rescue Exception + restore_capacity if reserved + raise + end + + def available_capacity + raise_if_fatal_error! + mutex.synchronize { @available_capacity } + end + + def idle? + available_capacity.positive? + end + + def shutdown + state_mutex.synchronize do + next false if @shutdown + + @shutdown = true + end + end + + def shutdown? + state_mutex.synchronize { @shutdown } + end + + def wait_for_termination(timeout) + reactor_thread.join(timeout) + end + + def metadata + { + fiber_pool_size: size, + inflight: size - available_capacity + } + end + + private + attr_reader :boot_queue, :mutex, :on_state_change, :pending_executions, :reactor_thread, :state_mutex + + def name + @name ||= "solid_queue-fiber-pool-#{object_id}" + end + + def start_reactor + create_thread do + Async do |task| + semaphore = Async::Semaphore.new(size, parent: task) + boot_queue << :ready + + wait_for_executions(semaphore) + wait_for_inflight_executions + end + rescue Exception => error + register_fatal_error(error) + raise + end + end + + def wait_for_executions(semaphore) + loop do + schedule_pending_executions(semaphore) + + break if shutdown? && pending_executions.empty? + + # Older versions of the async gem don't support waking the reactor from another + # thread reliably, so we cooperatively poll for newly posted work. + sleep(IDLE_WAIT_INTERVAL) if pending_executions.empty? + end + end + + def schedule_pending_executions(semaphore) + while execution = next_pending_execution + semaphore.async(execution) do |_execution_task, scheduled_execution| + perform_execution(scheduled_execution) + end + end + end + + def next_pending_execution + pending_executions.pop(true) + rescue ThreadError + nil + 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_thread_error(error) + ensure + restore_capacity + end + + def reserve_capacity! + mutex.synchronize do + raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 + + @available_capacity -= 1 + end + end + + def restore_capacity + should_notify = mutex.synchronize do + @available_capacity += 1 + @available_capacity.positive? + end + + on_state_change&.call if should_notify + end + + def register_fatal_error(error) + state_mutex.synchronize do + @fatal_error ||= error + end + + boot_queue << error if boot_queue.empty? + on_state_change&.call + end + + def raise_if_fatal_error! + error = state_mutex.synchronize { @fatal_error } + raise error if error + end + + def wait_for_inflight_executions + sleep(IDLE_WAIT_INTERVAL) while executions_in_flight? + end + + def executions_in_flight? + mutex.synchronize { @available_capacity < size } + end + end + end +end diff --git a/lib/solid_queue/execution_pools/thread_pool.rb b/lib/solid_queue/execution_pools/thread_pool.rb new file mode 100644 index 000000000..5b7d15653 --- /dev/null +++ b/lib/solid_queue/execution_pools/thread_pool.rb @@ -0,0 +1,82 @@ +# frozen_string_literal: true + +module SolidQueue + module ExecutionPools + class ThreadPool + include AppExecutor + + attr_reader :size + + delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor + + def initialize(size, on_state_change: nil) + @size = size + @on_state_change = on_state_change + @available_capacity = size + @mutex = Mutex.new + end + + def post(execution) + reserved = false + reserve_capacity! + reserved = true + + Concurrent::Promises.future_on(executor, execution) do |thread_execution| + wrap_in_app_executor { thread_execution.perform } + rescue Exception => error + handle_thread_error(error) + ensure + restore_capacity + end + rescue Exception + restore_capacity if reserved + raise + end + + def available_capacity + mutex.synchronize { @available_capacity } + end + + def idle? + available_capacity.positive? + end + + def metadata + { + inflight: size - available_capacity, + thread_pool_size: size + } + end + + private + attr_reader :mutex, :on_state_change + + DEFAULT_OPTIONS = { + min_threads: 0, + idletime: 60, + fallback_policy: :abort + } + + def executor + @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) + end + + def reserve_capacity! + mutex.synchronize do + raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 + + @available_capacity -= 1 + end + end + + def restore_capacity + should_notify = mutex.synchronize do + @available_capacity += 1 + @available_capacity.positive? + end + + on_state_change&.call if should_notify + end + end + end +end diff --git a/lib/solid_queue/pool.rb b/lib/solid_queue/pool.rb index 9c3d2a298..356bbaf35 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -1,54 +1,5 @@ # frozen_string_literal: true module SolidQueue - class Pool - include AppExecutor - - attr_reader :size - - delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - - def initialize(size, on_idle: nil) - @size = size - @on_idle = on_idle - @available_threads = Concurrent::AtomicFixnum.new(size) - @mutex = Mutex.new - end - - def post(execution) - available_threads.decrement - - Concurrent::Promises.future_on(executor, execution) do |thread_execution| - wrap_in_app_executor do - thread_execution.perform - ensure - available_threads.increment - mutex.synchronize { on_idle.try(:call) if idle? } - end - end.on_rejection! do |e| - handle_thread_error(e) - end - end - - def idle_threads - available_threads.value - end - - def idle? - idle_threads > 0 - end - - private - attr_reader :available_threads, :on_idle, :mutex - - DEFAULT_OPTIONS = { - min_threads: 0, - idletime: 60, - fallback_policy: :abort - } - - def executor - @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) - end - end + Pool = ExecutionPools::ThreadPool end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index e036a5fd9..31945b95b 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -11,18 +11,29 @@ class Worker < Processes::Poller attr_reader :queues, :pool def initialize(**options) - options = options.dup.with_defaults(SolidQueue::Configuration::WORKER_DEFAULTS) + options = options.dup + validate_execution_options!(options) + + execution_pool_type = options.key?(:fibers) ? :fiber : :thread + execution_pool_size = options[:fibers] || options[:threads] || SolidQueue::Configuration::WORKER_DEFAULTS[:threads] + options = options.with_defaults(worker_defaults_for(options)) # Ensure that the queues array is deep frozen to prevent accidental modification @queues = Array(options[:queues]).map(&:freeze).freeze + @metadata_state_mutex = Mutex.new + @metadata_dirty = false - @pool = Pool.new(options[:threads], on_idle: -> { wake_up }) + @pool_options = { + type: execution_pool_type, + size: execution_pool_size, + on_state_change: -> { mark_metadata_dirty; wake_up } + } super(**options) end def metadata - super.merge(queues: queues.join(","), thread_pool_size: pool.size) + super.merge(queues: queues.join(",")).merge(pool.metadata) end private @@ -32,16 +43,23 @@ def poll pool.post(execution) end + reload_metadata_if_needed(executions.any?) + pool.idle? ? polling_interval : 10.minutes end end def claim_executions with_polling_volume do - SolidQueue::ReadyExecution.claim(queues, pool.idle_threads, process_id) + SolidQueue::ReadyExecution.claim(queues, pool.available_capacity, process_id) end end + def boot + build_pool + super + end + def shutdown pool.shutdown pool.wait_for_termination(SolidQueue.shutdown_timeout) @@ -53,8 +71,49 @@ def all_work_completed? SolidQueue::ReadyExecution.aggregated_count_across(queues).zero? end + def heartbeat + super + reload_metadata + end + def set_procline procline "waiting for jobs in #{queues.join(",")}" end + + def build_pool + @pool ||= ExecutionPools.build(**@pool_options) + end + + def validate_execution_options!(options) + if options.key?(:threads) && options.key?(:fibers) + raise ArgumentError, "Workers can specify either `threads` or `fibers`, but not both." + end + end + + def worker_defaults_for(options) + if options.key?(:fibers) + SolidQueue::Configuration::WORKER_DEFAULTS.except(:threads) + else + SolidQueue::Configuration::WORKER_DEFAULTS + end + end + + def mark_metadata_dirty + metadata_state_mutex.synchronize { @metadata_dirty = true } + end + + def metadata_state_mutex + @metadata_state_mutex + end + + def reload_metadata_if_needed(executions_claimed) + needs_reload = metadata_state_mutex.synchronize do + claimed_or_dirty = executions_claimed || @metadata_dirty + @metadata_dirty = false + claimed_or_dirty + end + + reload_metadata if needs_reload + end end end diff --git a/solid_queue.gemspec b/solid_queue.gemspec index 425a5adc6..19172043d 100644 --- a/solid_queue.gemspec +++ b/solid_queue.gemspec @@ -34,6 +34,7 @@ Gem::Specification.new do |spec| spec.add_development_dependency "appraisal" spec.add_development_dependency "debug", "~> 1.9" + spec.add_development_dependency "async", ">= 2.24" spec.add_development_dependency "minitest", "~> 5.0" spec.add_development_dependency "mocha" spec.add_development_dependency "puma", "~> 7.0" diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb index 3ccd4d461..7f2e7007b 100644 --- a/test/integration/async_processes_lifecycle_test.rb +++ b/test/integration/async_processes_lifecycle_test.rb @@ -130,10 +130,16 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase no_pause = enqueue_store_result_job("no pause") pause = enqueue_store_result_job("pause", pause: SolidQueue.shutdown_timeout + 10.second) - # Wait for the "no pause" job to complete and the pause job to be claimed. - # This ensures the pause job is actively being processed. + # Wait for the "no pause" job to complete and the pause job to start. + # A claimed execution alone is not enough here because the worker may have + # claimed the job but not yet entered `perform`. wait_for_jobs_to_finish_for(3.seconds, except: pause) - wait_for(timeout: 2.seconds) { SolidQueue::ClaimedExecution.exists?(job_id: SolidQueue::Job.find_by(active_job_id: pause.job_id)&.id) } + wait_for(timeout: 5.seconds) do + skip_active_record_query_cache do + JobResult.where(queue_name: :background, status: "started", value: "pause").exists? + end + end + assert_started_job_result("pause") signal_process(@pid, :TERM, wait: 0.2.second) wait_for_jobs_to_finish_for(2.seconds, except: pause) @@ -147,8 +153,7 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase assert_completed_job_results("no pause") assert_job_status(no_pause, :finished) - # The pause job should have started but not completed - assert_started_job_result("pause") + # The pause job should not have completed assert_not_equal "completed", skip_active_record_query_cache { JobResult.find_by(value: "pause")&.status } # After shutdown, the pause job may be either: diff --git a/test/integration/instrumentation_test.rb b/test/integration/instrumentation_test.rb index 88f3fbaff..3ff0b13fd 100644 --- a/test/integration/instrumentation_test.rb +++ b/test/integration/instrumentation_test.rb @@ -10,7 +10,7 @@ class InstrumentationTest < ActiveSupport::TestCase travel_to 2.days.from_now dispatcher = SolidQueue::Dispatcher.new(polling_interval: 0.1, batch_size: 10).tap(&:start) - wait_while_with_timeout!(0.5.seconds) { SolidQueue::ScheduledExecution.any? } + wait_while_with_timeout!(3.seconds) { SolidQueue::ScheduledExecution.any? } dispatcher.stop end diff --git a/test/test_helper.rb b/test/test_helper.rb index 99d00ba27..ca90ddbce 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -29,6 +29,19 @@ def write(...) Logger::LogDevice.prepend(BlockLogDeviceTimeoutExceptions) class ExpectedTestError < RuntimeError; end +module ExecutionIsolationTestHelper + def with_execution_isolation(level) + previous_level = ActiveSupport::IsolatedExecutionState.isolation_level + ActiveSupport::IsolatedExecutionState.isolation_level = level + yield + ensure + ActiveSupport::IsolatedExecutionState.isolation_level = previous_level + end +end + +class Minitest::Test + include ExecutionIsolationTestHelper +end class ActiveSupport::TestCase include ConfigurationTestHelper, ProcessesTestHelper, JobsTestHelper diff --git a/test/unit/async_supervisor_test.rb b/test/unit/async_supervisor_test.rb index 962c4de7e..0ca6a3c2f 100644 --- a/test/unit/async_supervisor_test.rb +++ b/test/unit/async_supervisor_test.rb @@ -97,7 +97,7 @@ class AsyncSupervisorTest < ActiveSupport::TestCase supervisor.stop end - assert_match /Solid Queue is configured to use \d+ threads but the database connection pool is \d+\. Increase it in `config\/database.yml`/, log.string + assert_match /Solid Queue needs at least \d+ database connections for the configured workers but the database connection pool is \d+\. Increase it in `config\/database.yml`/, log.string end test "does not warn on boot when the database connection pool is large enough" do diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb index 1b9a96454..88cc35487 100644 --- a/test/unit/configuration_test.rb +++ b/test/unit/configuration_test.rb @@ -62,6 +62,65 @@ class ConfigurationTest < ActiveSupport::TestCase assert_processes configuration, :worker, 2 end + test "fibers configure workers to execute jobs in fibers" do + with_execution_isolation(:fiber) do + configuration = SolidQueue::Configuration.new( + workers: [ + { queues: "llm*", fibers: 10 }, + { queues: "*", threads: 3 } + ], + dispatchers: [], + skip_recurring: true + ) + + assert configuration.valid? + assert_processes configuration, :worker, 2, fibers: [ 10, nil ], threads: [ nil, 3 ] + end + end + + test "workers reject threads and fibers together" do + with_execution_isolation(:fiber) do + configuration = SolidQueue::Configuration.new( + workers: [ { queues: "llm*", threads: 1, fibers: 1 } ], + dispatchers: [], + skip_recurring: true + ) + + assert_not configuration.valid? + assert_match /either `threads` or `fibers`/, configuration.errors.full_messages.first + end + end + + test "fiber worker size inflates required database pool size on Rails 7.1" do + skip if fiber_workers_release_connections_between_queries? + + with_execution_isolation(:fiber) do + configuration = SolidQueue::Configuration.new( + workers: [ { queues: "llm*", fibers: 1000 } ], + dispatchers: [], + skip_recurring: true + ) + + assert configuration.valid? + assert_match /needs at least 1002 database connections/, configuration.warnings.full_messages.first + end + end + + test "fiber worker size does not inflate required database pool size on Rails 7.2+" do + skip unless fiber_workers_release_connections_between_queries? + + with_execution_isolation(:fiber) do + configuration = SolidQueue::Configuration.new( + workers: [ { queues: "llm*", fibers: 1000 } ], + dispatchers: [], + skip_recurring: true + ) + + assert configuration.valid? + assert_empty configuration.warnings + end + end + test "mulitple workers with the same configuration" do background_worker = { queues: "background", polling_interval: 10, processes: 3 } configuration = SolidQueue::Configuration.new(workers: [ background_worker ]) @@ -205,17 +264,33 @@ class ConfigurationTest < ActiveSupport::TestCase assert_not configuration.valid? assert_equal [ "No processes configured" ], configuration.errors.full_messages + # Fiber workers require fiber-scoped isolated execution state + configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { fibers: 3 } ]) + assert_not configuration.valid? + assert_match /requires fiber-scoped isolated execution state/, configuration.errors.full_messages.first + + # Fiber workers require the async gem + with_execution_isolation(:fiber) do + load_error = LoadError.new("cannot load such file -- async") + missing_dependency_error = SolidQueue::ExecutionPools::FiberPool::MissingDependencyError.new(load_error) + SolidQueue::ExecutionPools::FiberPool.expects(:ensure_dependency!).raises(missing_dependency_error) + + configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { fibers: 3 } ]) + assert_not configuration.valid? + assert_match /gem "async"/, configuration.errors.full_messages.first + end + # Not enough DB connections: still valid so boot is not blocked configuration = SolidQueue::Configuration.new(workers: [ { queues: "background", threads: 50, polling_interval: 10 } ]) assert configuration.valid? end - test "reports an undersized thread pool as a warning rather than an error" do + test "reports an undersized database pool as a warning rather than an error" do configuration = SolidQueue::Configuration.new(workers: [ { queues: "background", threads: 50, polling_interval: 10 } ], skip_recurring: true) assert configuration.valid? assert_equal 1, configuration.warnings.size - assert_match /Solid Queue is configured to use \d+ threads but the database connection pool is \d+\. Increase it in `config\/database.yml`/, configuration.warnings.full_messages.first + assert_match /Solid Queue needs at least \d+ database connections for the configured workers but the database connection pool is \d+\. Increase it in `config\/database.yml`/, configuration.warnings.full_messages.first end test "has no warnings when the database connection pool is large enough" do @@ -248,7 +323,7 @@ class ConfigurationTest < ActiveSupport::TestCase end assert_match "Solid Queue configuration is valid.", out - assert_match /Solid Queue is configured to use \d+ threads but the database connection pool is \d+/, err + assert_match /Solid Queue needs at least \d+ database connections for the configured workers but the database connection pool is \d+/, err end test "check prints errors to stderr and returns false for an invalid configuration" do @@ -261,6 +336,10 @@ class ConfigurationTest < ActiveSupport::TestCase end private + def fiber_workers_release_connections_between_queries? + ActiveRecord.gem_version >= SolidQueue::Configuration::FIBER_QUERY_SCOPED_CONNECTIONS_VERSION + end + def assert_processes(configuration, kind, count, **attributes) processes = configuration.configured_processes.select { |p| p.kind == kind } assert_equal count, processes.size diff --git a/test/unit/execution_pools/fiber_pool_test.rb b/test/unit/execution_pools/fiber_pool_test.rb new file mode 100644 index 000000000..668c00979 --- /dev/null +++ b/test/unit/execution_pools/fiber_pool_test.rb @@ -0,0 +1,135 @@ +require "test_helper" + +class FiberPoolTest < Minitest::Test + Execution = Struct.new(:started, :results, :pause) do + def perform + started << true if started + sleep(pause) if pause + results << [ Thread.current.object_id, Fiber.current.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_raises_a_clear_error_when_the_async_gem_is_unavailable + load_error = LoadError.new("cannot load such file -- async") + + SolidQueue::ExecutionPools::FiberPool.expects(:require).with("async").raises(load_error) + + error = assert_raises SolidQueue::ExecutionPools::FiberPool::MissingDependencyError do + SolidQueue::ExecutionPools::FiberPool.new(3) + end + + assert_match /gem "async"/, error.message + end + + def test_builds_a_fiber_pool + pool = mock + + SolidQueue::ExecutionPools::FiberPool.expects(:new).with(5, on_state_change: nil).returns(pool) + + assert_equal pool, SolidQueue::ExecutionPools.build(type: :fiber, size: 5) + end + + def test_raises_a_clear_error_when_isolation_level_is_not_fiber + error = assert_raises SolidQueue::ExecutionPools::FiberPool::UnsupportedIsolationLevelError do + SolidQueue::ExecutionPools::FiberPool.new(3) + end + + assert_match /isolation_level = :fiber/, error.message + end + + def test_adds_io_timeout_compatibility_for_older_rubies + io_class = Class.new + + SolidQueue::ExecutionPools::FiberPool.ensure_io_timeout_compatibility!(io_class) + + io = io_class.new + assert_nil io.timeout + + io.timeout = 1.second + + assert_equal 1.second, io.timeout + assert io_class.const_defined?(:TimeoutError, false) + end + + def test_executes_jobs_as_fibers_on_a_single_reactor_thread + with_execution_isolation(:fiber) do + pool = SolidQueue::ExecutionPools::FiberPool.new(2) + results = Thread::Queue.new + + pool.post Execution.new(nil, results, 0.05) + pool.post Execution.new(nil, results, 0.05) + + entries = 2.times.map { Timeout.timeout(1.second) { results.pop } } + + assert_equal 1, entries.map(&:first).uniq.count + assert_equal 2, entries.map(&:last).uniq.count + assert_equal 2, pool.available_capacity + assert_equal 0, pool.metadata[:inflight] + ensure + pool&.shutdown + pool&.wait_for_termination(1.second) + end + end + + def test_waits_for_in_flight_executions_during_shutdown + with_execution_isolation(:fiber) do + pool = SolidQueue::ExecutionPools::FiberPool.new(1) + started = Thread::Queue.new + + pool.post Execution.new(started, nil, 0.1) + Timeout.timeout(1.second) { started.pop } + + pool.shutdown + + assert_nil pool.wait_for_termination(0.01) + assert pool.wait_for_termination(1.second) + ensure + pool&.shutdown + pool&.wait_for_termination(1.second) + end + end + + def test_shutdown_wakes_the_reactor_when_idle + with_execution_isolation(:fiber) do + pool = SolidQueue::ExecutionPools::FiberPool.new(1) + + pool.shutdown + + assert pool.wait_for_termination(1.second) + ensure + pool&.shutdown + pool&.wait_for_termination(1.second) + end + end + + def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled + with_execution_isolation(:fiber) do + 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::ExecutionPools::FiberPool.new(1, on_state_change: -> { 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 + assert_raises(Async::Stop) { pool.metadata } + ensure + SolidQueue.on_thread_error = original_on_thread_error + pool&.shutdown + pool&.wait_for_termination(1.second) + end + end +end diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 3d692404b..a7baecb25 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -3,6 +3,26 @@ class WorkerTest < ActiveSupport::TestCase include ActiveSupport::Testing::MethodCallAssertions + self.use_transactional_tests = false + + EXECUTION_MODES = [ + { + name: "thread", + options: { threads: 3 }, + expected_metadata: { + inflight: 0, + thread_pool_size: 3 + } + }, + { + name: "fiber", + options: { fibers: 3 }, + expected_metadata: { + inflight: 0, + fiber_pool_size: 3 + } + } + ].freeze setup do @worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.2) @@ -13,13 +33,117 @@ class WorkerTest < ActiveSupport::TestCase JobBuffer.clear end - test "worker is registered as process" do - @worker.start + EXECUTION_MODES.each do |mode| + test "worker is registered as process in #{mode[:name]} mode" do + with_worker_execution_support(mode[:options]) do + worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.2, **mode[:options]) + + worker.start + wait_for_registered_processes(1, timeout: 1.second) + + process = SolidQueue::Process.first + assert_equal "Worker", process.kind + assert_metadata process, { + queues: "background", + polling_interval: 0.2 + }.merge(mode[:expected_metadata]) + ensure + worker&.stop + wait_for_registered_processes(0, timeout: 1.second) + end + end + + test "claim and process more enqueued jobs than the pool size allows to process at once in #{mode[:name]} mode" do + 5.times do + StoreResultJob.perform_later(:paused, pause: 0.1.second) + end + + 3.times do + StoreResultJob.perform_later(:immediate) + end + + with_worker_execution_support(mode[:options]) do + worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.2, **mode[:options]) + + worker.start + + wait_for_jobs_to_finish_for(2.second) + worker.wake_up + + assert_equal 5, JobResult.where(queue_name: :background, status: "completed", value: :paused).count + assert_equal 3, JobResult.where(queue_name: :background, status: "completed", value: :immediate).count + ensure + worker&.stop + wait_for_registered_processes(0, timeout: 1.second) + end + end + + test "updates inflight metadata after jobs finish in #{mode[:name]} mode" do + StoreResultJob.perform_later(:slow, pause: 0.1.second) + + with_worker_execution_support(mode[:options]) do + worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.05, **mode[:options]) + + worker.start + wait_for_registered_processes(1, timeout: 1.second) + + process = SolidQueue::Process.first + + wait_for(timeout: 2.seconds) { process.reload.metadata["inflight"] == 1 } + assert_equal 1, process.reload.metadata["inflight"] + + wait_for(timeout: 2.seconds) do + process.reload.metadata["inflight"] == 0 && + JobResult.where(queue_name: :background, status: "completed", value: :slow).count == 1 + end + assert_equal 0, process.reload.metadata["inflight"] + assert_equal 1, JobResult.where(queue_name: :background, status: "completed", value: :slow).count + ensure + worker&.stop + wait_for_registered_processes(0, timeout: 1.second) + end + end + end + + test "builds the execution pool on boot instead of initialize" do + pool = SolidQueue::ExecutionPools::ThreadPool.new(3) + + SolidQueue::ExecutionPools.expects(:build).once.with do |**options| + options[:type] == :thread && options[:size] == 3 && options[:on_state_change].respond_to?(:call) + end.returns(pool) + + worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.2) + + assert_nil worker.pool + + worker.start wait_for_registered_processes(1, timeout: 1.second) - process = SolidQueue::Process.first - assert_equal "Worker", process.kind - assert_metadata process, { queues: "background", polling_interval: 0.2, thread_pool_size: 3 } + assert_equal pool, worker.pool + ensure + worker&.stop + wait_for_registered_processes(0, timeout: 1.second) + end + + test "rejects multiple worker pool size options" do + error = assert_raises ArgumentError do + SolidQueue::Worker.new(queues: "background", threads: 3, fibers: 3, polling_interval: 0.2) + end + + assert_match /either `threads` or `fibers`/, error.message + end + + test "defaults thread workers to the configured thread pool size" do + worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.2) + + worker.start + wait_for_registered_processes(1, timeout: 1.second) + + assert_equal 3, worker.pool.size + assert_metadata SolidQueue::Process.first, thread_pool_size: 3, inflight: 0 + ensure + worker&.stop + wait_for_registered_processes(0, timeout: 1.second) end test "errors on polling are passed to on_thread_error and re-raised" do @@ -85,24 +209,6 @@ class WorkerTest < ActiveSupport::TestCase Rails.error.unsubscribe(subscriber) if Rails.error.respond_to?(:unsubscribe) end - test "claim and process more enqueued jobs than the pool size allows to process at once" do - 5.times do |i| - StoreResultJob.perform_later(:paused, pause: 0.1.second) - end - - 3.times do |i| - StoreResultJob.perform_later(:immediate) - end - - @worker.start - - wait_for_jobs_to_finish_for(2.second) - @worker.wake_up - - assert_equal 5, JobResult.where(queue_name: :background, status: "completed", value: :paused).count - assert_equal 3, JobResult.where(queue_name: :background, status: "completed", value: :immediate).count - end - test "polling queries are logged" do log = StringIO.new with_active_record_logger(ActiveSupport::Logger.new(log)) do @@ -173,28 +279,50 @@ class WorkerTest < ActiveSupport::TestCase end test "sleeps `10.minutes` if at capacity" do - 3.times { |i| StoreResultJob.perform_later(i, pause: 1.second) } + 3.times { |i| StoreResultJob.perform_later(i, pause: 5.seconds) } - @worker.expects(:interruptible_sleep).with(10.minutes).at_least_once - @worker.expects(:interruptible_sleep).with(@worker.polling_interval).never - @worker.expects(:handle_thread_error).never + delays = stub_interruptible_sleep(@worker) @worker.start - sleep 1.second + + first_delay = Timeout.timeout(1.second) { delays.pop } + + assert_equal 10.minutes, first_delay end test "sleeps `polling_interval` if worker not at capacity" do - 2.times { |i| StoreResultJob.perform_later(i, pause: 1.second) } + 2.times { |i| StoreResultJob.perform_later(i, pause: 5.seconds) } - @worker.expects(:interruptible_sleep).with(@worker.polling_interval).at_least_once - @worker.expects(:interruptible_sleep).with(10.minutes).never - @worker.expects(:handle_thread_error).never + delays = stub_interruptible_sleep(@worker) @worker.start - sleep 1.second + + first_delay = Timeout.timeout(1.second) { delays.pop } + + assert_equal @worker.polling_interval, first_delay end private + def stub_interruptible_sleep(worker) + delays = Thread::Queue.new + + worker.stubs(:handle_thread_error) + worker.define_singleton_method(:interruptible_sleep) do |delay| + delays << delay + sleep 0.01 + end + + delays + end + + def with_worker_execution_support(options, &block) + if options.key?(:fibers) + with_execution_isolation(:fiber, &block) + else + yield + end + end + def with_polling(silence:) old_silence_polling, SolidQueue.silence_polling = SolidQueue.silence_polling, silence yield From 95823897c970752990247762e9557b305a851381 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 22 Jul 2026 20:35:17 +0200 Subject: [PATCH 02/13] Remove IO timeout compatibility shim for Ruby 3.1 Now that the minimum supported Ruby version is 3.2, IO#timeout, IO#timeout= and IO::TimeoutError, which async expects, are always available natively, and we no longer need to patch them in. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools/fiber_pool.rb | 23 ------------------- test/unit/execution_pools/fiber_pool_test.rb | 14 ----------- 2 files changed, 37 deletions(-) diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index 3234e3430..f99d6014c 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -30,8 +30,6 @@ def initialize(level) class << self def ensure_dependency! - ensure_io_timeout_compatibility! - require "async" require "async/semaphore" rescue LoadError => error @@ -47,27 +45,6 @@ def ensure_supported_isolation_level! def supported_isolation_level? ActiveSupport::IsolatedExecutionState.isolation_level == :fiber end - - def ensure_io_timeout_compatibility!(io_class = IO) - unless io_class.method_defined?(:timeout) && io_class.method_defined?(:timeout=) - # Async 2.24, which Ruby 3.1 resolves to, expects Ruby's newer IO - # timeout API to exist on any socket it waits on. Older Rubies don't - # provide it, so give the gem the minimal accessor interface it needs. - io_class.class_eval do - def timeout - @timeout - end - - def timeout=(value) - @timeout = value - end - end - end - - return if io_class.const_defined?(:TimeoutError, false) - - io_class.const_set(:TimeoutError, Class.new(StandardError)) - end end attr_reader :size diff --git a/test/unit/execution_pools/fiber_pool_test.rb b/test/unit/execution_pools/fiber_pool_test.rb index 668c00979..16e8f75bb 100644 --- a/test/unit/execution_pools/fiber_pool_test.rb +++ b/test/unit/execution_pools/fiber_pool_test.rb @@ -44,20 +44,6 @@ def test_raises_a_clear_error_when_isolation_level_is_not_fiber assert_match /isolation_level = :fiber/, error.message end - def test_adds_io_timeout_compatibility_for_older_rubies - io_class = Class.new - - SolidQueue::ExecutionPools::FiberPool.ensure_io_timeout_compatibility!(io_class) - - io = io_class.new - assert_nil io.timeout - - io.timeout = 1.second - - assert_equal 1.second, io.timeout - assert io_class.const_defined?(:TimeoutError, false) - end - def test_executes_jobs_as_fibers_on_a_single_reactor_thread with_execution_isolation(:fiber) do pool = SolidQueue::ExecutionPools::FiberPool.new(2) From 1e09a00f71cc82c42a2294289ff6b516de809603 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 22 Jul 2026 20:35:28 +0200 Subject: [PATCH 03/13] Block on the pending executions queue instead of polling the reactor The fiber pool's reactor looped with a 10ms sleep to check for newly posted work, and again while waiting for in-flight executions to finish during shutdown, on the assumption that the async gem can't reliably wake the reactor from another thread. It can: Thread::Queue#pop is fiber-scheduler-aware, so a fiber blocking on it suspends properly while other fibers run, and a push from the poller thread wakes the reactor through the scheduler (via Scheduler#unblock and the selector's cross-thread wakeup, reliable since io-event 1.3, which all supported async versions require). This removes the idle CPU cost of the busy-poll and the up-to-10ms latency between claiming a job and executing it. Shutdown now closes the queue, which wakes the blocked pop, drains any already-queued executions and then returns nil, ending the loop. The separate wait for in-flight executions is unnecessary: they run in fibers that are children of the reactor's top-level task, which already waits for all its children before exiting. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools/fiber_pool.rb | 41 +++++-------------- 1 file changed, 11 insertions(+), 30 deletions(-) diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index f99d6014c..72023d643 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -5,8 +5,6 @@ module ExecutionPools class FiberPool include AppExecutor - IDLE_WAIT_INTERVAL = 0.01 - class MissingDependencyError < LoadError def initialize(error) super( @@ -96,6 +94,10 @@ def shutdown next false if @shutdown @shutdown = true + end.tap do |shut_down| + # Wake the reactor: already-queued executions are drained before the + # blocked pop in +wait_for_executions+ returns nil + pending_executions.close if shut_down end end @@ -127,8 +129,9 @@ def start_reactor semaphore = Async::Semaphore.new(size, parent: task) boot_queue << :ready + # The reactor exits when all in-flight execution fibers, children + # of this task, have finished wait_for_executions(semaphore) - wait_for_inflight_executions end rescue Exception => error register_fatal_error(error) @@ -137,31 +140,17 @@ def start_reactor end def wait_for_executions(semaphore) - loop do - schedule_pending_executions(semaphore) - - break if shutdown? && pending_executions.empty? - - # Older versions of the async gem don't support waking the reactor from another - # thread reliably, so we cooperatively poll for newly posted work. - sleep(IDLE_WAIT_INTERVAL) if pending_executions.empty? - end - end - - def schedule_pending_executions(semaphore) - while execution = next_pending_execution + # Thread::Queue#pop is fiber-scheduler-aware: it suspends this fiber, letting + # execution fibers run, and wakes the reactor when the poller thread pushes new + # work or closes the queue on shutdown, after which it drains any remaining + # executions and returns nil + while execution = pending_executions.pop semaphore.async(execution) do |_execution_task, scheduled_execution| perform_execution(scheduled_execution) end end end - def next_pending_execution - pending_executions.pop(true) - rescue ThreadError - nil - end - def perform_execution(execution) wrap_in_app_executor { execution.perform } rescue Async::Stop => error @@ -203,14 +192,6 @@ def raise_if_fatal_error! error = state_mutex.synchronize { @fatal_error } raise error if error end - - def wait_for_inflight_executions - sleep(IDLE_WAIT_INTERVAL) while executions_in_flight? - end - - def executions_in_flight? - mutex.synchronize { @available_capacity < size } - end end end end From 3960d37eb37a28e0e679947481a354f0a45f53ca Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 22 Jul 2026 20:59:59 +0200 Subject: [PATCH 04/13] Add integration test for fiber workers and clarify their docs Cover the full lifecycle of a supervised, forked fiber worker: jobs executing in fibers, graceful termination draining in-flight jobs, and abrupt termination releasing claimed jobs. The unit tests exercise the fiber pool in-process only, which wouldn't catch a pool built before forking, whose reactor thread wouldn't survive the fork. Also restore the generic first worker in the configuration example, adding the fiber worker as an extra entry instead of replacing it, and note that the fiber isolation level is an application-wide setting, not scoped to Solid Queue's processes. Co-Authored-By: Claude Opus 4.8 --- README.md | 11 +- .../fiber_processes_lifecycle_test.rb | 124 ++++++++++++++++++ 2 files changed, 132 insertions(+), 3 deletions(-) create mode 100644 test/integration/fiber_processes_lifecycle_test.rb diff --git a/README.md b/README.md index 24e7acfab..a0b9a2ea0 100644 --- a/README.md +++ b/README.md @@ -229,13 +229,16 @@ production: batch_size: 500 concurrency_maintenance_interval: 300 workers: - - queues: "llm*" - fibers: 100 - polling_interval: 0.05 + - queues: "*" + threads: 3 + polling_interval: 2 - queues: [ real_time, background ] threads: 5 polling_interval: 0.1 processes: 3 + - queues: "api*" + fibers: 100 + polling_interval: 0.05 scheduler: dynamic_tasks_enabled: true polling_interval: 5 @@ -379,6 +382,8 @@ Fiber worker execution is best suited for cooperative, mostly I/O-bound jobs. Bl Because fiber workers run multiple fibers on a single thread, Rails must also isolate execution state per fiber rather than per thread. If your app keeps the default thread-scoped isolation level, Solid Queue will raise a boot-time error instead of running fiber workers with shared Active Record state. +Keep in mind that `config.active_support.isolation_level = :fiber` applies to your whole application, not just to Solid Queue: if you run Solid Queue inside Puma via [the plugin](#puma-plugin), or combine fiber workers with thread workers in the same process using the supervisor's `async` mode, everything in that process will use fiber-scoped execution state. This is fully supported by Rails, but it's a global setting worth being deliberate about. + On Rails 7.2 and later, fiber workers can often use a much smaller queue database pool than an equivalent thread pool. A practical starting point is `3-5` queue database connections per worker process: one for job execution, one for polling, one for heartbeats, plus some headroom. In the default `fork` supervisor mode, that guidance applies per worker process. In supervisor `async` mode, all workers share one process, so add together the requirements for the workers running there. That lower-pool guidance depends on job code not holding connections open across non-blocking waits. APIs such as `ActiveRecord::Base.connection`, `lease_connection`, `connection_pool.checkout`, or long-lived `with_connection` / transaction blocks can pin connections and push fiber workers back toward thread-like pool usage. On Rails 7.1, plan conservatively and assume the configured fiber count can still grow queue database connection usage. diff --git a/test/integration/fiber_processes_lifecycle_test.rb b/test/integration/fiber_processes_lifecycle_test.rb new file mode 100644 index 000000000..3a1eafe35 --- /dev/null +++ b/test/integration/fiber_processes_lifecycle_test.rb @@ -0,0 +1,124 @@ +# frozen_string_literal: true + +require "test_helper" + +class FiberProcessesLifecycleTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + setup do + @pid = fork do + ActiveSupport::IsolatedExecutionState.isolation_level = :fiber + SolidQueue::Supervisor.start(skip_recurring: true, workers: [ { queues: :background, fibers: 3, polling_interval: 0.1 } ]) + end + + wait_for_registered_processes(2, timeout: 3.second) + assert_registered_workers_for(:background, supervisor_pid: @pid) + end + + teardown do + terminate_process(@pid) if process_exists?(@pid) + end + + test "process jobs in fibers in a forked worker" do + worker = find_processes_registered_as("Worker").first + assert_metadata worker, fiber_pool_size: 3 + + 6.times { |i| enqueue_store_result_job("job_#{i}") } + + wait_for_jobs_to_finish_for(3.seconds) + + assert_equal 6, skip_active_record_query_cache { JobResult.count } + 6.times { |i| assert_completed_job_results("job_#{i}", :background) } + + terminate_process(@pid) + assert_clean_termination + end + + test "term supervisor while there are jobs in-flight" do + no_pause = enqueue_store_result_job("no pause") + pause = enqueue_store_result_job("pause", pause: 1.second) + + signal_process(@pid, :TERM, wait: 0.3.second) + wait_for_jobs_to_finish_for(3.seconds) + + assert_completed_job_results("no pause") + assert_completed_job_results("pause") + + assert_job_status(no_pause, :finished) + assert_job_status(pause, :finished) + + wait_for_process_termination_with_timeout(@pid, timeout: 1.second) + assert_clean_termination + end + + test "quit supervisor while there are jobs in-flight" do + no_pause = enqueue_store_result_job("no pause") + # long enough pause to make sure it doesn't finish + pause = enqueue_store_result_job("pause", pause: 60.second) + + wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 0 } + + signal_process(@pid, :QUIT, wait: 0.4.second) + wait_for_jobs_to_finish_for(2.seconds, except: pause) + + wait_while_with_timeout(2.seconds) { process_exists?(@pid) } + assert_not process_exists?(@pid) + + assert_completed_job_results("no pause") + assert_job_status(no_pause, :finished) + + assert_started_job_result("pause") + # Workers were shutdown without a chance to terminate orderly, but + # since they're linked to the supervisor, the supervisor deregistering + # also deregistered them and released claimed jobs + assert_job_status(pause, :ready) + assert_clean_termination + end + + private + def assert_clean_termination + wait_for_registered_processes 0, timeout: 0.2.second + assert_no_registered_processes + assert_no_claimed_jobs + assert_not process_exists?(@pid) + end + + def assert_registered_workers_for(*queues, supervisor_pid: nil) + workers = find_processes_registered_as("Worker") + registered_queues = workers.map { |process| process.metadata["queues"] }.compact + assert_equal queues.map(&:to_s).sort, registered_queues.sort + if supervisor_pid + assert_equal [ supervisor_pid ], workers.map { |process| process.supervisor.pid }.uniq + end + end + + def enqueue_store_result_job(value, queue_name = :background, **options) + StoreResultJob.set(queue: queue_name).perform_later(value, **options) + end + + def assert_completed_job_results(value, queue_name = :background, count = 1) + actual = skip_active_record_query_cache { + JobResult.where(queue_name: queue_name, status: "completed", value: value).count + } + assert_equal count, actual + end + + def assert_started_job_result(value, queue_name = :background, count = 1) + actual = skip_active_record_query_cache { + JobResult.where(queue_name: queue_name, status: "started", value: value).count + } + assert_equal count, actual + end + + def assert_job_status(active_job, status) + # Make sure we skip AR query cache. Otherwise the queries done here + # might be cached and since we haven't done any non-SELECT queries + # after they were cached on the connection used in the test, the cache + # will still apply, even though the data returned by the cached queries + # might have been deleted in the forked processes. + skip_active_record_query_cache do + job = SolidQueue::Job.find_by(active_job_id: active_job.job_id) + assert job.public_send("#{status}?") + end + end +end From 10f0aed66057c9d961de2c73f8b6969898ab41dd Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 08:57:43 +0200 Subject: [PATCH 05/13] Store only the pool type and size in worker metadata The in-flight gauge was refreshed on heartbeats only, so it could show counts that were stale by up to a heartbeat interval; better not to show it than to show stale information, especially given the exact count is always available from claimed executions. The pool type and size are constant, and the pool knows both, so the worker can store them directly when registering, as it did before, and the pools don't need to provide changing metadata at all. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools.rb | 6 +-- lib/solid_queue/execution_pools/fiber_pool.rb | 21 ++++----- .../execution_pools/thread_pool.rb | 19 ++++---- lib/solid_queue/worker.rb | 31 +----------- .../fiber_processes_lifecycle_test.rb | 2 +- test/unit/execution_pools/fiber_pool_test.rb | 6 +-- test/unit/worker_test.rb | 47 ++----------------- 7 files changed, 30 insertions(+), 102 deletions(-) diff --git a/lib/solid_queue/execution_pools.rb b/lib/solid_queue/execution_pools.rb index 766cc1894..4facebfa2 100644 --- a/lib/solid_queue/execution_pools.rb +++ b/lib/solid_queue/execution_pools.rb @@ -3,12 +3,12 @@ module SolidQueue module ExecutionPools class << self - def build(type:, size:, on_state_change: nil) + def build(type:, size:, on_idle: nil) case type when :thread - ThreadPool.new(size, on_state_change: on_state_change) + ThreadPool.new(size, on_idle: on_idle) when :fiber - FiberPool.new(size, on_state_change: on_state_change) + FiberPool.new(size, on_idle: on_idle) else raise ArgumentError, "Unknown execution pool type #{type.inspect}. Expected one of: :thread, :fiber" end diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index 72023d643..5164c841e 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -47,9 +47,13 @@ def supported_isolation_level? attr_reader :size - def initialize(size, on_state_change: nil) + def type + :fiber + end + + def initialize(size, on_idle: nil) @size = size - @on_state_change = on_state_change + @on_idle = on_idle @available_capacity = size @mutex = Mutex.new @state_mutex = Mutex.new @@ -109,15 +113,8 @@ def wait_for_termination(timeout) reactor_thread.join(timeout) end - def metadata - { - fiber_pool_size: size, - inflight: size - available_capacity - } - end - private - attr_reader :boot_queue, :mutex, :on_state_change, :pending_executions, :reactor_thread, :state_mutex + attr_reader :boot_queue, :mutex, :on_idle, :pending_executions, :reactor_thread, :state_mutex def name @name ||= "solid_queue-fiber-pool-#{object_id}" @@ -176,7 +173,7 @@ def restore_capacity @available_capacity.positive? end - on_state_change&.call if should_notify + on_idle&.call if should_notify end def register_fatal_error(error) @@ -185,7 +182,7 @@ def register_fatal_error(error) end boot_queue << error if boot_queue.empty? - on_state_change&.call + on_idle&.call end def raise_if_fatal_error! diff --git a/lib/solid_queue/execution_pools/thread_pool.rb b/lib/solid_queue/execution_pools/thread_pool.rb index 5b7d15653..f517ae58e 100644 --- a/lib/solid_queue/execution_pools/thread_pool.rb +++ b/lib/solid_queue/execution_pools/thread_pool.rb @@ -9,9 +9,13 @@ class ThreadPool delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - def initialize(size, on_state_change: nil) + def type + :thread + end + + def initialize(size, on_idle: nil) @size = size - @on_state_change = on_state_change + @on_idle = on_idle @available_capacity = size @mutex = Mutex.new end @@ -41,15 +45,8 @@ def idle? available_capacity.positive? end - def metadata - { - inflight: size - available_capacity, - thread_pool_size: size - } - end - private - attr_reader :mutex, :on_state_change + attr_reader :mutex, :on_idle DEFAULT_OPTIONS = { min_threads: 0, @@ -75,7 +72,7 @@ def restore_capacity @available_capacity.positive? end - on_state_change&.call if should_notify + on_idle&.call if should_notify end end end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index 31945b95b..d8548a20d 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -20,20 +20,18 @@ def initialize(**options) # Ensure that the queues array is deep frozen to prevent accidental modification @queues = Array(options[:queues]).map(&:freeze).freeze - @metadata_state_mutex = Mutex.new - @metadata_dirty = false @pool_options = { type: execution_pool_type, size: execution_pool_size, - on_state_change: -> { mark_metadata_dirty; wake_up } + on_idle: -> { wake_up } } super(**options) end def metadata - super.merge(queues: queues.join(",")).merge(pool.metadata) + super.merge(queues: queues.join(","), pool_type: pool.type, pool_size: pool.size) end private @@ -43,8 +41,6 @@ def poll pool.post(execution) end - reload_metadata_if_needed(executions.any?) - pool.idle? ? polling_interval : 10.minutes end end @@ -71,11 +67,6 @@ def all_work_completed? SolidQueue::ReadyExecution.aggregated_count_across(queues).zero? end - def heartbeat - super - reload_metadata - end - def set_procline procline "waiting for jobs in #{queues.join(",")}" end @@ -97,23 +88,5 @@ def worker_defaults_for(options) SolidQueue::Configuration::WORKER_DEFAULTS end end - - def mark_metadata_dirty - metadata_state_mutex.synchronize { @metadata_dirty = true } - end - - def metadata_state_mutex - @metadata_state_mutex - end - - def reload_metadata_if_needed(executions_claimed) - needs_reload = metadata_state_mutex.synchronize do - claimed_or_dirty = executions_claimed || @metadata_dirty - @metadata_dirty = false - claimed_or_dirty - end - - reload_metadata if needs_reload - end end end diff --git a/test/integration/fiber_processes_lifecycle_test.rb b/test/integration/fiber_processes_lifecycle_test.rb index 3a1eafe35..22e207e79 100644 --- a/test/integration/fiber_processes_lifecycle_test.rb +++ b/test/integration/fiber_processes_lifecycle_test.rb @@ -21,7 +21,7 @@ class FiberProcessesLifecycleTest < ActiveSupport::TestCase test "process jobs in fibers in a forked worker" do worker = find_processes_registered_as("Worker").first - assert_metadata worker, fiber_pool_size: 3 + assert_metadata worker, pool_type: "fiber", pool_size: 3 6.times { |i| enqueue_store_result_job("job_#{i}") } diff --git a/test/unit/execution_pools/fiber_pool_test.rb b/test/unit/execution_pools/fiber_pool_test.rb index 16e8f75bb..c54b597ea 100644 --- a/test/unit/execution_pools/fiber_pool_test.rb +++ b/test/unit/execution_pools/fiber_pool_test.rb @@ -31,7 +31,7 @@ def test_raises_a_clear_error_when_the_async_gem_is_unavailable def test_builds_a_fiber_pool pool = mock - SolidQueue::ExecutionPools::FiberPool.expects(:new).with(5, on_state_change: nil).returns(pool) + SolidQueue::ExecutionPools::FiberPool.expects(:new).with(5, on_idle: nil).returns(pool) assert_equal pool, SolidQueue::ExecutionPools.build(type: :fiber, size: 5) end @@ -57,7 +57,6 @@ def test_executes_jobs_as_fibers_on_a_single_reactor_thread assert_equal 1, entries.map(&:first).uniq.count assert_equal 2, entries.map(&:last).uniq.count assert_equal 2, pool.available_capacity - assert_equal 0, pool.metadata[:inflight] ensure pool&.shutdown pool&.wait_for_termination(1.second) @@ -103,7 +102,7 @@ def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled original_on_thread_error = SolidQueue.on_thread_error SolidQueue.on_thread_error = ->(error) { reported_errors << error.class.name } - pool = SolidQueue::ExecutionPools::FiberPool.new(1, on_state_change: -> { notifications << :changed }) + pool = SolidQueue::ExecutionPools::FiberPool.new(1, on_idle: -> { notifications << :changed }) pool.post CancelledExecution.new(started) Timeout.timeout(1.second) { started.pop } @@ -111,7 +110,6 @@ def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled error = assert_raises(Async::Stop) { pool.available_capacity } assert_equal [ error.class.name ], reported_errors - assert_raises(Async::Stop) { pool.metadata } ensure SolidQueue.on_thread_error = original_on_thread_error pool&.shutdown diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index a7baecb25..847c0dcd2 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -8,19 +8,11 @@ class WorkerTest < ActiveSupport::TestCase EXECUTION_MODES = [ { name: "thread", - options: { threads: 3 }, - expected_metadata: { - inflight: 0, - thread_pool_size: 3 - } + options: { threads: 3 } }, { name: "fiber", - options: { fibers: 3 }, - expected_metadata: { - inflight: 0, - fiber_pool_size: 3 - } + options: { fibers: 3 } } ].freeze @@ -43,10 +35,7 @@ class WorkerTest < ActiveSupport::TestCase process = SolidQueue::Process.first assert_equal "Worker", process.kind - assert_metadata process, { - queues: "background", - polling_interval: 0.2 - }.merge(mode[:expected_metadata]) + assert_metadata process, queues: "background", polling_interval: 0.2, pool_type: mode[:name], pool_size: 3 ensure worker&.stop wait_for_registered_processes(0, timeout: 1.second) @@ -77,39 +66,13 @@ class WorkerTest < ActiveSupport::TestCase wait_for_registered_processes(0, timeout: 1.second) end end - - test "updates inflight metadata after jobs finish in #{mode[:name]} mode" do - StoreResultJob.perform_later(:slow, pause: 0.1.second) - - with_worker_execution_support(mode[:options]) do - worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.05, **mode[:options]) - - worker.start - wait_for_registered_processes(1, timeout: 1.second) - - process = SolidQueue::Process.first - - wait_for(timeout: 2.seconds) { process.reload.metadata["inflight"] == 1 } - assert_equal 1, process.reload.metadata["inflight"] - - wait_for(timeout: 2.seconds) do - process.reload.metadata["inflight"] == 0 && - JobResult.where(queue_name: :background, status: "completed", value: :slow).count == 1 - end - assert_equal 0, process.reload.metadata["inflight"] - assert_equal 1, JobResult.where(queue_name: :background, status: "completed", value: :slow).count - ensure - worker&.stop - wait_for_registered_processes(0, timeout: 1.second) - end - end end test "builds the execution pool on boot instead of initialize" do pool = SolidQueue::ExecutionPools::ThreadPool.new(3) SolidQueue::ExecutionPools.expects(:build).once.with do |**options| - options[:type] == :thread && options[:size] == 3 && options[:on_state_change].respond_to?(:call) + options[:type] == :thread && options[:size] == 3 && options[:on_idle].respond_to?(:call) end.returns(pool) worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.2) @@ -140,7 +103,7 @@ class WorkerTest < ActiveSupport::TestCase wait_for_registered_processes(1, timeout: 1.second) assert_equal 3, worker.pool.size - assert_metadata SolidQueue::Process.first, thread_pool_size: 3, inflight: 0 + assert_metadata SolidQueue::Process.first, pool_type: "thread", pool_size: 3 ensure worker&.stop wait_for_registered_processes(0, timeout: 1.second) From c83c9f6dcfd7d1f8f9fe6512766e7080335e743d Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 10:01:09 +0200 Subject: [PATCH 06/13] Rely on configuration validation for worker execution options The configuration already validates that workers specify either threads or fibers but not both, and that's where option validation belongs; the worker doesn't validate any of its other options. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/worker.rb | 7 ------- test/unit/worker_test.rb | 8 -------- 2 files changed, 15 deletions(-) diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index d8548a20d..64dd3cfa8 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -12,7 +12,6 @@ class Worker < Processes::Poller def initialize(**options) options = options.dup - validate_execution_options!(options) execution_pool_type = options.key?(:fibers) ? :fiber : :thread execution_pool_size = options[:fibers] || options[:threads] || SolidQueue::Configuration::WORKER_DEFAULTS[:threads] @@ -75,12 +74,6 @@ def build_pool @pool ||= ExecutionPools.build(**@pool_options) end - def validate_execution_options!(options) - if options.key?(:threads) && options.key?(:fibers) - raise ArgumentError, "Workers can specify either `threads` or `fibers`, but not both." - end - end - def worker_defaults_for(options) if options.key?(:fibers) SolidQueue::Configuration::WORKER_DEFAULTS.except(:threads) diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 847c0dcd2..b80fd7858 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -88,14 +88,6 @@ class WorkerTest < ActiveSupport::TestCase wait_for_registered_processes(0, timeout: 1.second) end - test "rejects multiple worker pool size options" do - error = assert_raises ArgumentError do - SolidQueue::Worker.new(queues: "background", threads: 3, fibers: 3, polling_interval: 0.2) - end - - assert_match /either `threads` or `fibers`/, error.message - end - test "defaults thread workers to the configured thread pool size" do worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.2) From 743ea65533491c1f8563a4eeb15f97a7ea909a9e Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 10:38:05 +0200 Subject: [PATCH 07/13] Start the fiber pool's reactor lazily on first use In the default fork supervisor mode, workers are instantiated by the supervisor before forking, so a reactor thread started when the pool is built wouldn't survive the fork. Starting it when the first execution is posted makes the fiber pool safe to build at any point, matching the thread pool, which already builds its executor lazily on first use. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools/fiber_pool.rb | 21 +++++++++++----- test/unit/execution_pools/fiber_pool_test.rb | 25 +++++++++++++++++++ 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index 5164c841e..b5b0e9764 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -61,14 +61,10 @@ def initialize(size, on_idle: nil) @fatal_error = nil @boot_queue = Thread::Queue.new @pending_executions = Thread::Queue.new + @reactor_thread = nil self.class.ensure_dependency! self.class.ensure_supported_isolation_level! - - @reactor_thread = start_reactor - - boot_result = @boot_queue.pop - raise boot_result if boot_result.is_a?(Exception) end def post(execution) @@ -78,6 +74,8 @@ def post(execution) reserve_capacity! reserved = true + + start_reactor_if_needed pending_executions << execution rescue Exception restore_capacity if reserved @@ -110,7 +108,7 @@ def shutdown? end def wait_for_termination(timeout) - reactor_thread.join(timeout) + reactor_thread&.join(timeout) end private @@ -120,6 +118,17 @@ def name @name ||= "solid_queue-fiber-pool-#{object_id}" end + # The reactor thread is started lazily, when the first execution is posted, + # so that the pool can be safely built before forking: in the default fork + # supervisor mode, workers are instantiated in the supervisor process, and + # a thread started there wouldn't survive the fork. + def start_reactor_if_needed + @reactor_thread ||= start_reactor.tap do + boot_result = boot_queue.pop + raise boot_result if boot_result.is_a?(Exception) + end + end + def start_reactor create_thread do Async do |task| diff --git a/test/unit/execution_pools/fiber_pool_test.rb b/test/unit/execution_pools/fiber_pool_test.rb index c54b597ea..cb013d531 100644 --- a/test/unit/execution_pools/fiber_pool_test.rb +++ b/test/unit/execution_pools/fiber_pool_test.rb @@ -84,6 +84,10 @@ def test_waits_for_in_flight_executions_during_shutdown def test_shutdown_wakes_the_reactor_when_idle with_execution_isolation(:fiber) do pool = SolidQueue::ExecutionPools::FiberPool.new(1) + results = Thread::Queue.new + + pool.post Execution.new(nil, results, nil) + Timeout.timeout(1.second) { results.pop } pool.shutdown @@ -94,6 +98,27 @@ def test_shutdown_wakes_the_reactor_when_idle end end + def test_starts_the_reactor_lazily_so_the_pool_can_be_built_before_forking + with_execution_isolation(:fiber) do + pool = SolidQueue::ExecutionPools::FiberPool.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 + end + def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled with_execution_isolation(:fiber) do notifications = Thread::Queue.new From d8a71828a71789fd5805e82064d5ef1615be1d37 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 10:38:05 +0200 Subject: [PATCH 08/13] Simplify worker initialization Build the execution pool directly when the worker is initialized, which is safe now that both pools start their threads lazily. Merge the worker defaults unconditionally and choose the pool type with a single check on the fibers option, relying on the configuration's validation to prevent threads and fibers from being combined. If a worker is instantiated directly with no pool options, it defaults to a thread pool; if it somehow gets both options, fibers win. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/worker.rb | 27 ++++----------------------- test/unit/worker_test.rb | 20 -------------------- 2 files changed, 4 insertions(+), 43 deletions(-) diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index 64dd3cfa8..a351fa677 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -11,20 +11,18 @@ class Worker < Processes::Poller attr_reader :queues, :pool def initialize(**options) - options = options.dup - execution_pool_type = options.key?(:fibers) ? :fiber : :thread - execution_pool_size = options[:fibers] || options[:threads] || SolidQueue::Configuration::WORKER_DEFAULTS[:threads] - options = options.with_defaults(worker_defaults_for(options)) + + options = options.dup.with_defaults(SolidQueue::Configuration::WORKER_DEFAULTS) + execution_pool_size = execution_pool_type == :fiber ? options[:fibers] : options[:threads] # Ensure that the queues array is deep frozen to prevent accidental modification @queues = Array(options[:queues]).map(&:freeze).freeze - @pool_options = { + @pool = ExecutionPools.build \ type: execution_pool_type, size: execution_pool_size, on_idle: -> { wake_up } - } super(**options) end @@ -50,11 +48,6 @@ def claim_executions end end - def boot - build_pool - super - end - def shutdown pool.shutdown pool.wait_for_termination(SolidQueue.shutdown_timeout) @@ -69,17 +62,5 @@ def all_work_completed? def set_procline procline "waiting for jobs in #{queues.join(",")}" end - - def build_pool - @pool ||= ExecutionPools.build(**@pool_options) - end - - def worker_defaults_for(options) - if options.key?(:fibers) - SolidQueue::Configuration::WORKER_DEFAULTS.except(:threads) - else - SolidQueue::Configuration::WORKER_DEFAULTS - end - end end end diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index b80fd7858..c21cd5741 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -68,26 +68,6 @@ class WorkerTest < ActiveSupport::TestCase end end - test "builds the execution pool on boot instead of initialize" do - pool = SolidQueue::ExecutionPools::ThreadPool.new(3) - - SolidQueue::ExecutionPools.expects(:build).once.with do |**options| - options[:type] == :thread && options[:size] == 3 && options[:on_idle].respond_to?(:call) - end.returns(pool) - - worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.2) - - assert_nil worker.pool - - worker.start - wait_for_registered_processes(1, timeout: 1.second) - - assert_equal pool, worker.pool - ensure - worker&.stop - wait_for_registered_processes(0, timeout: 1.second) - end - test "defaults thread workers to the configured thread pool size" do worker = SolidQueue::Worker.new(queues: "background", polling_interval: 0.2) From a2e9f8f9a3a3ac28c0c97f7c01d78b1cd03d7f6b Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 11:24:10 +0200 Subject: [PATCH 09/13] Report errors escaping the execution future The rescue inside the future covers job execution, but an error raised when restoring capacity or waking up the worker would escape it and reject the future with nobody watching. Attach the on_rejection callback the old pool had as a backstop, so those errors are also reported via handle_thread_error. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools/thread_pool.rb | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/lib/solid_queue/execution_pools/thread_pool.rb b/lib/solid_queue/execution_pools/thread_pool.rb index f517ae58e..ff76721b8 100644 --- a/lib/solid_queue/execution_pools/thread_pool.rb +++ b/lib/solid_queue/execution_pools/thread_pool.rb @@ -31,6 +31,10 @@ def post(execution) handle_thread_error(error) ensure restore_capacity + end.on_rejection! do |error| + # Backstop for errors raised outside the rescue above, such as when + # restoring capacity or waking up the worker + handle_thread_error(error) end rescue Exception restore_capacity if reserved From 70844752ab0c3197e6c1ea7fc6a749f0cec9825f Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 11:31:13 +0200 Subject: [PATCH 10/13] Rescue only the code that runs with capacity reserved when posting Instead of tracking whether capacity had been reserved with a local flag, scope the rescue to the code that runs after the reservation, which is the only part that needs to restore it on failure. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools/fiber_pool.rb | 14 +++++----- .../execution_pools/thread_pool.rb | 28 +++++++++---------- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index b5b0e9764..f15dff97c 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -68,18 +68,18 @@ def initialize(size, on_idle: nil) end def post(execution) - reserved = false raise_if_fatal_error! raise RuntimeError, "Execution pool is shutting down" if shutdown? reserve_capacity! - reserved = true - start_reactor_if_needed - pending_executions << execution - rescue Exception - restore_capacity if reserved - raise + begin + start_reactor_if_needed + pending_executions << execution + rescue Exception + restore_capacity + raise + end end def available_capacity diff --git a/lib/solid_queue/execution_pools/thread_pool.rb b/lib/solid_queue/execution_pools/thread_pool.rb index ff76721b8..2cbd9a3d3 100644 --- a/lib/solid_queue/execution_pools/thread_pool.rb +++ b/lib/solid_queue/execution_pools/thread_pool.rb @@ -21,24 +21,24 @@ def initialize(size, on_idle: nil) end def post(execution) - reserved = false reserve_capacity! - reserved = true - Concurrent::Promises.future_on(executor, execution) do |thread_execution| - wrap_in_app_executor { thread_execution.perform } - rescue Exception => error - handle_thread_error(error) - ensure + begin + Concurrent::Promises.future_on(executor, execution) do |thread_execution| + wrap_in_app_executor { thread_execution.perform } + rescue Exception => error + handle_thread_error(error) + ensure + restore_capacity + end.on_rejection! do |error| + # Backstop for errors raised outside the rescue above, such as when + # restoring capacity or waking up the worker + handle_thread_error(error) + end + rescue Exception restore_capacity - end.on_rejection! do |error| - # Backstop for errors raised outside the rescue above, such as when - # restoring capacity or waking up the worker - handle_thread_error(error) + raise end - rescue Exception - restore_capacity if reserved - raise end def available_capacity From d3057d147fedc1d70e0e7963ac676988924c5441 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 11:38:14 +0200 Subject: [PATCH 11/13] Extract a base class for the execution pools Both pools share the capacity ledger: the reserve/restore accounting, the idle notification, and the reserve-then-schedule shape of post. Move all of that to ExecutionPools::Base, leaving each pool with just its scheduling mechanism: the thread pool schedules executions on a Concurrent::ThreadPoolExecutor via schedule, and the fiber pool hands them to its reactor, adding its fatal-error and shutdown guards on top of the base's post. The default way of performing an execution, wrapped in the app executor, reporting errors and restoring capacity, also lives in the base class; the fiber pool overrides it to treat Async::Stop as fatal. The pool type is derived from the class name, and the factory performs the lookup in reverse, trusting the type it receives: the worker only ever passes :thread or :fiber. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools.rb | 13 +--- lib/solid_queue/execution_pools/base.rb | 73 +++++++++++++++++++ lib/solid_queue/execution_pools/fiber_pool.rb | 56 +++----------- .../execution_pools/thread_pool.rb | 73 +++---------------- 4 files changed, 96 insertions(+), 119 deletions(-) create mode 100644 lib/solid_queue/execution_pools/base.rb diff --git a/lib/solid_queue/execution_pools.rb b/lib/solid_queue/execution_pools.rb index 4facebfa2..501984ab4 100644 --- a/lib/solid_queue/execution_pools.rb +++ b/lib/solid_queue/execution_pools.rb @@ -2,17 +2,8 @@ module SolidQueue module ExecutionPools - class << self - def build(type:, size:, on_idle: nil) - case type - when :thread - ThreadPool.new(size, on_idle: on_idle) - when :fiber - FiberPool.new(size, on_idle: on_idle) - else - raise ArgumentError, "Unknown execution pool type #{type.inspect}. Expected one of: :thread, :fiber" - end - end + def self.build(type:, size:, on_idle: nil) + const_get("#{type.to_s.camelize}Pool").new(size, on_idle: on_idle) end end end diff --git a/lib/solid_queue/execution_pools/base.rb b/lib/solid_queue/execution_pools/base.rb new file mode 100644 index 000000000..338e16b63 --- /dev/null +++ b/lib/solid_queue/execution_pools/base.rb @@ -0,0 +1,73 @@ +# frozen_string_literal: true + +module SolidQueue + module ExecutionPools + class Base + include AppExecutor + + attr_reader :size + + def initialize(size, on_idle: nil) + @size = size + @on_idle = on_idle + @available_capacity = size + @mutex = Mutex.new + end + + def type + self.class.name.demodulize.delete_suffix("Pool").underscore.to_sym + end + + def post(execution) + reserve_capacity! + + begin + schedule(execution) + rescue Exception + restore_capacity + raise + end + end + + def available_capacity + mutex.synchronize { @available_capacity } + end + + def idle? + available_capacity.positive? + end + + private + attr_reader :mutex, :on_idle + + def schedule(execution) + raise NotImplementedError + end + + def perform_execution(execution) + wrap_in_app_executor { execution.perform } + rescue Exception => error + handle_thread_error(error) + ensure + restore_capacity + end + + def reserve_capacity! + mutex.synchronize do + raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 + + @available_capacity -= 1 + end + end + + def restore_capacity + should_notify = mutex.synchronize do + @available_capacity += 1 + @available_capacity.positive? + end + + on_idle&.call if should_notify + end + end + end +end diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index f15dff97c..5b0be63f6 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -2,9 +2,7 @@ module SolidQueue module ExecutionPools - class FiberPool - include AppExecutor - + class FiberPool < Base class MissingDependencyError < LoadError def initialize(error) super( @@ -45,17 +43,9 @@ def supported_isolation_level? end end - attr_reader :size - - def type - :fiber - end - def initialize(size, on_idle: nil) - @size = size - @on_idle = on_idle - @available_capacity = size - @mutex = Mutex.new + super + @state_mutex = Mutex.new @shutdown = false @fatal_error = nil @@ -71,24 +61,12 @@ def post(execution) raise_if_fatal_error! raise RuntimeError, "Execution pool is shutting down" if shutdown? - reserve_capacity! - - begin - start_reactor_if_needed - pending_executions << execution - rescue Exception - restore_capacity - raise - end + super end def available_capacity raise_if_fatal_error! - mutex.synchronize { @available_capacity } - end - - def idle? - available_capacity.positive? + super end def shutdown @@ -112,12 +90,17 @@ def wait_for_termination(timeout) end private - attr_reader :boot_queue, :mutex, :on_idle, :pending_executions, :reactor_thread, :state_mutex + attr_reader :boot_queue, :pending_executions, :reactor_thread, :state_mutex def name @name ||= "solid_queue-fiber-pool-#{object_id}" end + def schedule(execution) + start_reactor_if_needed + pending_executions << execution + end + # The reactor thread is started lazily, when the first execution is posted, # so that the pool can be safely built before forking: in the default fork # supervisor mode, workers are instantiated in the supervisor process, and @@ -168,23 +151,6 @@ def perform_execution(execution) restore_capacity end - def reserve_capacity! - mutex.synchronize do - raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 - - @available_capacity -= 1 - end - end - - def restore_capacity - should_notify = mutex.synchronize do - @available_capacity += 1 - @available_capacity.positive? - end - - on_idle&.call if should_notify - end - def register_fatal_error(error) state_mutex.synchronize do @fatal_error ||= error diff --git a/lib/solid_queue/execution_pools/thread_pool.rb b/lib/solid_queue/execution_pools/thread_pool.rb index 2cbd9a3d3..75b135af0 100644 --- a/lib/solid_queue/execution_pools/thread_pool.rb +++ b/lib/solid_queue/execution_pools/thread_pool.rb @@ -2,81 +2,28 @@ module SolidQueue module ExecutionPools - class ThreadPool - include AppExecutor - - attr_reader :size - + class ThreadPool < Base delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - def type - :thread - end - - def initialize(size, on_idle: nil) - @size = size - @on_idle = on_idle - @available_capacity = size - @mutex = Mutex.new - end - - def post(execution) - reserve_capacity! - - begin - Concurrent::Promises.future_on(executor, execution) do |thread_execution| - wrap_in_app_executor { thread_execution.perform } - rescue Exception => error - handle_thread_error(error) - ensure - restore_capacity - end.on_rejection! do |error| - # Backstop for errors raised outside the rescue above, such as when - # restoring capacity or waking up the worker - handle_thread_error(error) - end - rescue Exception - restore_capacity - raise - end - end - - def available_capacity - mutex.synchronize { @available_capacity } - end - - def idle? - available_capacity.positive? - end - private - attr_reader :mutex, :on_idle - DEFAULT_OPTIONS = { min_threads: 0, idletime: 60, fallback_policy: :abort } - def executor - @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) - end - - def reserve_capacity! - mutex.synchronize do - raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 - - @available_capacity -= 1 + def schedule(execution) + Concurrent::Promises.future_on(executor, execution) do |thread_execution| + perform_execution(thread_execution) + end.on_rejection! do |error| + # Backstop for errors raised outside perform_execution's own rescue, + # such as when restoring capacity or waking up the worker + handle_thread_error(error) end end - def restore_capacity - should_notify = mutex.synchronize do - @available_capacity += 1 - @available_capacity.positive? - end - - on_idle&.call if should_notify + def executor + @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) end end end From c49b2fd2bd1205f55156836e516cf1cec6720904 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 11:55:08 +0200 Subject: [PATCH 12/13] Move the fiber worker checks to Configuration The async dependency and isolation level checks were surfaced through Configuration's validations but implemented in the fiber pool, which also ran them when built. Configuration is where these belong: it validates every supervised path before any process is instantiated, and reports through valid? and check. The pool now just requires the async gem lazily when starting its reactor, keeping setups without fiber workers from ever loading it, and otherwise trusts its caller, like the rest of the internal classes. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/configuration.rb | 16 ++++-- lib/solid_queue/execution_pools/fiber_pool.rb | 57 ++++--------------- test/unit/configuration_test.rb | 8 +-- test/unit/execution_pools/fiber_pool_test.rb | 20 ------- 4 files changed, 23 insertions(+), 78 deletions(-) diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb index 9d538fbfd..b2153edeb 100644 --- a/lib/solid_queue/configuration.rb +++ b/lib/solid_queue/configuration.rb @@ -135,17 +135,21 @@ def ensure_valid_worker_execution_options def ensure_fiber_workers_have_required_dependency return unless workers_options.any? { |options| fiber_worker?(options) } - SolidQueue::ExecutionPools::FiberPool.ensure_dependency! - rescue LoadError => error - errors.add(:base, error.message) + require "async" + require "async/semaphore" + rescue LoadError + errors.add(:base, "Fiber workers require the `async` gem. " \ + "Add `gem \"async\"` to your Gemfile to configure workers with `fibers`.") end def ensure_fiber_workers_use_supported_isolation_level return unless workers_options.any? { |options| fiber_worker?(options) } - SolidQueue::ExecutionPools::FiberPool.ensure_supported_isolation_level! - rescue ArgumentError => error - errors.add(:base, error.message) + unless ActiveSupport::IsolatedExecutionState.isolation_level == :fiber + errors.add(:base, "Fiber workers require fiber-scoped isolated execution state. " \ + "Set `config.active_support.isolation_level = :fiber` in your Rails configuration " \ + "(or `ActiveSupport::IsolatedExecutionState.isolation_level = :fiber` outside Rails).") + end end def default_options diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb index 5b0be63f6..4039bd82e 100644 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ b/lib/solid_queue/execution_pools/fiber_pool.rb @@ -3,46 +3,6 @@ module SolidQueue module ExecutionPools class FiberPool < Base - class MissingDependencyError < LoadError - def initialize(error) - super( - "Fiber worker execution requires the `async` gem. " \ - "Add `gem \"async\"` to your Gemfile to configure workers with `fibers`. " \ - "Original error: #{error.message}" - ) - end - end - - class UnsupportedIsolationLevelError < ArgumentError - def initialize(level) - super( - "Fiber worker execution requires fiber-scoped isolated execution state. " \ - "Set `ActiveSupport::IsolatedExecutionState.isolation_level = :fiber` " \ - "(or `config.active_support.isolation_level = :fiber` in Rails). " \ - "Current isolation level: #{level.inspect}" - ) - end - end - - class << self - def ensure_dependency! - require "async" - require "async/semaphore" - rescue LoadError => error - raise MissingDependencyError.new(error) - end - - def ensure_supported_isolation_level! - return if supported_isolation_level? - - raise UnsupportedIsolationLevelError.new(ActiveSupport::IsolatedExecutionState.isolation_level) - end - - def supported_isolation_level? - ActiveSupport::IsolatedExecutionState.isolation_level == :fiber - end - end - def initialize(size, on_idle: nil) super @@ -52,9 +12,6 @@ def initialize(size, on_idle: nil) @boot_queue = Thread::Queue.new @pending_executions = Thread::Queue.new @reactor_thread = nil - - self.class.ensure_dependency! - self.class.ensure_supported_isolation_level! end def post(execution) @@ -104,11 +61,17 @@ def schedule(execution) # The reactor thread is started lazily, when the first execution is posted, # so that the pool can be safely built before forking: in the default fork # supervisor mode, workers are instantiated in the supervisor process, and - # a thread started there wouldn't survive the fork. + # a thread started there wouldn't survive the fork. The async gem is also + # required lazily here, so that setups without fiber workers never load it. def start_reactor_if_needed - @reactor_thread ||= start_reactor.tap do - boot_result = boot_queue.pop - raise boot_result if boot_result.is_a?(Exception) + @reactor_thread ||= begin + require "async" + require "async/semaphore" + + start_reactor.tap do + boot_result = boot_queue.pop + raise boot_result if boot_result.is_a?(Exception) + end end end diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb index 88cc35487..2a969a93c 100644 --- a/test/unit/configuration_test.rb +++ b/test/unit/configuration_test.rb @@ -267,15 +267,13 @@ class ConfigurationTest < ActiveSupport::TestCase # Fiber workers require fiber-scoped isolated execution state configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { fibers: 3 } ]) assert_not configuration.valid? - assert_match /requires fiber-scoped isolated execution state/, configuration.errors.full_messages.first + assert_match /require fiber-scoped isolated execution state/, configuration.errors.full_messages.first # Fiber workers require the async gem with_execution_isolation(:fiber) do - load_error = LoadError.new("cannot load such file -- async") - missing_dependency_error = SolidQueue::ExecutionPools::FiberPool::MissingDependencyError.new(load_error) - SolidQueue::ExecutionPools::FiberPool.expects(:ensure_dependency!).raises(missing_dependency_error) - configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [ { fibers: 3 } ]) + configuration.expects(:require).with("async").raises(LoadError.new("cannot load such file -- async")) + assert_not configuration.valid? assert_match /gem "async"/, configuration.errors.full_messages.first end diff --git a/test/unit/execution_pools/fiber_pool_test.rb b/test/unit/execution_pools/fiber_pool_test.rb index cb013d531..14900d267 100644 --- a/test/unit/execution_pools/fiber_pool_test.rb +++ b/test/unit/execution_pools/fiber_pool_test.rb @@ -16,18 +16,6 @@ def perform end end - def test_raises_a_clear_error_when_the_async_gem_is_unavailable - load_error = LoadError.new("cannot load such file -- async") - - SolidQueue::ExecutionPools::FiberPool.expects(:require).with("async").raises(load_error) - - error = assert_raises SolidQueue::ExecutionPools::FiberPool::MissingDependencyError do - SolidQueue::ExecutionPools::FiberPool.new(3) - end - - assert_match /gem "async"/, error.message - end - def test_builds_a_fiber_pool pool = mock @@ -36,14 +24,6 @@ def test_builds_a_fiber_pool assert_equal pool, SolidQueue::ExecutionPools.build(type: :fiber, size: 5) end - def test_raises_a_clear_error_when_isolation_level_is_not_fiber - error = assert_raises SolidQueue::ExecutionPools::FiberPool::UnsupportedIsolationLevelError do - SolidQueue::ExecutionPools::FiberPool.new(3) - end - - assert_match /isolation_level = :fiber/, error.message - end - def test_executes_jobs_as_fibers_on_a_single_reactor_thread with_execution_isolation(:fiber) do pool = SolidQueue::ExecutionPools::FiberPool.new(2) From 2d9ee53db7c8729c61c61d6d54ed8bf970aa797e Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 23 Jul 2026 12:02:28 +0200 Subject: [PATCH 13/13] Reorganize the execution pools under SolidQueue::Pool Replace the ExecutionPools module with a Pool abstract class that the thread and fiber pools inherit from, and that builds the right pool for a worker, mirroring how Supervisor relates to ForkSupervisor and AsyncSupervisor. Co-Authored-By: Claude Opus 4.8 --- lib/solid_queue/execution_pools.rb | 9 -- lib/solid_queue/execution_pools/base.rb | 73 ---------- lib/solid_queue/execution_pools/fiber_pool.rb | 132 ------------------ .../execution_pools/thread_pool.rb | 30 ---- lib/solid_queue/fiber_pool.rb | 130 +++++++++++++++++ lib/solid_queue/pool.rb | 72 +++++++++- lib/solid_queue/thread_pool.rb | 28 ++++ lib/solid_queue/worker.rb | 2 +- .../{execution_pools => }/fiber_pool_test.rb | 14 +- 9 files changed, 237 insertions(+), 253 deletions(-) delete mode 100644 lib/solid_queue/execution_pools.rb delete mode 100644 lib/solid_queue/execution_pools/base.rb delete mode 100644 lib/solid_queue/execution_pools/fiber_pool.rb delete mode 100644 lib/solid_queue/execution_pools/thread_pool.rb create mode 100644 lib/solid_queue/fiber_pool.rb create mode 100644 lib/solid_queue/thread_pool.rb rename test/unit/{execution_pools => }/fiber_pool_test.rb (86%) diff --git a/lib/solid_queue/execution_pools.rb b/lib/solid_queue/execution_pools.rb deleted file mode 100644 index 501984ab4..000000000 --- a/lib/solid_queue/execution_pools.rb +++ /dev/null @@ -1,9 +0,0 @@ -# frozen_string_literal: true - -module SolidQueue - module ExecutionPools - def self.build(type:, size:, on_idle: nil) - const_get("#{type.to_s.camelize}Pool").new(size, on_idle: on_idle) - end - end -end diff --git a/lib/solid_queue/execution_pools/base.rb b/lib/solid_queue/execution_pools/base.rb deleted file mode 100644 index 338e16b63..000000000 --- a/lib/solid_queue/execution_pools/base.rb +++ /dev/null @@ -1,73 +0,0 @@ -# frozen_string_literal: true - -module SolidQueue - module ExecutionPools - class Base - include AppExecutor - - attr_reader :size - - def initialize(size, on_idle: nil) - @size = size - @on_idle = on_idle - @available_capacity = size - @mutex = Mutex.new - end - - def type - self.class.name.demodulize.delete_suffix("Pool").underscore.to_sym - end - - def post(execution) - reserve_capacity! - - begin - schedule(execution) - rescue Exception - restore_capacity - raise - end - end - - def available_capacity - mutex.synchronize { @available_capacity } - end - - def idle? - available_capacity.positive? - end - - private - attr_reader :mutex, :on_idle - - def schedule(execution) - raise NotImplementedError - end - - def perform_execution(execution) - wrap_in_app_executor { execution.perform } - rescue Exception => error - handle_thread_error(error) - ensure - restore_capacity - end - - def reserve_capacity! - mutex.synchronize do - raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 - - @available_capacity -= 1 - end - end - - def restore_capacity - should_notify = mutex.synchronize do - @available_capacity += 1 - @available_capacity.positive? - end - - on_idle&.call if should_notify - end - end - end -end diff --git a/lib/solid_queue/execution_pools/fiber_pool.rb b/lib/solid_queue/execution_pools/fiber_pool.rb deleted file mode 100644 index 4039bd82e..000000000 --- a/lib/solid_queue/execution_pools/fiber_pool.rb +++ /dev/null @@ -1,132 +0,0 @@ -# frozen_string_literal: true - -module SolidQueue - module ExecutionPools - class FiberPool < Base - def initialize(size, on_idle: nil) - super - - @state_mutex = Mutex.new - @shutdown = false - @fatal_error = nil - @boot_queue = Thread::Queue.new - @pending_executions = Thread::Queue.new - @reactor_thread = nil - 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| - # Wake the reactor: already-queued executions are drained before the - # blocked pop in +wait_for_executions+ returns nil - pending_executions.close if shut_down - end - end - - def shutdown? - state_mutex.synchronize { @shutdown } - end - - def wait_for_termination(timeout) - reactor_thread&.join(timeout) - end - - private - attr_reader :boot_queue, :pending_executions, :reactor_thread, :state_mutex - - def name - @name ||= "solid_queue-fiber-pool-#{object_id}" - end - - def schedule(execution) - start_reactor_if_needed - pending_executions << execution - end - - # The reactor thread is started lazily, when the first execution is posted, - # so that the pool can be safely built before forking: in the default fork - # supervisor mode, workers are instantiated in the supervisor process, and - # a thread started there wouldn't survive the fork. The async gem is also - # required lazily here, so that setups without fiber workers never load it. - def start_reactor_if_needed - @reactor_thread ||= begin - require "async" - require "async/semaphore" - - start_reactor.tap do - boot_result = boot_queue.pop - raise boot_result if boot_result.is_a?(Exception) - end - end - end - - def start_reactor - create_thread do - Async do |task| - semaphore = Async::Semaphore.new(size, parent: task) - boot_queue << :ready - - # The reactor exits when all in-flight execution fibers, children - # of this task, have finished - wait_for_executions(semaphore) - end - rescue Exception => error - register_fatal_error(error) - raise - end - end - - def wait_for_executions(semaphore) - # Thread::Queue#pop is fiber-scheduler-aware: it suspends this fiber, letting - # execution fibers run, and wakes the reactor when the poller thread pushes new - # work or closes the queue on shutdown, after which it drains any remaining - # executions and returns nil - while execution = pending_executions.pop - semaphore.async(execution) do |_execution_task, scheduled_execution| - perform_execution(scheduled_execution) - end - 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_thread_error(error) - ensure - restore_capacity - end - - def register_fatal_error(error) - state_mutex.synchronize do - @fatal_error ||= error - end - - boot_queue << error if boot_queue.empty? - on_idle&.call - end - - def raise_if_fatal_error! - error = state_mutex.synchronize { @fatal_error } - raise error if error - end - end - end -end diff --git a/lib/solid_queue/execution_pools/thread_pool.rb b/lib/solid_queue/execution_pools/thread_pool.rb deleted file mode 100644 index 75b135af0..000000000 --- a/lib/solid_queue/execution_pools/thread_pool.rb +++ /dev/null @@ -1,30 +0,0 @@ -# frozen_string_literal: true - -module SolidQueue - module ExecutionPools - class ThreadPool < Base - delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - - private - DEFAULT_OPTIONS = { - min_threads: 0, - idletime: 60, - fallback_policy: :abort - } - - def schedule(execution) - Concurrent::Promises.future_on(executor, execution) do |thread_execution| - perform_execution(thread_execution) - end.on_rejection! do |error| - # Backstop for errors raised outside perform_execution's own rescue, - # such as when restoring capacity or waking up the worker - handle_thread_error(error) - end - end - - def executor - @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) - end - end - end -end diff --git a/lib/solid_queue/fiber_pool.rb b/lib/solid_queue/fiber_pool.rb new file mode 100644 index 000000000..0a29795fa --- /dev/null +++ b/lib/solid_queue/fiber_pool.rb @@ -0,0 +1,130 @@ +# frozen_string_literal: true + +module SolidQueue + class FiberPool < Pool + def initialize(size, on_idle: nil) + super + + @state_mutex = Mutex.new + @shutdown = false + @fatal_error = nil + @boot_queue = Thread::Queue.new + @pending_executions = Thread::Queue.new + @reactor_thread = nil + 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| + # Wake the reactor: already-queued executions are drained before the + # blocked pop in +wait_for_executions+ returns nil + pending_executions.close if shut_down + end + end + + def shutdown? + state_mutex.synchronize { @shutdown } + end + + def wait_for_termination(timeout) + reactor_thread&.join(timeout) + end + + private + attr_reader :boot_queue, :pending_executions, :reactor_thread, :state_mutex + + def name + @name ||= "solid_queue-fiber-pool-#{object_id}" + end + + def schedule(execution) + start_reactor_if_needed + pending_executions << execution + end + + # The reactor thread is started lazily, when the first execution is posted, + # so that the pool can be safely built before forking: in the default fork + # supervisor mode, workers are instantiated in the supervisor process, and + # a thread started there wouldn't survive the fork. The async gem is also + # required lazily here, so that setups without fiber workers never load it. + def start_reactor_if_needed + @reactor_thread ||= begin + require "async" + require "async/semaphore" + + start_reactor.tap do + boot_result = boot_queue.pop + raise boot_result if boot_result.is_a?(Exception) + end + end + end + + def start_reactor + create_thread do + Async do |task| + semaphore = Async::Semaphore.new(size, parent: task) + boot_queue << :ready + + # The reactor exits when all in-flight execution fibers, children + # of this task, have finished + wait_for_executions(semaphore) + end + rescue Exception => error + register_fatal_error(error) + raise + end + end + + def wait_for_executions(semaphore) + # Thread::Queue#pop is fiber-scheduler-aware: it suspends this fiber, letting + # execution fibers run, and wakes the reactor when the poller thread pushes new + # work or closes the queue on shutdown, after which it drains any remaining + # executions and returns nil + while execution = pending_executions.pop + semaphore.async(execution) do |_execution_task, scheduled_execution| + perform_execution(scheduled_execution) + end + 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_thread_error(error) + ensure + restore_capacity + end + + def register_fatal_error(error) + state_mutex.synchronize do + @fatal_error ||= error + end + + boot_queue << error if boot_queue.empty? + 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/pool.rb b/lib/solid_queue/pool.rb index 356bbaf35..f58ac8fc4 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -1,5 +1,75 @@ # frozen_string_literal: true module SolidQueue - Pool = ExecutionPools::ThreadPool + class Pool + include AppExecutor + + def self.build(type:, size:, on_idle: nil) + SolidQueue.const_get("#{type.to_s.camelize}Pool").new(size, on_idle: on_idle) + end + + attr_reader :size + + def initialize(size, on_idle: nil) + @size = size + @on_idle = on_idle + @available_capacity = size + @mutex = Mutex.new + end + + def type + self.class.name.demodulize.delete_suffix("Pool").underscore.to_sym + end + + def post(execution) + reserve_capacity! + + begin + schedule(execution) + rescue Exception + restore_capacity + raise + end + end + + def available_capacity + mutex.synchronize { @available_capacity } + end + + def idle? + available_capacity.positive? + end + + private + attr_reader :mutex, :on_idle + + def schedule(execution) + raise NotImplementedError + end + + def perform_execution(execution) + wrap_in_app_executor { execution.perform } + rescue Exception => error + handle_thread_error(error) + ensure + restore_capacity + end + + def reserve_capacity! + mutex.synchronize do + raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 + + @available_capacity -= 1 + end + end + + def restore_capacity + should_notify = mutex.synchronize do + @available_capacity += 1 + @available_capacity.positive? + end + + on_idle&.call if should_notify + end + end end diff --git a/lib/solid_queue/thread_pool.rb b/lib/solid_queue/thread_pool.rb new file mode 100644 index 000000000..b14852640 --- /dev/null +++ b/lib/solid_queue/thread_pool.rb @@ -0,0 +1,28 @@ +# frozen_string_literal: true + +module SolidQueue + class ThreadPool < Pool + delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor + + private + DEFAULT_OPTIONS = { + min_threads: 0, + idletime: 60, + fallback_policy: :abort + } + + def schedule(execution) + Concurrent::Promises.future_on(executor, execution) do |thread_execution| + perform_execution(thread_execution) + end.on_rejection! do |error| + # Backstop for errors raised outside perform_execution's own rescue, + # such as when restoring capacity or waking up the worker + handle_thread_error(error) + end + end + + def executor + @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) + end + end +end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index a351fa677..9ba65bf3b 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -19,7 +19,7 @@ def initialize(**options) # Ensure that the queues array is deep frozen to prevent accidental modification @queues = Array(options[:queues]).map(&:freeze).freeze - @pool = ExecutionPools.build \ + @pool = Pool.build \ type: execution_pool_type, size: execution_pool_size, on_idle: -> { wake_up } diff --git a/test/unit/execution_pools/fiber_pool_test.rb b/test/unit/fiber_pool_test.rb similarity index 86% rename from test/unit/execution_pools/fiber_pool_test.rb rename to test/unit/fiber_pool_test.rb index 14900d267..f82694ba5 100644 --- a/test/unit/execution_pools/fiber_pool_test.rb +++ b/test/unit/fiber_pool_test.rb @@ -19,14 +19,14 @@ def perform def test_builds_a_fiber_pool pool = mock - SolidQueue::ExecutionPools::FiberPool.expects(:new).with(5, on_idle: nil).returns(pool) + SolidQueue::FiberPool.expects(:new).with(5, on_idle: nil).returns(pool) - assert_equal pool, SolidQueue::ExecutionPools.build(type: :fiber, size: 5) + assert_equal pool, SolidQueue::Pool.build(type: :fiber, size: 5) end def test_executes_jobs_as_fibers_on_a_single_reactor_thread with_execution_isolation(:fiber) do - pool = SolidQueue::ExecutionPools::FiberPool.new(2) + pool = SolidQueue::FiberPool.new(2) results = Thread::Queue.new pool.post Execution.new(nil, results, 0.05) @@ -45,7 +45,7 @@ def test_executes_jobs_as_fibers_on_a_single_reactor_thread def test_waits_for_in_flight_executions_during_shutdown with_execution_isolation(:fiber) do - pool = SolidQueue::ExecutionPools::FiberPool.new(1) + pool = SolidQueue::FiberPool.new(1) started = Thread::Queue.new pool.post Execution.new(started, nil, 0.1) @@ -63,7 +63,7 @@ def test_waits_for_in_flight_executions_during_shutdown def test_shutdown_wakes_the_reactor_when_idle with_execution_isolation(:fiber) do - pool = SolidQueue::ExecutionPools::FiberPool.new(1) + pool = SolidQueue::FiberPool.new(1) results = Thread::Queue.new pool.post Execution.new(nil, results, nil) @@ -80,7 +80,7 @@ def test_shutdown_wakes_the_reactor_when_idle def test_starts_the_reactor_lazily_so_the_pool_can_be_built_before_forking with_execution_isolation(:fiber) do - pool = SolidQueue::ExecutionPools::FiberPool.new(1) + pool = SolidQueue::FiberPool.new(1) pid = fork do results = Thread::Queue.new @@ -107,7 +107,7 @@ def test_marks_the_pool_as_fatal_when_an_execution_is_cancelled original_on_thread_error = SolidQueue.on_thread_error SolidQueue.on_thread_error = ->(error) { reported_errors << error.class.name } - pool = SolidQueue::ExecutionPools::FiberPool.new(1, on_idle: -> { notifications << :changed }) + pool = SolidQueue::FiberPool.new(1, on_idle: -> { notifications << :changed }) pool.post CancelledExecution.new(started) Timeout.timeout(1.second) { started.pop }