Merge pull request #58 from basecamp/refactor-processes

Refactor processes: supervisor, dispatcher, worker, and all related concerns

Author: rosa

lib/solid_queue/queue_selector.rb renamed to app/models/solid_queue/queue_selector.rb

@@ -5,15 +5,17 @@
 require "active_job/concurrency_controls"
 
 require "solid_queue/app_executor"
-require "solid_queue/interruptible"
-require "solid_queue/pidfile"
-require "solid_queue/procline"
-require "solid_queue/signals"
+require "solid_queue/processes/supervised"
+require "solid_queue/processes/registrable"
+require "solid_queue/processes/interruptible"
+require "solid_queue/processes/pidfile"
+require "solid_queue/processes/procline"
+require "solid_queue/processes/poller"
+require "solid_queue/processes/base"
+require "solid_queue/processes/runnable"
+require "solid_queue/processes/signals"
 require "solid_queue/configuration"
 require "solid_queue/pool"
-require "solid_queue/queue_selector"
-require "solid_queue/runner"
-require "solid_queue/process_registration"
 require "solid_queue/worker"
 require "solid_queue/dispatcher"
 require "solid_queue/supervisor"

lib/solid_queue.rb

@@ -11,8 +11,9 @@ def wrap_in_app_executor(&block)
     end
 
     def handle_thread_error(error)
+      SolidQueue.logger.error("[SolidQueue] #{error}")
+
       if SolidQueue.on_thread_error
-        SolidQueue.logger.error("[SolidQueue] #{error}")
         SolidQueue.on_thread_error.call(error)
       end
     end

lib/solid_queue/app_executor.rb

@@ -19,7 +19,7 @@ def initialize(mode: :work, load_from: nil)
       @raw_config = config_from(load_from)
     end
 
-    def runners
+    def processes
       case mode
       when :dispatch then dispatcher
       when :work     then workers

lib/solid_queue/configuration.rb

@@ -1,12 +1,12 @@
 # frozen_string_literal: true
 
 module SolidQueue
-  class Dispatcher
-    include Runner
+  class Dispatcher < Processes::Base
+    include Processes::Runnable, Processes::Poller
 
-    attr_accessor :batch_size, :polling_interval, :concurrency_maintenance_interval
+    attr_accessor :batch_size, :concurrency_maintenance_interval
 
-    set_callback :start, :before, :launch_concurrency_maintenance
+    set_callback :boot, :after, :launch_concurrency_maintenance
     set_callback :shutdown, :before, :stop_concurrency_maintenance
 
     def initialize(**options)
@@ -67,7 +67,7 @@ def initial_jitter
       end
 
       def metadata
-        super.merge(batch_size: batch_size, polling_interval: polling_interval)
+        super.merge(batch_size: batch_size)
       end
   end
 end

lib/solid_queue/dispatcher.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  module Processes
+    class Base
+      include ActiveSupport::Callbacks
+      define_callbacks :boot, :shutdown
+
+      include AppExecutor, Registrable, Interruptible, Procline
+
+      private
+        def observe_initial_delay
+          interruptible_sleep(initial_jitter)
+        end
+
+        def boot
+        end
+
+        def shutdown
+        end
+
+        def initial_jitter
+          0
+        end
+    end
+  end
+end

lib/solid_queue/interruptible.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+module SolidQueue::Processes
+  module Interruptible
+    def wake_up
+      interrupt
+    end
+
+    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 interruptible_sleep(time)
+        if time > 0 && self_pipe[:reader].wait_readable(time)
+          loop { self_pipe[:reader].read_nonblock(SELF_PIPE_BLOCK_SIZE) }
+        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
+
+      def create_self_pipe
+        reader, writer = IO.pipe
+        { reader: reader, writer: writer }
+      end
+  end
+end

lib/solid_queue/processes/base.rb

@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 
-module SolidQueue
+module SolidQueue::Processes
   class Pidfile
     def initialize(path)
       @path = path

lib/solid_queue/processes/interruptible.rb

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+module SolidQueue::Processes
+  module Poller
+    extend ActiveSupport::Concern
+
+    included do
+      attr_accessor :polling_interval
+    end
+
+    private
+      def with_polling_volume
+        if SolidQueue.silence_polling?
+          ActiveRecord::Base.logger.silence { yield }
+        else
+          yield
+        end
+      end
+
+      def metadata
+        super.merge(polling_interval: polling_interval)
+      end
+  end
+end