From bee889a01de8a205a41905f0025af8a46641d38b Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 10 Aug 2023 11:01:28 +0200 Subject: [PATCH 1/3] Add additional logging to thread error handling --- lib/solid_queue/app_executor.rb | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/solid_queue/app_executor.rb b/lib/solid_queue/app_executor.rb index d228d3e37..b74194c67 100644 --- a/lib/solid_queue/app_executor.rb +++ b/lib/solid_queue/app_executor.rb @@ -12,6 +12,7 @@ def wrap_in_app_executor(&block) def handle_thread_error(error) if SolidQueue.on_thread_error + SolidQueue.logger.error("[SolidQueue] #{error}") SolidQueue.on_thread_error.call(error) end end From 282756363f24b2edc43623f7de024e367df82715 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 10 Aug 2023 18:10:21 +0200 Subject: [PATCH 2/3] Check number of available threads before posting work to the thread pool Otherwise we risk having all jobs in memory waiting for available threads, that will be lost if the worker is shutdown, and will be left claimed in limbo. --- lib/solid_queue/pool.rb | 25 ++++++++++++++++++++++--- lib/solid_queue/worker.rb | 2 +- test/unit/worker_test.rb | 5 ++++- 3 files changed, 27 insertions(+), 5 deletions(-) diff --git a/lib/solid_queue/pool.rb b/lib/solid_queue/pool.rb index c0de5e36c..3c8a03dc0 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -4,19 +4,21 @@ module SolidQueue class Pool include AppExecutor - attr_accessor :size, :executor - delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor def initialize(size) @size = size - @executor = Concurrent::FixedThreadPool.new(size) + @idle_threads = Concurrent::AtomicFixnum.new(size) end def post(execution, process) + idle_threads.decrement + future = Concurrent::Future.new(args: [ execution, process ], executor: executor) do |thread_execution, thread_process| wrap_in_app_executor do thread_execution.perform(thread_process) + ensure + idle_threads.increment end end @@ -26,5 +28,22 @@ def post(execution, process) future.execute end + + def available_threads + idle_threads.value + end + + private + attr_accessor :size, :idle_threads + + 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 end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index ad4aff104..20d63b1ac 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -18,7 +18,7 @@ def initialize(**options) private def run - claimed_executions = SolidQueue::ReadyExecution.claim(queue, pool_size) + claimed_executions = SolidQueue::ReadyExecution.claim(queue, pool.available_threads) if claimed_executions.size > 0 procline "performing #{claimed_executions.count} jobs in #{queue}" diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 9877cea0a..486107568 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -13,7 +13,9 @@ class WorkerTest < ActiveSupport::TestCase JobBuffer.clear end - test "errors on claiming executions are reported via Rails error subscriber" do + test "errors on claiming executions are reported via Rails error subscriber regardless of on_thread_error setting" do + original_on_thread_error, SolidQueue.on_thread_error = SolidQueue.on_thread_error, nil + subscriber = ErrorBuffer.new Rails.error.subscribe(subscriber) @@ -29,5 +31,6 @@ class WorkerTest < ActiveSupport::TestCase assert_equal "everything is broken", subscriber.messages.first ensure Rails.error.unsubscribe(subscriber) if Rails.error.respond_to?(:unsubscribe) + SolidQueue.on_thread_error = original_on_thread_error end end From 3430c0c0d61849bd3fc8e7e40f45188eeba49433 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 10 Aug 2023 18:43:24 +0200 Subject: [PATCH 3/3] Allow waking up worker when there are idle threads waiting for work To prevent sleeping for the whole polling interval if there are jobs waiting but there weren't any available threads to take them. --- app/models/solid_queue/ready_execution.rb | 2 + lib/solid_queue/interruptible.rb | 47 ++++++++++++----------- lib/solid_queue/pool.rb | 23 +++++++---- lib/solid_queue/worker.rb | 9 ++--- test/unit/worker_test.rb | 21 +++++++++- 5 files changed, 67 insertions(+), 35 deletions(-) diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb index d08aa1ccf..22aa03430 100644 --- a/app/models/solid_queue/ready_execution.rb +++ b/app/models/solid_queue/ready_execution.rb @@ -6,6 +6,8 @@ class SolidQueue::ReadyExecution < SolidQueue::Execution class << self def claim(queues, limit) + return [] unless limit > 0 + candidate_job_ids = [] transaction do diff --git a/lib/solid_queue/interruptible.rb b/lib/solid_queue/interruptible.rb index 27504d46e..c8f9f552c 100644 --- a/lib/solid_queue/interruptible.rb +++ b/lib/solid_queue/interruptible.rb @@ -1,32 +1,35 @@ # frozen_string_literal: true module SolidQueue::Interruptible - private + def wake_up + interrupt + end - SELF_PIPE_BLOCK_SIZE = 11 + private + SELF_PIPE_BLOCK_SIZE = 11 - def interrupt - self_pipe[:writer].write_nonblock( "." ) - rescue Errno::EAGAIN, Errno::EINTR - # Ignore writes that would block and retry - # if another signal arrived while writing - retry - end + def interrupt + self_pipe[:writer].write_nonblock( "." ) + rescue Errno::EAGAIN, Errno::EINTR + # Ignore writes that would block and retry + # if another signal arrived while writing + retry + end - def interruptible_sleep(time) - if self_pipe[:reader].wait_readable(time) - loop { self_pipe[:reader].read_nonblock(SELF_PIPE_BLOCK_SIZE) } + def interruptible_sleep(time) + if self_pipe[:reader].wait_readable(time) + loop { self_pipe[:reader].read_nonblock(SELF_PIPE_BLOCK_SIZE) } + end + rescue Errno::EAGAIN, Errno::EINTR end - rescue Errno::EAGAIN, Errno::EINTR - end - # Self-pipe for signal-handling (http://cr.yp.to/docs/selfpipe.html) - def self_pipe - @self_pipe ||= create_self_pipe - end + # Self-pipe for signal-handling (http://cr.yp.to/docs/selfpipe.html) + def self_pipe + @self_pipe ||= create_self_pipe + end - def create_self_pipe - reader, writer = IO.pipe - { reader: reader, writer: writer } - end + def create_self_pipe + reader, writer = IO.pipe + { reader: reader, writer: writer } + end end diff --git a/lib/solid_queue/pool.rb b/lib/solid_queue/pool.rb index 3c8a03dc0..8d7dd4261 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -4,21 +4,26 @@ module SolidQueue class Pool include AppExecutor + attr_reader :size + delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - def initialize(size) + def initialize(size, on_idle: nil) @size = size - @idle_threads = Concurrent::AtomicFixnum.new(size) + @on_idle = on_idle + @available_threads = Concurrent::AtomicFixnum.new(size) + @mutex = Mutex.new end def post(execution, process) - idle_threads.decrement + available_threads.decrement future = Concurrent::Future.new(args: [ execution, process ], executor: executor) do |thread_execution, thread_process| wrap_in_app_executor do thread_execution.perform(thread_process) ensure - idle_threads.increment + available_threads.increment + mutex.synchronize { on_idle.try(:call) if idle? } end end @@ -29,12 +34,16 @@ def post(execution, process) future.execute end - def available_threads - idle_threads.value + def idle_threads + available_threads.value + end + + def idle? + idle_threads > 0 end private - attr_accessor :size, :idle_threads + attr_reader :available_threads, :on_idle, :mutex DEFAULT_OPTIONS = { min_threads: 0, diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index 20d63b1ac..998ab5eb5 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -4,7 +4,7 @@ module SolidQueue class Worker include Runner - attr_accessor :queue, :polling_interval, :pool, :pool_size + attr_accessor :queue, :polling_interval, :pool def initialize(**options) options = options.dup.with_defaults(SolidQueue::Configuration::WORKER_DEFAULTS) @@ -12,13 +12,12 @@ def initialize(**options) @queue = options[:queue_name].to_s @polling_interval = options[:polling_interval] - @pool_size = options[:pool_size] - @pool = Pool.new(options[:pool_size]) + @pool = Pool.new(options[:pool_size], on_idle: -> { wake_up }) end private def run - claimed_executions = SolidQueue::ReadyExecution.claim(queue, pool.available_threads) + claimed_executions = SolidQueue::ReadyExecution.claim(queue, pool.idle_threads) if claimed_executions.size > 0 procline "performing #{claimed_executions.count} jobs in #{queue}" @@ -44,7 +43,7 @@ def shutdown_completed? end def metadata - super.merge(queue: queue, pool_size: pool_size, polling_interval: polling_interval) + super.merge(queue: queue, pool_size: pool.size, idle_threads: pool.idle_threads, polling_interval: polling_interval) end end end diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 486107568..a371f0d51 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -5,7 +5,7 @@ class WorkerTest < ActiveSupport::TestCase include ActiveSupport::Testing::MethodCallAssertions setup do - @worker = SolidQueue::Worker.new(queue_name: "background", pool_size: 3, polling_interval: 1) + @worker = SolidQueue::Worker.new(queue_name: "background", pool_size: 3, polling_interval: 10) end teardown do @@ -26,6 +26,7 @@ class WorkerTest < ActiveSupport::TestCase @worker.start(mode: :async) wait_for_jobs_to_finish_for(0.5.second) + @worker.wake_up assert_equal 1, subscriber.errors.count assert_equal "everything is broken", subscriber.messages.first @@ -33,4 +34,22 @@ class WorkerTest < ActiveSupport::TestCase Rails.error.unsubscribe(subscriber) if Rails.error.respond_to?(:unsubscribe) SolidQueue.on_thread_error = original_on_thread_error 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: 1.second) + end + + 3.times do |i| + StoreResultJob.perform_later(:immediate) + end + + @worker.start(mode: :async) + + wait_for_jobs_to_finish_for(2.3.seconds) # 3 jobs of 1 second in parallel + 2 jobs 1 second in parallel + 3 immediate jobs + @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 end