delayed_jobs lock_with_read_ahead for postgres

- monkey patch for delayed_jobs to allow configurable lock method
- config via jobs.read_ahead (number of jobs to read)
  - postgres default = 0 (use lock_with_for_update)
  - mysql default = 5 (use lock_with_read_ahead)
- jobs.read_ahead = 0: lock_with_for_update = lock jobs using SELECT FOR UPDATE
- jobs.read_ahead > 0: lock_with_read_ahead = optimistic locking of jobs
- add rake task generate_load to create jobs

Using lock_with_for_update showed severe performance problems on postgres
when there is a high number of jobs in the queue (>50k). Job processing
starves due to row locks and temp IO gets high.
lock_with_read_ahead performs much better under such load situations.

Performance on low/normal job load is comparable.

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/delayed_job/delayed_worker.rb

@@ -25,6 +25,8 @@ def start_working
     setup_metrics(config) if @publish_metrics
     BackgroundJobEnvironment.new(config).setup_environment(readiness_port)
 
+    # load monkey patch for sequel backend to support configurable job lock method (postgres only)
+    require 'delayed_job/sequel_patch'
     logger = Steno.logger('cc-worker')
     logger.info("Starting job with options #{@queue_options}")
     setup_app_log_emitter(config, logger)
@@ -56,6 +58,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) || Delayed::Worker::DEFAULT_READ_AHEAD
+      # lock for update is not configurable for mysql
+      read_ahead = Delayed::Worker::DEFAULT_READ_AHEAD if read_ahead <= 0
+    else
+      # read_ahead 0 = lock for update (default for postgres)
+      read_ahead = config.get(:jobs, :read_ahead) || 0
+      read_ahead = 0 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

lib/tasks/jobs.rake

@@ -75,4 +75,24 @@ namespace :jobs do
                                        thread_grace_period_seconds: args.thread_grace_period_seconds,
                                        publish_metrics: publish_metrics).start_working
   end
+
+  desc 'Generate dummy jobs for tests. Usage: rake jobs:generate_load\'[num,delay]\''
+  task :generate_load, %i[num delay] => :environment do |_t, args|
+    args.with_defaults(num: '1')
+    args.with_defaults(delay: '1')
+    num = args[:num].to_i
+    delay = args[:delay].to_f
+    puts "Generating #{num} dummy job(s) with delay of #{delay} seconds"
+
+    ENV['PROCESS_TYPE'] = 'cc-worker'
+    RakeConfig.context = :worker
+    VCAP::CloudController::DB.load_models(RakeConfig.config.get(:db), @logger)
+    RakeConfig.config.configure_components
+
+    num.times do
+      dummy_job = VCAP::CloudController::Jobs::Runtime::BlobstoreDelete.new('00000000-0000-0000-0000-000000000000/0000000000000000000000000000000000000000', :droplet_blobstore)
+      VCAP::CloudController::Jobs::Enqueuer.new(dummy_job, queue: VCAP::CloudController::Jobs::Queues.generic).enqueue
+      sleep delay
+    end
+  end
 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' do
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(Delayed::Worker::DEFAULT_READ_AHEAD)
+      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(Delayed::Worker::DEFAULT_READ_AHEAD)
+      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 0' do
+        cc_delayed_worker.start_working
+        expect(Delayed::Worker.read_ahead).to eq(0)
+      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(0)
+      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