add flag for batch enqueue delayed

Author: onyxraven

.gitignore

@@ -16,4 +16,5 @@ nbproject
 .env
 .env.*
 /nul
-vendor/
+vendor/
+.vscode

README.md

@@ -129,24 +129,18 @@ Both the Rake task and standalone executable support the following
 environment variables:
 
 * `APP_NAME` - Application name used in procline (`$0`) (default empty)
-* `BACKGROUND` - [Run in the background](#running-in-the-background) if
-non-empty (via `Process.daemon`, if supported) (default `false`)
-* `DYNAMIC_SCHEDULE` - Enables [dynamic scheduling](#dynamic-schedules)
-if non-empty (default `false`)
-* `RAILS_ENV` - Environment to use in procline (`$0`) (default empty)
-* `INITIALIZER_PATH` - Path to a Ruby file that will be loaded *before*
-requiring `resque` and `resque/scheduler` (default empty).
-* `RESQUE_SCHEDULER_INTERVAL` - Interval in seconds for checking if a
-scheduled job must run (coerced with `Kernel#Float()`) (default `5`)
+* `BACKGROUND` - [Run in the background](#running-in-the-background) if non-empty (via `Process.daemon`, if supported) (default `false`)
+* `DELAYED_REQUEUE_BATCH_SIZE` - Set the delayed job batch size if enabled (default `100`). If `<= 1`, this disables batching.
+* `DISABLE_DELAYED_REQUEUE_BATCH` - Disable batched delayed job queuing (default `false`) - [See section below on consequences](#batched-delayed-job-and-resque-enqueue-hooks)
+* `DYNAMIC_SCHEDULE` - Enables [dynamic scheduling](#dynamic-schedules) if non-empty (default `false`)
+* `INITIALIZER_PATH` - Path to a Ruby file that will be loaded *before* requiring `resque` and `resque/scheduler` (default empty).
 * `LOGFILE` - Log file name (default empty, meaning `$stdout`)
-* `LOGFORMAT` - Log output format to use (either `'text'`, `'json'` or `'logfmt'`,
-default `'text'`)
+* `LOGFORMAT` - Log output format to use (either `'text'`, `'json'` or `'logfmt'`, default `'text'`)
 * `PIDFILE` - If non-empty, write process PID to file (default empty)
-* `QUIET` - Silence most output if non-empty (equivalent to a level of
-`MonoLogger::FATAL`, default `false`)
-* `VERBOSE` - Maximize log verbosity if non-empty (equivalent to a level
-of `MonoLogger::DEBUG`, default `false`)
-
+* `QUIET` - Silence most output if non-empty (equivalent to a level of `MonoLogger::FATAL`, default `false`)
+* `RAILS_ENV` - Environment to use in procline (`$0`) (default empty)
+* `RESQUE_SCHEDULER_INTERVAL` - Interval in seconds for checking if a scheduled job must run (coerced with `Kernel#Float()`) (default `5`)
+* `VERBOSE` - Maximize log verbosity if non-empty (equivalent to a level of `MonoLogger::DEBUG`, default `false`)
 
 ### Resque Pool integration
 
@@ -755,6 +749,24 @@ This table explains the version requirements for rufus-scheduler
 | `~> 4.0`         | `~> 3.0`        |
 | `< 4.0`          | `~> 2.0`        |
 
+##### Batched delayed job and resque enqueue hooks
+
+Batching delayed job queuing can speed up when per-second job counts grows,
+avoiding situations that may cause delayed enqueues to fall behind. This
+batching wraps enqueues in a `multi` pipeline, making far fewer roundtrips to
+the server.
+
+However, in `redis` gem `>= 4.0`, any operations to redis within the `multi`
+block must use the multi handle so that the actions are captured. Resque's hooks
+do not currently have a way to pass this around, and so compatibility with other
+resque plugins or hooks which access redis at enqueue time is impacted with
+batch mode. In these cases, you should consider disabling the batching by setting
+the `DISABLE_DELAYED_REQUEUE_BATCH` environment variable to `true`.
+
+Detecting when this occurs can be tricky, you must watch for logs
+emitted by your `resque-scheduler` process such as `Redis::CommandError: ERR
+MULTI calls can not be nested` or `NoMethodError: undefined method nil? for
+<Redis::Future`, and delayed jobs you expect would not be enqueued.
 
 ### Contributing
 

lib/resque/scheduler.rb

@@ -204,25 +204,36 @@ def enqueue_next_item(timestamp)
         item
       end
 
+      def batch_delayed_items?
+        !disable_delayed_requeue_batches && delayed_requeue_batch_size > 1
+      end
+
       # Enqueues all delayed jobs for a timestamp
       def enqueue_delayed_items_for_timestamp(timestamp)
         count = 0
-        batch_size = delayed_requeue_batch_size
-        actual_batch_size = nil
+        batch_size = batch_delayed_items? ? delayed_requeue_batch_size : 1
 
-        log "Processing delayed items for timestamp #{timestamp}, in batches of #{batch_size}"
+        message = "Processing delayed items for timestamp #{timestamp}"
+        message += ", in batches of #{batch_size}" if batch_delayed_items?
+        log message
 
         loop do
+          actual_batch_size = 0
+
           handle_shutdown do
             # Continually check that it is still the master
             if am_master
-              actual_batch_size = enqueue_items_in_batch_for_timestamp(timestamp,
-                                                                       batch_size)
+              if batch_delayed_items?
+                actual_batch_size = enqueue_items_in_batch_for_timestamp(timestamp, batch_size)
+                log "queued batch of #{actual_batch_size} jobs" if actual_batch_size != -1
+              else
+                item = enqueue_next_item(timestamp)
+                actual_batch_size = item.nil? ? 0 : 1
+              end
             end
           end
 
           count += actual_batch_size
-          log "queued #{count} jobs" if actual_batch_size != -1
 
           # continue processing until there are no more items in this
           # timestamp. If we don't have a full batch, this is the last one.
@@ -231,7 +242,7 @@ def enqueue_delayed_items_for_timestamp(timestamp)
           break if actual_batch_size < batch_size
         end
 
-        log "finished queueing #{count} total jobs for timestamp #{timestamp}" if count != -1
+        log "finished queueing #{count} total jobs for timestamp #{timestamp}"
       end
 
       def timestamp_key(timestamp)

lib/resque/scheduler/configuration.rb

@@ -65,12 +65,20 @@ def app_name
         @app_name ||= environment['APP_NAME']
       end
 
+      attr_writer :delayed_requeue_batch_size
+
       def delayed_requeue_batch_size
         @delayed_requeue_batch_size ||= \
           ENV['DELAYED_REQUEUE_BATCH_SIZE'].to_i if environment['DELAYED_REQUEUE_BATCH_SIZE']
         @delayed_requeue_batch_size ||= 100
       end
 
+      attr_writer :disable_delayed_requeue_batches
+
+      def disable_delayed_requeue_batches
+        @enable_delayed_requeue_batches ||= to_bool(environment['DISABLE_DELAYED_REQUEUE_BATCH'])
+      end
+
       # Amount of time in seconds to sleep between polls of the delayed
       # queue.  Defaults to 5
       attr_writer :poll_sleep_amount

test/delayed_queue_test.rb

@@ -383,60 +383,115 @@
     assert_equal(1, Resque.delayed_timestamp_peek(t, 0, 3).length)
   end
 
-  test 'enqueue_delayed_items_for_timestamp enqueues jobs in 2 batches' do
-    t = Time.now + 60
+  context "non-batch delayed item queue" do
+    batch_disabled = Resque::Scheduler.disable_delayed_requeue_batches
+    batch_size = Resque::Scheduler.delayed_requeue_batch_size
+    setup do
+      Resque::Scheduler.quiet = true
+      Resque.data_store.redis.flushall
+      Resque::Scheduler.disable_delayed_requeue_batches = true
+      Resque::Scheduler.delayed_requeue_batch_size = 1
+    end
 
-    # create 120 jobs
-    120.times { Resque.enqueue_at(t, SomeIvarJob) }
-    assert_equal(120, Resque.delayed_timestamp_size(t))
+    teardown do
+      Resque::Scheduler.disable_delayed_requeue_batches = batch_disabled
+      Resque::Scheduler.delayed_requeue_batch_size = batch_size
+    end
 
-    Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
-    assert_equal(0, Resque.delayed_timestamp_size(t))
+    test 'enqueue_delayed_items_for_timestamp enqueues jobs for the timestamp' do
+      t = Time.now + 60
 
-    # assert that the active queue is now 120
-    assert_equal(120, Resque.size(Resque.queue_from_class(SomeIvarJob)))
-  end
+      Resque::Scheduler.expects(:enqueue_items_in_batch_for_timestamp).never
 
-  test 'enqueue_delayed_items_for_timestamp enqueues jobs in one batch for the timestamp' do
-    t = Time.now + 60
+      # create 90 jobs
+      90.times { Resque.enqueue_at(t, SomeIvarJob) }
+      assert_equal(90, Resque.delayed_timestamp_size(t))
 
-    # create 90 jobs
-    90.times { Resque.enqueue_at(t, SomeIvarJob) }
-    assert_equal(90, Resque.delayed_timestamp_size(t))
+      Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
+      assert_equal(0, Resque.delayed_timestamp_size(t))
 
-    Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
-    assert_equal(0, Resque.delayed_timestamp_size(t))
+      # assert that the active queue is now 90
+      assert_equal(90, Resque.size(Resque.queue_from_class(SomeIvarJob)))
+    end
 
-    # assert that the active queue is now 90
-    assert_equal(90, Resque.size(Resque.queue_from_class(SomeIvarJob)))
+    # TODO clean up timestamp tests?
   end
 
-  # test to make sure the timestamp is cleaned up
+  context "batch delayed item queue" do
+    batch_disabled = Resque::Scheduler.disable_delayed_requeue_batches
+    batch_size = Resque::Scheduler.delayed_requeue_batch_size
+    setup do
+      Resque::Scheduler.quiet = true
+      Resque.data_store.redis.flushall
+      Resque::Scheduler.disable_delayed_requeue_batches = false
+      Resque::Scheduler.delayed_requeue_batch_size = 100
+    end
 
-  test 'enqueue_delayed_items_for_timestamp handles a watch failure' do
-    t = Time.now + 60
+    teardown do
+      Resque::Scheduler.disable_delayed_requeue_batches = batch_disabled
+      Resque::Scheduler.delayed_requeue_batch_size = batch_size
+    end
 
-    # create 100 jobs
-    100.times { Resque.enqueue_at(t, SomeIvarJob) }
-    assert_equal(100, Resque.delayed_timestamp_size(t))
+    test 'enqueue_delayed_items_for_timestamp enqueues jobs in 2 batches' do
+      t = Time.now + 60
 
-    Resque.redis.stubs(:watch).returns(nil)
+      Resque::Scheduler.expects(:enqueue_next_item).never
 
-    Resque.expects(:clean_up_timestamp).never
+      # create 120 jobs
+      120.times { Resque.enqueue_at(t, SomeIvarJob) }
+      assert_equal(120, Resque.delayed_timestamp_size(t))
 
-    Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
-  end
+      Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
+      assert_equal(0, Resque.delayed_timestamp_size(t))
 
-  test 'enqueue_delayed_items_for_timestamp cleans up a timestamp' do
-    t = Time.now + 60
+      # assert that the active queue is now 120
+      assert_equal(120, Resque.size(Resque.queue_from_class(SomeIvarJob)))
+    end
 
-    # create 100 jobs
-    100.times { Resque.enqueue_at(t, SomeIvarJob) }
-    assert_equal(100, Resque.delayed_timestamp_size(t))
+    test 'enqueue_delayed_items_for_timestamp enqueues jobs in one batch for the timestamp' do
+      t = Time.now + 60
 
-    Resque.expects(:clean_up_timestamp).once
+      Resque::Scheduler.expects(:enqueue_next_item).never
+
+      # create 90 jobs
+      90.times { Resque.enqueue_at(t, SomeIvarJob) }
+      assert_equal(90, Resque.delayed_timestamp_size(t))
+
+      Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
+      assert_equal(0, Resque.delayed_timestamp_size(t))
+
+      # assert that the active queue is now 90
+      assert_equal(90, Resque.size(Resque.queue_from_class(SomeIvarJob)))
+    end
+
+    # test to make sure the timestamp is cleaned up
+
+    test 'enqueue_delayed_items_for_timestamp handles a watch failure' do
+      t = Time.now + 60
+
+      # create 100 jobs
+      100.times { Resque.enqueue_at(t, SomeIvarJob) }
+      assert_equal(100, Resque.delayed_timestamp_size(t))
+
+      Resque.redis.stubs(:watch).returns(nil)
+
+      Resque.expects(:clean_up_timestamp).never
+
+      Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
+    end
+
+    test 'enqueue_delayed_items_for_timestamp cleans up a timestamp' do
+      t = Time.now + 60
+
+      # create 100 jobs
+      100.times { Resque.enqueue_at(t, SomeIvarJob) }
+      assert_equal(100, Resque.delayed_timestamp_size(t))
+
+      Resque.expects(:clean_up_timestamp).once
+
+      Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
+    end
 
-    Resque::Scheduler.enqueue_delayed_items_for_timestamp(t)
   end
 
   test 'enqueue_delayed_items_for_timestamp creates jobs ' \