scheduler.rb in good_job-4.6.0

- old
+ new

@@ -27,27 +27,35 @@
       idletime: 60,
       max_queue: Configuration::DEFAULT_MAX_THREADS,
       fallback_policy: :discard,
     }.freeze
 
+    # In CRuby, this sets the thread quantum to ~12.5ms ( 100ms * 2^(-3) ).
+    LOW_THREAD_PRIORITY = -3
+
     # @!attribute [r] instances
     #   @!scope class
     #   List of all instantiated Schedulers in the current process.
     #   @return [Array<GoodJob::Scheduler>, nil]
     cattr_reader :instances, default: Concurrent::Array.new, instance_reader: false
 
     # Human readable name of the scheduler that includes configuration values.
     # @return [String]
     attr_reader :name
 
+    # Whether to lower the thread priority to a fixed value
+    # @return [Boolean]
+    attr_accessor :lower_thread_priority
+
     # @param performer [GoodJob::JobPerformer]
     # @param max_threads [Numeric, nil] number of seconds between polls for jobs
     # @param max_cache [Numeric, nil] maximum number of scheduled jobs to cache in memory
     # @param warm_cache_on_initialize [Boolean] whether to warm the cache immediately, or manually by calling +warm_cache+
     # @param cleanup_interval_seconds [Numeric, nil] number of seconds between cleaning up job records
     # @param cleanup_interval_jobs [Numeric, nil] number of executed jobs between cleaning up job records
-    def initialize(performer, max_threads: nil, max_cache: nil, warm_cache_on_initialize: false, cleanup_interval_seconds: nil, cleanup_interval_jobs: nil)
+    # @param lower_thread_priority [Boolean] whether to lower the thread priority of execution threads
+    def initialize(performer, max_threads: nil, max_cache: nil, warm_cache_on_initialize: false, cleanup_interval_seconds: nil, cleanup_interval_jobs: nil, lower_thread_priority: false)
       raise ArgumentError, "Performer argument must implement #next" unless performer.respond_to?(:next)
 
       @performer = performer
 
       @max_cache = max_cache || 0
@@ -60,10 +68,12 @@
       @executor_options[:name] = name
 
       @cleanup_tracker = CleanupTracker.new(cleanup_interval_seconds: cleanup_interval_seconds, cleanup_interval_jobs: cleanup_interval_jobs)
       @executor_options[:name] = name
 
+      self.lower_thread_priority = lower_thread_priority
+
       create_executor
       warm_cache if warm_cache_on_initialize
       self.class.instances << self
     end
 
@@ -269,9 +279,10 @@
     # @return [void]
     def create_task(delay = 0, fanout: false)
       future = Concurrent::ScheduledTask.new(delay, args: [self, performer], executor: executor, timer_set: timer_set) do |thr_scheduler, thr_performer|
         Thread.current.name = Thread.current.name.sub("-worker-", "-thread-") if Thread.current.name
         Thread.current[:good_job_scheduler] = thr_scheduler
+        Thread.current.priority = -3 if thr_scheduler.lower_thread_priority
 
         Rails.application.reloader.wrap do
           thr_performer.next do |found|
             thr_scheduler.create_thread({ fanout: fanout }) if found && fanout
           end