Run non-standalone supervisor in a separate thread

Otherwise `supervise` will block and we won't be able to stop it
externally (from the Puma plugin).

Author: rosa

lib/solid_queue/app_executor.rb

@@ -17,5 +17,15 @@ def handle_thread_error(error)
         SolidQueue.on_thread_error.call(error)
       end
     end
+
+    def create_thread(&block)
+      Thread.new do
+        Thread.current.name = name
+        block.call
+      rescue Exception => exception
+        handle_thread_error(exception)
+        raise
+      end
+    end
   end
 end

lib/solid_queue/async_supervisor.rb

@@ -2,36 +2,47 @@
 
 module SolidQueue
   class AsyncSupervisor < Supervisor
+    def stop
+      super
+      @thread&.join
+    end
+
     private
+      def supervise
+        if standalone? then super
+        else
+          @thread = create_thread { super }
+        end
+      end
 
-    def check_and_replace_terminated_processes
-      terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
-      terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
-    end
+      def check_and_replace_terminated_processes
+        terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
+        terminated_threads.each { |thread_id, instance| replace_thread(thread_id, instance) }
+      end
 
-    def replace_thread(thread_id, instance)
-      SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
-        payload[:thread] = instance
+      def replace_thread(thread_id, instance)
+        SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
+          payload[:thread] = instance
 
-        error = Processes::ThreadTerminatedError.new(terminated_instance.name)
-        release_claimed_jobs_by(terminated_instance, with_error: error)
+          error = Processes::ThreadTerminatedError.new(terminated_instance.name)
+          release_claimed_jobs_by(terminated_instance, with_error: error)
 
-        start_process(configured_processes.delete(thread_id))
+          start_process(configured_processes.delete(thread_id))
+        end
       end
-    end
 
-    def perform_graceful_termination
-      process_instances.values.each(&:stop)
+      def perform_graceful_termination
+        process_instances.values.each(&:stop)
 
-      Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
-    end
+        Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+      end
 
-    def perform_immediate_termination
-      exit!
-    end
+      def perform_immediate_termination
+        exit!
+      end
 
-    def all_processes_terminated?
-      process_instances.values.none?(&:alive?)
-    end
+      def all_processes_terminated?
+        process_instances.values.none?(&:alive?)
+      end
   end
 end

lib/solid_queue/dispatcher.rb

@@ -3,6 +3,7 @@
 module SolidQueue
   class Dispatcher < Processes::Poller
     include LifecycleHooks
+
     attr_reader :batch_size
 
     after_boot :run_start_hooks

lib/solid_queue/processes/runnable.rb

@@ -92,16 +92,5 @@ def running_async?
       def running_as_fork?
         mode.fork?
       end
-
-
-      def create_thread(&block)
-        Thread.new do
-          Thread.current.name = name
-          block.call
-        rescue Exception => exception
-          handle_thread_error(exception)
-          raise
-        end
-      end
   end
 end

test/integration/instrumentation_test.rb

@@ -171,6 +171,7 @@ class InstrumentationTest < ActiveSupport::TestCase
         wait_for_registered_processes(1, timeout: 1.second)
 
         worker.stop
+        wait_for_registered_processes(0, timeout: 1.second)
       end
     end
 

test/unit/async_supervisor_test.rb

@@ -4,15 +4,14 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   test "start as non-standalone" do
-    supervisor, thread = run_supervisor_as_thread
+    supervisor = run_supervisor_as_thread
     wait_for_registered_processes(4)
 
     assert_registered_processes(kind: "Supervisor(async)")
     assert_registered_processes(kind: "Worker", supervisor_id: supervisor.process_id, count: 2)
     assert_registered_processes(kind: "Dispatcher", supervisor_id: supervisor.process_id)
 
     supervisor.stop
-    thread.join
 
     assert_no_registered_processes
   end
@@ -30,15 +29,14 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
   end
 
   test "start as non-standalone with provided configuration" do
-    supervisor, thread = run_supervisor_as_thread(workers: [], dispatchers: [ { batch_size: 100 } ])
+    supervisor = run_supervisor_as_thread(workers: [], dispatchers: [ { batch_size: 100 } ])
     wait_for_registered_processes(2) # supervisor + dispatcher
 
     assert_registered_processes(kind: "Supervisor(async)")
     assert_registered_processes(kind: "Worker", count: 0)
     assert_registered_processes(kind: "Dispatcher", supervisor_id: supervisor.process_id)
 
     supervisor.stop
-    thread.join
 
     assert_no_registered_processes
   end
@@ -51,14 +49,13 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
       dispatchers: []
     }
 
-    supervisor, thread = run_supervisor_as_thread(**config)
+    supervisor = run_supervisor_as_thread(**config)
     wait_for_registered_processes(2) # supervisor + 1 worker
     assert_registered_processes(kind: "Supervisor(async)")
 
     wait_while_with_timeout(1.second) { SolidQueue::ClaimedExecution.count > 0 }
 
     supervisor.stop
-    thread.join
 
     skip_active_record_query_cache do
       assert_equal 0, SolidQueue::ClaimedExecution.count
@@ -89,13 +86,8 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
   end
 
   private
-    def run_supervisor_as_thread(**kwargs)
-      configuration = SolidQueue::Configuration.new(mode: :async, standalone: false, **kwargs)
-      supervisor = SolidQueue::AsyncSupervisor.new(configuration)
-
-      thread = Thread.new { supervisor.start }
-
-      [ supervisor, thread ]
+    def run_supervisor_as_thread(**options)
+      SolidQueue::Supervisor.start(mode: :async, standalone: false, **options)
     end
 
     def simulate_orphaned_executions(count)

test/unit/dispatcher_test.rb

@@ -12,7 +12,6 @@ class DispatcherTest < ActiveSupport::TestCase
 
   teardown do
     @dispatcher.stop
-    wait_for_registered_processes(0, timeout: 2.seconds)
   end
 
   test "dispatcher is registered as process" do
@@ -76,52 +75,59 @@ class DispatcherTest < ActiveSupport::TestCase
     assert_no_registered_processes
   end
 
-  test "dispatch scheduled executions" do
-    skip_active_record_query_cache do
-      15.times do
-        AddToBufferJob.set(wait: 0.5.second).perform_later("I'm scheduled")
-      end
-      sleep 0.5.second
-      assert_equal 15, SolidQueue::ScheduledExecution.count
+  test "run more than one instance of the dispatcher" do
+    15.times do
+      AddToBufferJob.set(wait: 0.5.second).perform_later("I'm scheduled")
+    end
+    sleep 0.5.second
+    assert_equal 15, SolidQueue::ScheduledExecution.count
 
-      @dispatcher.start
-      wait_for_registered_processes(1, timeout: 2.seconds)
+    another_dispatcher = SolidQueue::Dispatcher.new(polling_interval: 0.1, batch_size: 10)
 
-      wait_while_with_timeout(3.seconds) { SolidQueue::ScheduledExecution.any? }
+    @dispatcher.start
+    another_dispatcher.start
+
+    wait_while_with_timeout(1.second) { SolidQueue::ScheduledExecution.any? }
 
+    skip_active_record_query_cache do
       assert_equal 0, SolidQueue::ScheduledExecution.count
       assert_equal 15, SolidQueue::ReadyExecution.count
     end
+  ensure
+    another_dispatcher&.stop
   end
 
   test "sleeps `0.seconds` between polls if there are ready to dispatch jobs" do
-    skip_active_record_query_cache do
-      @dispatcher = SolidQueue::Dispatcher.new(polling_interval: 10, batch_size: 1)
-      @dispatcher.expects(:interruptible_sleep).with(0.seconds).at_least(3)
-      @dispatcher.expects(:interruptible_sleep).with(@dispatcher.polling_interval).at_least_once
-      @dispatcher.expects(:handle_thread_error).never
+    dispatcher = SolidQueue::Dispatcher.new(polling_interval: 10, batch_size: 1)
+    dispatcher.expects(:interruptible_sleep).with(0.seconds).at_least(3)
+    dispatcher.expects(:interruptible_sleep).with(dispatcher.polling_interval).at_least_once
+    dispatcher.expects(:handle_thread_error).never
 
-      3.times { AddToBufferJob.set(wait: 0.5.second).perform_later("I'm scheduled") }
-      sleep 0.5.second
-      assert_equal 3, SolidQueue::ScheduledExecution.count
+    3.times { AddToBufferJob.set(wait: 0.5.second).perform_later("I'm scheduled") }
+    sleep 0.5.second
+    assert_equal 3, SolidQueue::ScheduledExecution.count
 
-      @dispatcher.start
-      wait_for_registered_processes(1, timeout: 2.seconds)
-      wait_while_with_timeout(3.seconds) { SolidQueue::ScheduledExecution.any? }
+    dispatcher.start
+    wait_while_with_timeout(1.second) { SolidQueue::ScheduledExecution.any? }
 
+    skip_active_record_query_cache do
       assert_equal 0, SolidQueue::ScheduledExecution.count
       assert_equal 3, SolidQueue::ReadyExecution.count
     end
+  ensure
+    dispatcher.stop
   end
 
   test "sleeps `polling_interval` between polls if there are no un-dispatched jobs" do
-    @dispatcher = SolidQueue::Dispatcher.new(polling_interval: 10, batch_size: 1)
-    @dispatcher.expects(:interruptible_sleep).with(0.seconds).never
-    @dispatcher.expects(:interruptible_sleep).with(@dispatcher.polling_interval).at_least_once
-    @dispatcher.expects(:handle_thread_error).never
+    dispatcher = SolidQueue::Dispatcher.new(polling_interval: 10, batch_size: 1)
+    dispatcher.expects(:interruptible_sleep).with(0.seconds).never
+    dispatcher.expects(:interruptible_sleep).with(dispatcher.polling_interval).at_least_once
+    dispatcher.expects(:handle_thread_error).never
 
-    @dispatcher.start
+    dispatcher.start
     wait_while_with_timeout(1.second) { !SolidQueue::ScheduledExecution.exists? }
+  ensure
+    dispatcher.stop
   end
 
   private