Introduce poll-interval-variance option

Author: tomgi

Gemfile

@@ -26,6 +26,8 @@ group :test do
 
   gem 'pry'
   gem 'pg_examiner', '~> 0.5.2'
+
+  gem 'timecop', '~> 0.9.10'
 end
 
 gemspec

lib/que/command_line_interface.rb

@@ -18,14 +18,15 @@ def parse(
         default_require_file: RAILS_ENVIRONMENT_FILE
       )
 
-        options           = {}
-        queues            = []
-        log_level         = 'info'
-        log_internals     = false
-        poll_interval     = 5
-        connection_url    = nil
-        worker_count      = nil
-        worker_priorities = nil
+        options                = {}
+        queues                 = []
+        log_level              = 'info'
+        log_internals          = false
+        poll_interval          = 5
+        poll_interval_variance = 0
+        connection_url         = nil
+        worker_count           = nil
+        worker_priorities      = nil
 
         parser =
           OptionParser.new do |opts|
@@ -50,6 +51,15 @@ def parse(
               poll_interval = i
             end
 
+            opts.on(
+              '-j',
+              '--poll-interval-variance [INTERVAL]',
+              Float,
+              "Set maximum variance in poll interval, in seconds (default: 0)",
+            ) do |j|
+              poll_interval_variance = j.to_f
+            end
+
             opts.on(
               '--listen [LISTEN]',
               String,
@@ -232,7 +242,8 @@ def parse(
           options[:queues] = queues_hash
         end
 
-        options[:poll_interval] = poll_interval
+        options[:poll_interval]          = poll_interval
+        options[:poll_interval_variance] = poll_interval_variance
 
         locker =
           begin

lib/que/locker.rb

@@ -33,7 +33,7 @@ class << self
     }
 
   class Locker
-    attr_reader :thread, :workers, :job_buffer, :locks, :queues, :poll_interval
+    attr_reader :thread, :workers, :job_buffer, :locks, :queues, :poll_interval, :poll_interval_variance
 
     MESSAGE_RESOLVERS = {}
     RESULT_RESOLVERS  = {}
@@ -47,22 +47,24 @@ class Locker
     RESULT_RESOLVERS[:job_finished] =
       -> (messages) { finish_jobs(messages.map{|m| m.fetch(:metajob)}) }
 
-    DEFAULT_POLL_INTERVAL       = 5.0
-    DEFAULT_WAIT_PERIOD         = 50
-    DEFAULT_MAXIMUM_BUFFER_SIZE = 8
-    DEFAULT_WORKER_PRIORITIES   = [10, 30, 50, nil, nil, nil].freeze
+    DEFAULT_POLL_INTERVAL          = 5.0
+    DEFAULT_POLL_INTERVAL_VARIANCE = 0.0
+    DEFAULT_WAIT_PERIOD            = 50
+    DEFAULT_MAXIMUM_BUFFER_SIZE    = 8
+    DEFAULT_WORKER_PRIORITIES      = [10, 30, 50, nil, nil, nil].freeze
 
     def initialize(
-      queues:              [Que.default_queue],
-      connection_url:      nil,
-      listen:              true,
-      poll:                true,
-      poll_interval:       DEFAULT_POLL_INTERVAL,
-      wait_period:         DEFAULT_WAIT_PERIOD,
-      maximum_buffer_size: DEFAULT_MAXIMUM_BUFFER_SIZE,
-      worker_priorities:   DEFAULT_WORKER_PRIORITIES,
-      on_worker_start:     nil,
-      pidfile:             nil
+      queues:                 [Que.default_queue],
+      connection_url:         nil,
+      listen:                 true,
+      poll:                   true,
+      poll_interval:          DEFAULT_POLL_INTERVAL,
+      poll_interval_variance: DEFAULT_POLL_INTERVAL_VARIANCE,
+      wait_period:            DEFAULT_WAIT_PERIOD,
+      maximum_buffer_size:    DEFAULT_MAXIMUM_BUFFER_SIZE,
+      worker_priorities:      DEFAULT_WORKER_PRIORITIES,
+      on_worker_start:        nil,
+      pidfile:                nil
     )
 
       # Sanity-check all our arguments, since some users may instantiate Locker
@@ -71,6 +73,7 @@ def initialize(
       Que.assert [TrueClass, FalseClass], poll
 
       Que.assert Numeric, poll_interval
+      Que.assert Numeric, poll_interval_variance
       Que.assert Numeric, wait_period
 
       Que.assert Array, worker_priorities
@@ -94,20 +97,22 @@ def initialize(
 
       Que.internal_log :locker_instantiate, self do
         {
-          queues:              queues,
-          listen:              listen,
-          poll:                poll,
-          poll_interval:       poll_interval,
-          wait_period:         wait_period,
-          maximum_buffer_size: maximum_buffer_size,
-          worker_priorities:   worker_priorities,
+          queues:                 queues,
+          listen:                 listen,
+          poll:                   poll,
+          poll_interval:          poll_interval,
+          poll_interval_variance: poll_interval_variance,
+          wait_period:            wait_period,
+          maximum_buffer_size:    maximum_buffer_size,
+          worker_priorities:      worker_priorities,
         }
       end
 
       # Local cache of which advisory locks are held by this connection.
       @locks = Set.new
 
       @poll_interval = poll_interval
+      @poll_interval_variance = poll_interval_variance
 
       if queues.is_a?(Hash)
         @queue_names = queues.keys
@@ -204,9 +209,10 @@ def initialize(
               if poll
                 @queues.map do |queue_name, interval|
                   Poller.new(
-                    connection:    @connection,
-                    queue:         queue_name,
-                    poll_interval: interval,
+                    connection:             @connection,
+                    queue:                  queue_name,
+                    poll_interval:          interval,
+                    poll_interval_variance: poll_interval_variance,
                   )
                 end
               end

lib/que/poller.rb

@@ -116,25 +116,32 @@ class Poller
       :connection,
       :queue,
       :poll_interval,
+      :poll_interval_variance,
       :last_polled_at,
-      :last_poll_satisfied
+      :last_poll_satisfied,
+      :next_poll_at
 
     def initialize(
       connection:,
       queue:,
-      poll_interval:
+      poll_interval:,
+      poll_interval_variance:
     )
-      @connection          = connection
-      @queue               = queue
-      @poll_interval       = poll_interval
+      @connection             = connection
+      @queue                  = queue
+      @poll_interval          = poll_interval
+      @poll_interval_variance = poll_interval_variance
+
       @last_polled_at      = nil
       @last_poll_satisfied = nil
+      @next_poll_at        = Time.now
 
       Que.internal_log :poller_instantiate, self do
         {
-          backend_pid:   connection.backend_pid,
-          queue:         queue,
-          poll_interval: poll_interval,
+          backend_pid:            connection.backend_pid,
+          queue:                  queue,
+          poll_interval:          poll_interval,
+          poll_interval_variance: poll_interval_variance,
         }
       end
     end
@@ -158,31 +165,36 @@ def poll(
 
       @last_polled_at      = Time.now
       @last_poll_satisfied = poll_satisfied?(priorities, jobs)
+      @next_poll_at        = last_polled_at +
+                               poll_interval +
+                               rand(-poll_interval_variance..poll_interval_variance)
 
       Que.internal_log :poller_polled, self do
         {
-          queue:        @queue,
-          locked:       jobs.count,
-          priorities:   priorities,
-          held_locks:   held_locks.to_a,
-          newly_locked: jobs.map { |key| key.fetch(:id) },
+          queue:               @queue,
+          locked:              jobs.count,
+          priorities:          priorities,
+          held_locks:          held_locks.to_a,
+          newly_locked:        jobs.map { |key| key.fetch(:id) },
+          last_polled_at:      last_polled_at,
+          last_poll_satisfied: last_poll_satisfied,
+          next_poll_at:        next_poll_at,
         }
       end
 
       jobs.map! { |job| Metajob.new(job) }
     end
 
     def should_poll?
+      # polling is disabled for this queue
+      return false if poll_interval.nil?
+
       # Never polled before?
       last_poll_satisfied.nil? ||
       # Plenty of jobs were available last time?
       last_poll_satisfied == true ||
-      poll_interval_elapsed?
-    end
-
-    def poll_interval_elapsed?
-      return unless interval = poll_interval
-      (Time.now - last_polled_at) > interval
+      # It's due time to poll again regardless of the last poll's results?
+      next_poll_at < Time.now
     end
 
     class << self

spec/gemfiles/Gemfile-rails-6.0

@@ -20,4 +20,5 @@ group :test do
   gem 'minitest-hooks',   '1.4.0'
   gem 'pry'
   gem 'pg_examiner', '~> 0.5.2'
+  gem 'timecop', '~> 0.9.10'
 end

spec/gemfiles/Gemfile-rails-6.1

@@ -20,4 +20,5 @@ group :test do
   gem 'minitest-hooks',   '1.4.0'
   gem 'pry'
   gem 'pg_examiner', '~> 0.5.2'
+  gem 'timecop', '~> 0.9.10'
 end

spec/gemfiles/Gemfile-rails-7.0

@@ -20,4 +20,5 @@ group :test do
   gem 'minitest-hooks',   '1.4.0'
   gem 'pry'
   gem 'pg_examiner', '~> 0.5.2'
+  gem 'timecop', '~> 0.9.10'
 end

spec/gemfiles/Gemfile-rails-7.1

@@ -20,4 +20,5 @@ group :test do
   gem 'minitest-hooks',   '1.4.0'
   gem 'pry'
   gem 'pg_examiner', '~> 0.5.2'
+  gem 'timecop', '~> 0.9.10'
 end

spec/que/command_line_interface_spec.rb

@@ -188,6 +188,7 @@ def write_file
     def assert_locker_instantiated(
       worker_priorities: [10, 30, 50, nil, nil, nil],
       poll_interval: 5,
+      poll_interval_variance: 0.0,
       listen: true,
       wait_period: 50,
       queues: ['default'],
@@ -199,13 +200,14 @@ def assert_locker_instantiated(
 
       locker_instantiate = locker_instantiates.first
 
-      assert_equal listen,              locker_instantiate[:listen]
-      assert_equal true,                locker_instantiate[:poll]
-      assert_equal queues,              locker_instantiate[:queues]
-      assert_equal poll_interval,       locker_instantiate[:poll_interval]
-      assert_equal wait_period,         locker_instantiate[:wait_period]
-      assert_equal maximum_buffer_size, locker_instantiate[:maximum_buffer_size]
-      assert_equal worker_priorities,   locker_instantiate[:worker_priorities]
+      assert_equal listen,                 locker_instantiate[:listen]
+      assert_equal true,                   locker_instantiate[:poll]
+      assert_equal queues,                 locker_instantiate[:queues]
+      assert_equal poll_interval,          locker_instantiate[:poll_interval]
+      assert_equal poll_interval_variance, locker_instantiate[:poll_interval_variance]
+      assert_equal wait_period,            locker_instantiate[:wait_period]
+      assert_equal maximum_buffer_size,    locker_instantiate[:maximum_buffer_size]
+      assert_equal worker_priorities,      locker_instantiate[:worker_priorities]
     end
 
     def assert_locker_started(
@@ -258,6 +260,14 @@ def assert_locker_started(
       end
     end
 
+    ["-j", "--poll-interval-variance"].each do |command|
+      it "with #{command} to configure the poll interval variance" do
+        assert_successful_invocation "./#{filename} #{command} 5"
+        assert_locker_instantiated(poll_interval_variance: 5)
+        assert_locker_started
+      end
+    end
+
     it "with --listen false to disable listen mode" do
       assert_successful_invocation "./#{filename} --listen false"
       assert_locker_instantiated(listen: false)

spec/que/poller_spec.rb

@@ -44,6 +44,7 @@ def poll(
         connection: override_connection || connection,
         queue: queue_name,
         poll_interval: 5,
+        poll_interval_variance: 0,
       )
 
     Que::Poller.setup(override_connection || connection)
@@ -245,6 +246,7 @@ def assert_poll(priorities:, locked:)
         connection: connection,
         queue: 'default',
         poll_interval: 5,
+        poll_interval_variance: 0,
       )
     end
 
@@ -275,14 +277,15 @@ def assert_poll(priorities:, locked:)
       assert_equal false, poller.should_poll?
     end
 
-    it "should be false if the number of jobs returned from the last poll was less than the lowest priority request, but the poll_interval has elapsed" do
+    it "should be true if the number of jobs returned from the last poll was less than the lowest priority request, but the poll_interval has elapsed" do
       job_ids = 5.times.map { Que::Job.enqueue.que_attrs[:id] }
 
       result = poller.poll(priorities: { 500 => 7 }, held_locks: Set.new)
       assert_equal job_ids, result.map(&:id)
 
-      poller.instance_variable_set(:@last_polled_at, Time.now - 30)
-      assert_equal true, poller.should_poll?
+      Timecop.freeze(Time.now + 30) do
+        assert_equal true, poller.should_poll?
+      end
     end
   end
 end