scheduler.rb in good_job-3.26.0

- old
+ new

@@ -69,13 +69,15 @@
 
     # Tests whether the scheduler is running.
     # @return [Boolean, nil]
     delegate :running?, to: :executor, allow_nil: true
 
-    # Tests whether the scheduler is shutdown.
+    # Tests whether the scheduler is shutdown and no tasks are running.
     # @return [Boolean, nil]
-    delegate :shutdown?, to: :executor, allow_nil: true
+    def shutdown?
+      @executor.nil? || (executor.shutdown? && !executor.shuttingdown?)
+    end
 
     # Shut down the scheduler.
     # This stops all threads in the thread pool.
     # Use {#shutdown?} to determine whether threads have stopped.
     # @param timeout [Numeric, nil] Seconds to wait for actively executing jobs to finish
@@ -83,26 +85,26 @@
     #   * +-1+, the scheduler will wait until the shutdown is complete.
     #   * +0+, the scheduler will immediately shutdown and stop any active tasks.
     #   * A positive number will wait that many seconds before stopping any remaining active tasks.
     # @return [void]
     def shutdown(timeout: -1)
-      return if executor.nil? || executor.shutdown?
+      return if executor.nil? || (executor.shutdown? && !executor.shuttingdown?)
 
       instrument("scheduler_shutdown_start", { timeout: timeout })
       instrument("scheduler_shutdown", { timeout: timeout }) do
         if executor.running?
           @timer_set.shutdown
           executor.shutdown
         end
 
         if executor.shuttingdown? && timeout
           executor_wait = timeout.negative? ? nil : timeout
+          return if executor.wait_for_termination(executor_wait)
 
-          unless executor.wait_for_termination(executor_wait)
-            instrument("scheduler_shutdown_kill", { active_job_ids: @performer.performing_active_job_ids.to_a })
-            executor.kill
-          end
+          instrument("scheduler_shutdown_kill", { active_job_ids: @performer.performing_active_job_ids.to_a })
+          executor.kill
+          executor.wait_for_termination
         end
       end
     end
 
     # Restart the Scheduler.
@@ -266,9 +268,11 @@
     # @param fanout [Boolean] Whether to eagerly create a 2nd execution thread if a job is found.
     # @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
+
         Rails.application.reloader.wrap do
           thr_performer.next do |found|
             thr_scheduler.create_thread({ fanout: fanout }) if found && fanout
           end
         end