From f6acc17658236670405d05c21f14c4140b27b8c5 Mon Sep 17 00:00:00 2001 From: Stephen Anderson Date: Thu, 3 Sep 2026 20:25:59 +1000 Subject: [PATCH] Accept notified jobs for workers waiting on an empty buffer accept? decided how many notified jobs to take from the free buffer space alone, so a locker running with maximum_size: 0 accepted none of them. Its buffer is always empty, so there is never free space and never anything buffered to displace, and every job waited for the next poll instead - up to poll_interval late, or indefinitely with polling off. A buffer size of zero is a reasonable setting for a queue of long running jobs, where a locker that pre-locks a job the running job goes on to wait for will deadlock. Fall back to the workers currently blocked waiting for a job when the buffer has no space to offer. That is safe because push hands jobs straight to waiting workers, before it trims the buffer to maximum_size, so a job claimed by an idle worker never occupies buffer space and is never pre-locked. available_priorities already sizes the poller's fetch as the waiting workers plus the buffer space, so accept? was the one place left that ignored them. Falling back only when there is no space leaves accept? behaving exactly as it did before whenever the buffer has room to fill. With a full buffer it now also takes what the waiting workers will accept on top of what it can displace; push places those and returns the remainder for push_jobs to unlock, as it does for any other overage. The count is priority aware, since a worker only accepts jobs its threshold admits. metajobs is sorted and a threshold admits every job at least as important as the ones it rejects, so the eligible worker count only falls as the walk advances and the first job that cannot be placed for certain ends it. The walk is therefore conservative rather than exact: push offers each job to the most permissive waiting worker first, so counting every placement a perfect assignment could make would accept jobs push then has to displace. --- lib/que/job_buffer.rb | 37 ++++++++++++++- spec/que/job_buffer_spec.rb | 92 +++++++++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 1 deletion(-) diff --git a/lib/que/job_buffer.rb b/lib/que/job_buffer.rb index 711721e8..1fba02cb 100644 --- a/lib/que/job_buffer.rb +++ b/lib/que/job_buffer.rb @@ -77,7 +77,7 @@ def accept?(metajobs) sync do return [] if _stopping? - start_index = _buffer_space + start_index = _free_capacity(metajobs) final_index = metajobs.length - 1 return metajobs if start_index > final_index @@ -157,6 +157,41 @@ def _buffer_space maximum_size - _size end + # How many of these metajobs can be taken on without evicting anything + # that's already buffered. That's the free buffer space, except when there + # is none at all, in which case it's the number of workers currently blocked + # waiting for a job they'd accept - push hands jobs straight to waiting + # workers before it trims the buffer to maximum_size, so a job claimed by an + # idle worker never occupies buffer space. Without falling back to them a + # buffer with maximum_size: 0 rejects every job offered, even when every + # worker is idle, leaving them to be found by polling. + # + # metajobs is sorted, and a worker's priority threshold admits every job at + # least as important as the ones it rejects, so the eligible worker count + # only falls as we advance and the first job we can't be sure of placing + # ends the run. + def _free_capacity(metajobs) + space = _buffer_space + return space if space > 0 + + thresholds = _waiting_priorities + index = 0 + + while index < metajobs.length && index < _sufficient_count(thresholds, metajobs[index]) + index += 1 + end + + index + end + + def _waiting_priorities + priority_queues.flat_map { |priority, pq| Array.new(pq.waiting_count, priority) } + end + + def _sufficient_count(thresholds, metajob) + thresholds.count { |threshold| metajob.priority_sufficient?(threshold) } + end + def pop(count) @array.pop(count) end diff --git a/spec/que/job_buffer_spec.rb b/spec/que/job_buffer_spec.rb index d9dfca4b..2f50dbeb 100644 --- a/spec/que/job_buffer_spec.rb +++ b/spec/que/job_buffer_spec.rb @@ -271,6 +271,98 @@ def new_metajob(key) assert_equal [], job_buffer.accept?(job_array) end + + describe "when the buffer has no spare space" do + let(:maximum_size) { 0 } + + it "should not accept a job when no worker is waiting for one" do + assert_equal [], job_buffer.accept?([job_array[0]]) + end + + it "should accept no more jobs than there are workers waiting" do + job_buffer # Pre-initialize to avoid race conditions. + + 2.times { Thread.new { job_buffer.shift } } + sleep_until_equal(2) { job_buffer.waiting_count } + + assert_equal job_array[0..1], job_buffer.accept?(job_array[0..2]) + end + + it "should pass an accepted job to the waiting worker without buffering it" do + job_buffer # Pre-initialize to avoid race conditions. + + t = Thread.new { Thread.current[:job] = job_buffer.shift } + sleep_until_equal(1) { job_buffer.waiting_count } + + assert_equal [job_array[0]], job_buffer.accept?([job_array[0]]) + assert_nil job_buffer.push(job_array[0]) + sleep_until_equal(false) { t.status } + + assert_equal job_array[0], t[:job] + assert_equal [], job_buffer.to_a + end + end + + describe "when the buffer has spare space" do + let(:maximum_size) { 2 } + + it "should not accept more jobs than the buffer has space for, whatever workers are waiting" do + job_buffer # Pre-initialize to avoid race conditions. + + 4.times { Thread.new { job_buffer.shift } } + sleep_until_equal(4) { job_buffer.waiting_count } + + assert_equal job_array[0..1], job_buffer.accept?(job_array[0..3]) + end + end + + describe "when the buffer is full" do + before do + job_buffer.push(*8.times.map { |i| new_metajob(priority: 30, run_at: old, id: i + 1) }) + + @waiter = Thread.new { job_buffer.shift(10) } + sleep_until_equal(1) { job_buffer.waiting_count } + end + + after do + job_buffer.stop + @waiter.join + end + + it "should accept a job for the waiting worker on top of the ones it can displace" do + jobs = 10.times.map { |i| new_metajob(priority: 5, run_at: now, id: i + 10) } + + assert_equal 9, job_buffer.accept?(jobs).length + end + end + + describe "when the only worker waiting has a priority threshold" do + let :job_buffer do + Que::JobBuffer.new(maximum_size: 0, priorities: [10]) + end + + before do + @waiter = Thread.new { job_buffer.shift(10) } + sleep_until_equal(1) { job_buffer.waiting_count } + end + + after do + job_buffer.stop + @waiter.join + end + + it "should accept a job the worker's priority admits" do + job = new_metajob(priority: 5, run_at: now, id: 1) + + assert_equal [job], job_buffer.accept?([job]) + end + + it "should not accept a job the worker's priority does not admit" do + job = new_metajob(priority: 50, run_at: now, id: 1) + + assert_equal [], job_buffer.accept?([job]) + end + end end describe "buffer_space" do