diff --git a/lib/solid_queue/pool.rb b/lib/solid_queue/pool.rb index 9c3d2a29..37ca5ccb 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -20,6 +20,7 @@ def post(execution) Concurrent::Promises.future_on(executor, execution) do |thread_execution| wrap_in_app_executor do + # Worker Lifecycle 8 - Perform the job in the thread pool thread_execution.perform ensure available_threads.increment diff --git a/lib/solid_queue/processes/poller.rb b/lib/solid_queue/processes/poller.rb index 75df6104..6e9d7316 100644 --- a/lib/solid_queue/processes/poller.rb +++ b/lib/solid_queue/processes/poller.rb @@ -18,6 +18,7 @@ def metadata private def run + # Worker Lifecycle 5 - Start loop start_loop end @@ -26,6 +27,7 @@ def start_loop break if shutting_down? delay = wrap_in_app_executor do + # Worker Lifecycle 6 - Start polling poll end diff --git a/lib/solid_queue/processes/runnable.rb b/lib/solid_queue/processes/runnable.rb index 33b441f6..02b166a5 100644 --- a/lib/solid_queue/processes/runnable.rb +++ b/lib/solid_queue/processes/runnable.rb @@ -7,9 +7,11 @@ module Runnable attr_writer :mode def start + # Worker Lifecycle 3 - We start the worker lifecycle by booting boot if running_async? + # Worker Lifecycle 4 - Start running @thread = create_thread { run } else run diff --git a/lib/solid_queue/supervisor.rb b/lib/solid_queue/supervisor.rb index 7d010593..121b96c6 100644 --- a/lib/solid_queue/supervisor.rb +++ b/lib/solid_queue/supervisor.rb @@ -81,6 +81,7 @@ def start_process(configured_process) end pid = fork do + # Worker Lifecycle 1 - For each process the supervisor will start the worker lifecycle process_instance.start end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index e036a5fd..783d0419 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -16,6 +16,7 @@ def initialize(**options) # Ensure that the queues array is deep frozen to prevent accidental modification @queues = Array(options[:queues]).map(&:freeze).freeze + # Worker Lifecycle 2 - Setup up pool based on the number of threads set. @pool = Pool.new(options[:threads], on_idle: -> { wake_up }) super(**options) @@ -27,6 +28,7 @@ def metadata private def poll + # Worker Lifecycle 7 - Claim executions from the queues and post them to the thread pool claim_executions.then do |executions| executions.each do |execution| pool.post(execution)