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/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 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 c0de5e36c..8d7dd4261 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -4,19 +4,26 @@ module SolidQueue class Pool include AppExecutor - attr_accessor :size, :executor + attr_reader :size delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - def initialize(size) + def initialize(size, on_idle: nil) @size = size - @executor = Concurrent::FixedThreadPool.new(size) + @on_idle = on_idle + @available_threads = Concurrent::AtomicFixnum.new(size) + @mutex = Mutex.new end def post(execution, process) + 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 + available_threads.increment + mutex.synchronize { on_idle.try(:call) if idle? } end end @@ -26,5 +33,26 @@ def post(execution, process) future.execute 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 end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index ad4aff104..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_size) + 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 9877cea0a..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 @@ -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) @@ -24,10 +26,30 @@ 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 ensure 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