Take into account the number of available threads before posting work to the thread pool

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

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

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

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

lib/solid_queue/worker.rb

@@ -4,21 +4,20 @@ 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)
 
       @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

test/unit/worker_test.rb

@@ -5,15 +5,17 @@ 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
     @worker.stop if @worker.running?
     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