diff --git a/lib/sidekiq_unique_jobs/orphans/ruby_reaper.rb b/lib/sidekiq_unique_jobs/orphans/ruby_reaper.rb index 92af99ca..7c911be0 100644 --- a/lib/sidekiq_unique_jobs/orphans/ruby_reaper.rb +++ b/lib/sidekiq_unique_jobs/orphans/ruby_reaper.rb @@ -70,8 +70,6 @@ def initialize(conn) # @return [Integer] the number of reaped locks # def call - return if queues_very_full? - BatchDelete.call(expired_digests, conn) BatchDelete.call(orphans, conn) end @@ -185,12 +183,7 @@ def active?(digest) # rubocop:disable Metrics/MethodLength, Metrics/CyclomaticCo procs.sort.each do |key| valid, workers = conn.pipelined do |pipeline| - # TODO: Remove the if statement in the future - if pipeline.respond_to?(:exists?) - pipeline.exists?(key) - else - pipeline.exists(key) - end + pipeline.exists(key) pipeline.hgetall("#{key}:work") end @@ -286,7 +279,7 @@ def queues_very_full? # @return [false] when missing # def in_sorted_set?(key, digest) - conn.zscan(key, match: "*#{digest}*", count: 1).to_a.any? + conn.zscan(key, match: "*#{digest}*", count: 1000).to_a.any? end end # rubocop:enable Metrics/ClassLength diff --git a/spec/performance/lock_digest_spec.rb b/spec/performance/lock_digest_spec.rb index 49f8bf8f..5a6620fc 100644 --- a/spec/performance/lock_digest_spec.rb +++ b/spec/performance/lock_digest_spec.rb @@ -1,11 +1,11 @@ # frozen_string_literal: true RSpec.describe SidekiqUniqueJobs::LockDigest, perf: true do - let(:lock_digest) { described_class.new(item) } - let(:job_class) { UntilExecutedJob } - let(:class_name) { worker_class.to_s } - let(:queue) { "myqueue" } - let(:args) { [[1, 2]] } + let(:service) { described_class.new(item) } + let(:job_class) { UntilExecutedJob } + let(:class_name) { job_class.to_s } + let(:queue) { "myqueue" } + let(:args) { [[1, 2]] } let(:item) do { "class" => class_name, @@ -15,7 +15,7 @@ end describe "#lock_digest" do - subject(:lock_digest) { lock_digest.lock_digest } + subject(:lock_digest) { service.lock_digest } it "performs in under 0.1 ms" do expect { lock_digest }.to perform_under(0.1).ms diff --git a/spec/performance/locksmith_spec.rb b/spec/performance/locksmith_spec.rb index 90b57737..8cd41ade 100644 --- a/spec/performance/locksmith_spec.rb +++ b/spec/performance/locksmith_spec.rb @@ -42,10 +42,10 @@ it "unlocks in under 1 ms" do locksmith_one.lock - expect { locksmith_one.unlock }.to perform_under(1).ms + expect { locksmith_one.unlock }.to perform_under(2).ms end it "locks with expected allocations" do - expect { locksmith_one.lock {} }.to perform_allocation(Array => 12_640, Hash => 13_888).bytes + expect { locksmith_one.lock {} }.to perform_allocation({ Array => 12_640, Hash => 13_888 }).bytes end end diff --git a/spec/performance/orphans/ruby_reaper_spec.rb b/spec/performance/orphans/ruby_reaper_spec.rb new file mode 100644 index 00000000..551ebc4d --- /dev/null +++ b/spec/performance/orphans/ruby_reaper_spec.rb @@ -0,0 +1,46 @@ +# frozen_string_literal: true + +RSpec.describe SidekiqUniqueJobs::Orphans::RubyReaper do + let(:service) { redis { |conn| described_class.new(conn) } } + let(:digest) { "uniquejobs:digest" } + let(:job_id) { "job_id" } + let(:item) { raw_item } + let(:lock) { SidekiqUniqueJobs::Lock.create(digest, job_id, lock_info) } + let(:raw_item) { { "class" => MyUniqueJob, "args" => [], "jid" => job_id, "lock_digest" => digest } } + let(:lock_info) do + { + "job_id" => job_id, + "limit" => 1, + "lock" => :while_executing, + "time" => now_f, + "timeout" => nil, + "ttl" => nil, + "lock_args" => [], + "worker" => "MyUniqueJob", + } + end + + describe "#in_sorted_set?" do + subject(:in_sorted_set?) { service.send(:in_sorted_set?, "retry", digest) } + + context "when retried" do + let(:item) { raw_item.merge("retry_count" => 2, "failed_at" => now_f) } + + context "with job in retry", perf: true do + before do + puts "#{Time.now} - Adding 100_000 to retry queue" + + 1_000_000.times do |i| + zadd("retry", (Time.now.to_f - i).to_s, dump_json(item.except("lock_digest"))) + end + + zadd("retry", (Time.now.to_f + 200_000).to_s, dump_json(item)) + + puts "#{Time.now} - Done adding to retry queue" + end + + it { expect { service.send(:in_sorted_set?, "retry", digest) }.to perform_under(2).ms } + end + end + end +end diff --git a/spec/support/rspec_benchmark.rb b/spec/support/rspec_benchmark.rb index 61a0c7b6..9355487a 100644 --- a/spec/support/rspec_benchmark.rb +++ b/spec/support/rspec_benchmark.rb @@ -9,6 +9,8 @@ end RSpec::Benchmark.configure do |config| + config.run_in_subprocess = false + config.disable_gc = false config.samples = 10 end rescue LoadError