Merge pull request #352 from que-rb/revert-348-make-poller-more-easily-satisfied

Revert "Continue polling for jobs when only some of the waiting worker priorities will be fully utilised"

Author: ZimbiX

lib/que/poller.rb

@@ -146,6 +146,8 @@ def poll(
 
       return unless should_poll?
 
+      expected_count = priorities.inject(0){|s,(_,c)| s + c}
+
       jobs =
         connection.execute_prepared(
           :poll_jobs,
@@ -157,7 +159,7 @@ def poll(
         )
 
       @last_polled_at      = Time.now
-      @last_poll_satisfied = any_priority_satisfied?(priorities, jobs)
+      @last_poll_satisfied = expected_count == jobs.count
 
       Que.internal_log :poller_polled, self do
         {
@@ -263,21 +265,5 @@ def cleanup(connection)
         SQL
       end
     end
-
-    private
-
-    def any_priority_satisfied?(priorities, jobs)
-      job_priorities = jobs.map { |job| job.fetch(:priority) }.sort
-      worker_job_counts = Hash.new(0)
-      priorities.any? do |worker_priority, waiting_workers_count|
-        waiting_workers_count.times do
-          return false if job_priorities.empty?
-          break if job_priorities.first > worker_priority
-          job_priorities.shift
-          worker_job_counts[worker_priority] += 1
-        end
-        worker_job_counts[worker_priority] == waiting_workers_count
-      end
-    end
   end
 end

lib/que/poller.spec.rb

@@ -255,61 +255,29 @@ def assert_poll(priorities:, locked:)
       assert poller.should_poll?
     end
 
-    it "should be true if the jobs returned from the last poll satisfied all priority requests" do
-      job_ids_p10 = 3.times.map { Que::Job.enqueue(job_options: { priority: 10 }).que_attrs[:id] }
-      job_ids_p20 = 2.times.map { Que::Job.enqueue(job_options: { priority: 20 }).que_attrs[:id] }
+    it "should be true if the last poll returned a full complement of jobs" do
+      jobs = 5.times.map { Que::Job.enqueue }
 
-      result = poller.poll(priorities: { 10 => 3, 20 => 2 }, held_locks: Set.new)
-      assert_equal (job_ids_p10 + job_ids_p20), result.map(&:id)
+      result = poller.poll(priorities: {500 => 3}, held_locks: Set.new)
+      assert_equal 3, result.length
 
       assert_equal true, poller.should_poll?
     end
 
-    it "should be true if the jobs returned from the last poll satisfied any priority request" do
-      job_ids_p10 = 2.times.map { Que::Job.enqueue(job_options: { priority: 10 }).que_attrs[:id] }
-      job_ids_p20 = 2.times.map { Que::Job.enqueue(job_options: { priority: 20 }).que_attrs[:id] }
+    it "should be false if the last poll didn't return a full complement of jobs" do
+      jobs = 5.times.map { Que::Job.enqueue }
 
-      result = poller.poll(priorities: { 10 => 2, 20 => 1 }, held_locks: Set.new)
-      assert_equal (job_ids_p10 + [job_ids_p20.first]), result.map(&:id)
-
-      assert_equal true, poller.should_poll?
-    end
-
-    it "should be true if the jobs returned from the last poll satisfied any priority request and were slightly higher priority than each priority requested" do
-      job_ids_p10 = 2.times.map { Que::Job.enqueue(job_options: { priority: 10 }).que_attrs[:id] }
-      job_ids_p20 = 2.times.map { Que::Job.enqueue(job_options: { priority: 20 }).que_attrs[:id] }
-
-      result = poller.poll(priorities: { 11 => 2, 21 => 1 }, held_locks: Set.new)
-      assert_equal (job_ids_p10 + [job_ids_p20.first]), result.map(&:id)
-
-      assert_equal true, poller.should_poll?
-    end
-
-    it "should be true if the jobs returned from the last poll satisfied any priority request and a lower priority request was upgraded to high priority" do
-      job_ids_p10 = 5.times.map { Que::Job.enqueue(job_options: { priority: 10 }).que_attrs[:id] }
-      job_ids_p20 = 2.times.map { Que::Job.enqueue(job_options: { priority: 20 }).que_attrs[:id] }
-
-      result = poller.poll(priorities: { 10 => 3, 20 => 2 }, held_locks: Set.new)
-      assert_equal job_ids_p10, result.map(&:id)
-
-      assert_equal true, poller.should_poll?
-    end
-
-    it "should be false if the jobs returned from the last poll didn't satisfy any priority request" do
-      job_ids_p10 = 5.times.map { Que::Job.enqueue(job_options: { priority: 10 }).que_attrs[:id] }
-      job_ids_p20 = 2.times.map { Que::Job.enqueue(job_options: { priority: 20 }).que_attrs[:id] }
-
-      result = poller.poll(priorities: { 10 => 6, 20 => 3 }, held_locks: Set.new)
-      assert_equal (job_ids_p10 + job_ids_p20), result.map(&:id)
+      result = poller.poll(priorities: {500 => 7}, held_locks: Set.new)
+      assert_equal 5, result.length
 
       assert_equal false, poller.should_poll?
     end
 
-    it "should be true if the jobs returned from the last poll didn't satisfy any priority request, but the poll_interval has elapsed" do
-      job_ids = 5.times.map { Que::Job.enqueue.que_attrs[:id] }
+    it "should be true if the last poll didn't return a full complement of jobs, but the poll_interval has elapsed" do
+      jobs = 5.times.map { Que::Job.enqueue }
 
-      result = poller.poll(priorities: { 500 => 7 }, held_locks: Set.new)
-      assert_equal job_ids, result.map(&:id)
+      result = poller.poll(priorities: {500 => 7}, held_locks: Set.new)
+      assert_equal 5, result.length
 
       poller.instance_variable_set(:@last_polled_at, Time.now - 30)
       assert_equal true, poller.should_poll?