delayed_jobs lock_with_read_ahead for postgres (#4231)

Author: stephanme

lib/cloud_controller/config_schemas/base/api_schema.rb

@@ -352,6 +352,7 @@ class ApiSchema < VCAP::Config
               queues: {
                 optional(:cc_generic) => { timeout_in_seconds: Integer }
               },
+              optional(:read_ahead) => Integer,
               optional(:enable_dynamic_job_priorities) => bool,
               optional(:app_usage_events_cleanup) => { timeout_in_seconds: Integer },
               optional(:blobstore_delete) => { timeout_in_seconds: Integer },

lib/cloud_controller/config_schemas/base/clock_schema.rb

@@ -174,6 +174,7 @@ class ClockSchema < VCAP::Config
             allow_app_ssh_access: bool,
             jobs: {
               global: { timeout_in_seconds: Integer },
+              optional(:read_ahead) => Integer,
               optional(:app_usage_events_cleanup) => { timeout_in_seconds: Integer },
               optional(:blobstore_delete) => { timeout_in_seconds: Integer },
               optional(:diego_sync) => { timeout_in_seconds: Integer },

lib/cloud_controller/config_schemas/base/worker_schema.rb

@@ -175,6 +175,7 @@ class WorkerSchema < VCAP::Config
               queues: {
                 optional(:cc_generic) => { timeout_in_seconds: Integer }
               },
+              optional(:read_ahead) => Integer,
               optional(:enable_dynamic_job_priorities) => bool,
               optional(:app_usage_events_cleanup) => { timeout_in_seconds: Integer },
               optional(:blobstore_delete) => { timeout_in_seconds: Integer },

lib/cloud_controller/db.rb

@@ -85,13 +85,17 @@ def self.load_models(db_config, logger)
 
       require 'models'
       require 'delayed_job_sequel'
+      # load monkey patch for sequel backend to support configurable job lock method (postgres only)
+      require 'delayed_job/sequel_patch'
     end
 
     def self.load_models_without_migrations_check(db_config, logger)
       connect(db_config, logger)
 
       require 'models'
       require 'delayed_job_sequel'
+      # load monkey patch for sequel backend to support configurable job lock method (postgres only)
+      require 'delayed_job/sequel_patch'
     end
   end
 end

lib/delayed_job/delayed_worker.rb

@@ -4,6 +4,9 @@
 require 'prometheus/middleware/exporter'
 
 class CloudController::DelayedWorker
+  DEFAULT_READ_AHEAD_POSTGRES = 0
+  DEFAULT_READ_AHEAD_MYSQL = Delayed::Worker::DEFAULT_READ_AHEAD
+
   def initialize(options)
     @queue_options = {
       min_priority: ENV.fetch('MIN_PRIORITY', nil),
@@ -56,6 +59,16 @@ def get_initialized_delayed_worker(config, logger)
     Delayed::Worker.max_run_time = config.get(:jobs, :global, :timeout_in_seconds) + 1
     Delayed::Worker.sleep_delay = config.get(:jobs, :global, :worker_sleep_delay_in_seconds)
     Delayed::Worker.logger = logger
+    if ::Sequel::Model.db.database_type == :mysql
+      read_ahead = config.get(:jobs, :read_ahead) || DEFAULT_READ_AHEAD_MYSQL
+      # lock for update is not configurable for mysql
+      read_ahead = DEFAULT_READ_AHEAD_MYSQL if read_ahead <= 0
+    else
+      # read_ahead 0 = lock for update (default for postgres)
+      read_ahead = config.get(:jobs, :read_ahead) || DEFAULT_READ_AHEAD_POSTGRES
+      read_ahead = DEFAULT_READ_AHEAD_POSTGRES if read_ahead < 0
+    end
+    Delayed::Worker.read_ahead = read_ahead
 
     unless @queue_options[:num_threads].nil?
       # Dynamically alias Delayed::Worker to ThreadedWorker to ensure plugins etc are working correctly

lib/delayed_job/sequel_patch.rb

@@ -0,0 +1,25 @@
+require 'delayed/backend/sequel'
+
+module Delayed
+  module Backend
+    module Sequel
+      class Job
+        # monkey patch to allow explicit configuration of job lock method
+        def self.reserve(worker, max_run_time=Worker.max_run_time)
+          ds = ready_to_run(worker.name, max_run_time)
+
+          ds = ds.filter(::Sequel.lit('priority >= ?', Worker.min_priority)) if Worker.min_priority
+          ds = ds.filter(::Sequel.lit('priority <= ?', Worker.max_priority)) if Worker.max_priority
+          ds = ds.filter(queue: Worker.queues) if Worker.queues.any?
+          ds = ds.by_priority
+
+          if Worker.read_ahead > 0
+            lock_with_read_ahead(ds, worker)
+          else
+            lock_with_for_update(ds, worker)
+          end
+        end
+      end
+    end
+  end
+end

scripts/generate_jobs.rb

@@ -0,0 +1,16 @@
+# usage: pipe this script into bin/console on a cc-worker vm
+
+begin
+  NUM_JOBS = 1
+  DELAY = 1
+
+  puts "Generating #{NUM_JOBS} dummy job(s) with delay of #{DELAY} seconds"
+  enqueuer = VCAP::CloudController::Jobs::Enqueuer.new(queue: VCAP::CloudController::Jobs::Queues.generic)
+  start_time = Time.now
+  NUM_JOBS.times do
+    dummy_job = VCAP::CloudController::Jobs::Runtime::BlobstoreDelete.new('00000000-0000-0000-0000-000000000000/0000000000000000000000000000000000000000', :droplet_blobstore)
+    enqueuer.enqueue(dummy_job)
+    sleep DELAY
+  end
+  puts "Generated #{NUM_JOBS} dummy job(s) in #{Time.now - start_time} seconds"
+end

spec/unit/lib/delayed_job/delayed_worker_spec.rb

@@ -141,6 +141,56 @@
       end
     end
 
+    context 'when DB type is mysql' do
+      before do
+        db = instance_double(Sequel::Database)
+        allow(db).to receive(:database_type).and_return(:mysql)
+        allow(Sequel::Model).to receive(:db).and_return(db)
+      end
+
+      it 'read_ahead defaults to DEFAULT_READ_AHEAD_MYSQL' do
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(CloudController::DelayedWorker::DEFAULT_READ_AHEAD_MYSQL)
+      end
+
+      it 'read_ahead can be configured' do
+        TestConfig.config[:jobs][:read_ahead] = 3
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(3)
+      end
+
+      it 'read_ahead cant be set to 0' do
+        TestConfig.config[:jobs][:read_ahead] = 0
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(CloudController::DelayedWorker::DEFAULT_READ_AHEAD_MYSQL)
+      end
+    end
+
+    context 'when DB type is postgres' do
+      before do
+        db = instance_double(Sequel::Database)
+        allow(db).to receive(:database_type).and_return(:postgres)
+        allow(Sequel::Model).to receive(:db).and_return(db)
+      end
+
+      it 'read_ahead defaults to DEFAULT_READ_AHEAD_POSTGRES' do
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(CloudController::DelayedWorker::DEFAULT_READ_AHEAD_POSTGRES)
+      end
+
+      it 'read_ahead can be configured' do
+        TestConfig.config[:jobs][:read_ahead] = 3
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(3)
+      end
+
+      it 'read_ahead cant be set to negative values' do
+        TestConfig.config[:jobs][:read_ahead] = -1
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(CloudController::DelayedWorker::DEFAULT_READ_AHEAD_POSTGRES)
+      end
+    end
+
     describe 'publish metrics' do
       before do
         allow(Prometheus::Client::DataStores::DirectFileStore).to receive(:new)

spec/unit/lib/delayed_job/sequel_patch_spec.rb

@@ -0,0 +1,8 @@
+RSpec.describe 'sequel_patch' do
+  describe 'version' do
+    it 'is not updated' do
+      expect(Gem.loaded_specs['talentbox-delayed_job_sequel'].version).to eq('4.3.0'),
+                                                                          'revisit monkey patch in lib/delayed_job/sequel_patch.rb'
+    end
+  end
+end