diff --git a/lib/async/job/processor/redis/processing_list.rb b/lib/async/job/processor/redis/processing_list.rb index 53fafb3..009a19d 100644 --- a/lib/async/job/processor/redis/processing_list.rb +++ b/lib/async/job/processor/redis/processing_list.rb @@ -3,6 +3,8 @@ # Released under the MIT License. # Copyright, 2024-2025, by Samuel Williams. +require "kernel/sync" + module Async module Job module Processor @@ -75,6 +77,10 @@ def initialize(client, key, id, ready_list, job_store) @complete = @client.script(:load, COMPLETE) @complete_count = 0 + @task = nil + @mutex = Mutex.new + @condition = ConditionVariable.new + @stop_requested = false end # @attribute [String] The base Redis key for this processing list. @@ -129,26 +135,83 @@ def requeue(start_time, delay, factor) return count end - # Start the background heartbeat and abandoned job recovery task. + # Start the background heartbeat and abandoned job recovery thread. # @parameter delay [Integer] The heartbeat update interval in seconds. # @parameter factor [Integer] The heartbeat expiration factor. - # @parameter parent [Async::Task] The parent task to run the background loop in. - # @returns [Async::Task] The background processing task. - def start(delay: 5, factor: 2, parent: Async::Task.current) + # @returns [Thread | false] The background processing thread, or `false` if already started. + def start(delay: 5, factor: 2) + @mutex.synchronize do + return false if @task + + @stop_requested = false + @task = Thread.new do + Thread.current.report_on_exception = false + Thread.current.name = "#{self.class.name}:#{@id}" if Thread.current.respond_to?(:name=) + + run_heartbeat_loop(delay: delay, factor: factor) + rescue => error + Console.error(self, "Heartbeat loop failed!", exception: error) + raise + ensure + @mutex.synchronize do + @task = nil if @task.equal?(Thread.current) + end + end + end + end + + # Stop the background heartbeat and abandoned job recovery thread. + def stop + task = @mutex.synchronize do + @stop_requested = true + @condition.broadcast + + @task + end + + task&.join unless task == Thread.current + end + + private + + def run_heartbeat_loop(delay:, factor:) start_time = Time.now.to_f + client = Async::Redis::Client.new(@client.endpoint) + requeue = Sync do + client.script(:load, REQUEUE) + end - parent.async do |task| - while true - task.defer_stop do - count = self.requeue(start_time, delay, factor) - - if count > 0 - Console.warn(self, "Requeued #{count} abandoned jobs.") - end - end - - sleep(delay) + loop do + count = Sync do + requeue_with(client, requeue, start_time, delay, factor) end + + if count > 0 + Console.warn(self, "Requeued #{count} abandoned jobs.") + end + + break if wait_for_stop(delay) + end + ensure + Sync do + client&.close + end + end + + def requeue_with(client, requeue, start_time, delay, factor) + uptime = (Time.now.to_f - start_time).round(2) + expiry = (delay*factor).ceil + client.set(@heartbeat_key, JSON.dump(uptime: uptime), seconds: expiry) + client.evalsha(requeue, 2, @key, @ready_list.key) + end + + def wait_for_stop(delay) + @mutex.synchronize do + return true if @stop_requested + + @condition.wait(@mutex, delay) + + @stop_requested end end end diff --git a/lib/async/job/processor/redis/server.rb b/lib/async/job/processor/redis/server.rb index d3ead22..3499643 100644 --- a/lib/async/job/processor/redis/server.rb +++ b/lib/async/job/processor/redis/server.rb @@ -28,8 +28,10 @@ class Server < Generic # @parameter prefix [String] The Redis key prefix for job data. # @parameter coder [Async::Job::Coder] The job serialization codec. # @parameter resolution [Integer] The resolution in seconds for delayed job processing. + # @parameter heartbeat_interval [Integer] The interval in seconds between processing heartbeats. + # @parameter heartbeat_factor [Integer] The factor used to calculate heartbeat expiry. # @parameter parent [Async::Task] The parent task for background processing. - def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, resolution: 10, parent: nil) + def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, resolution: 10, heartbeat_interval: 5, heartbeat_factor: 2, parent: nil) super(delegate) @id = SecureRandom.uuid @@ -37,6 +39,8 @@ def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, res @prefix = prefix @coder = coder @resolution = resolution + @heartbeat_interval = heartbeat_interval + @heartbeat_factor = heartbeat_factor @job_store = JobStore.new(@client, "#{@prefix}:jobs") @delayed_jobs = DelayedJobs.new(@client, "#{@prefix}:delayed") @@ -73,7 +77,7 @@ def start @delayed_jobs.start(@ready_list, resolution: @resolution) # Start the processing processor, which will move jobs to the ready processor when they are abandoned: - @processing_list.start + @processing_list.start(delay: @heartbeat_interval, factor: @heartbeat_factor) self.start! end @@ -81,6 +85,7 @@ def start # Stop the server and all background processing tasks. def stop @task&.stop + @processing_list.stop super end diff --git a/test/async/job/processor/processing_list.rb b/test/async/job/processor/processing_list.rb index ab37ac5..b2de41e 100644 --- a/test/async/job/processor/processing_list.rb +++ b/test/async/job/processor/processing_list.rb @@ -150,11 +150,12 @@ client.del("#{prefix}:processing:#{dead_server_id}") task = processing_list.start(delay: 0.1, factor: 2) + expect(task).to be_a(Thread) fetched_job = processing_list.fetch expect(fetched_job).to be == abandoned_job_id ensure - task&.stop + processing_list.stop end end end diff --git a/test/async/job/processor/server.rb b/test/async/job/processor/server.rb index a56075b..d30f9be 100644 --- a/test/async/job/processor/server.rb +++ b/test/async/job/processor/server.rb @@ -80,6 +80,12 @@ expect(server.status_string).to be == "R=0 D=0 P=0/0" server.call(job) + buffer.pop + + deadline = Time.now + 0.5 + while server.status_string != "R=0 D=0 P=0/1" && Time.now < deadline + sleep(0.01) + end expect(server.status_string).to be == "R=0 D=0 P=0/1" end