More consistent shutdown/kill behavior for executors.

Author: jdantonio

lib/concurrent/executor/executor.rb

@@ -21,7 +21,7 @@ module Executor
     def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
       mutex.synchronize do
-        break false unless running?
+        return false unless running?
         execute(*args, &task)
         true
       end
@@ -41,14 +41,14 @@ def <<(task)
     #
     # @return [Boolean] `true` when running, `false` when shutting down or shutdown
     def running?
-      ! @stop_event.set?
+      ! stop_event.set?
     end
 
     # Is the executor shutdown?
     #
     # @return [Boolean] `true` when shutdown, `false` when shutting down or running
     def shutdown?
-      @stop_event.set?
+      stop_event.set?
     end
 
     def shutdown
@@ -67,7 +67,7 @@ def kill
     #
     # @return [Boolean] `true` if shutdown complete or false on `timeout`
     def wait_for_termination(timeout)
-      @stopped_event.wait(timeout.to_i)
+      stopped_event.wait(timeout.to_i)
     end
 
     protected

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -41,6 +41,7 @@ def kill
         @thread.kill if alive?
         stopped_event.set unless alive?
       end
+      true
     end
 
     protected

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -90,7 +90,6 @@ def initialize(opts = {})
       init_executor
 
       @pool = []
-      @stopped_event = Event.new
       @queue = Queue.new
       @scheduled_task_count = 0
       @completed_task_count = 0
@@ -133,19 +132,6 @@ def status
       mutex.synchronize { @pool.collect { |worker| worker.status } }
     end
 
-    # Block until thread pool shutdown is complete or until `timeout` seconds have
-    # passed.
-    #
-    # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
-    #   must be called before this method (or on another thread).
-    #
-    # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
-    #
-    # @return [Boolean] `true` if shutdown complete or false on `timeout`
-    def wait_for_termination(timeout)
-      return @stopped_event.wait(timeout.to_i)
-    end
-
     # Begin an orderly shutdown. Tasks already in the queue will be executed,
     # but no new tasks will be accepted. Has no additional effect if the
     # thread pool is not running.
@@ -155,11 +141,12 @@ def shutdown
         @queue.clear
         stop_event.set
         if @pool.empty?
-          @stopped_event.set
+          stopped_event.set
         else
           @pool.length.times{ @queue << :stop }
         end
       end
+      true
     end
 
     # Begin an immediate shutdown. In-progress tasks will be allowed to

lib/concurrent/executor/timer_set.rb

@@ -22,37 +22,10 @@ class TimerSet
     # @option opts [object] :executor when provided will run all operations on
     #   this executor rather than the global thread pool (overrides :operation)
     def initialize(opts = {})
-      @mutex = Mutex.new
-      @shutdown = Event.new
       @queue = PriorityQueue.new(order: :min)
       @executor = get_executor_from(opts)
       @thread = nil
-    end
-
-    # Am I running?
-    #
-    # @return [Boolean] `true` when running, `false` when shutting down or shutdown
-    def running?
-      ! @shutdown.set?
-    end
-
-    # Am I shutdown?
-    #
-    # @return [Boolean] `true` when shutdown, `false` when shutting down or running
-    def shutdown?
-      @shutdown.set?
-    end
-
-    # Block until shutdown is complete or until `timeout` seconds have passed.
-    #
-    # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
-    #   must be called before this method (or on another thread).
-    #
-    # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
-    #
-    # @return [Boolean] `true` if shutdown complete or false on `timeout`
-    def wait_for_termination(timeout)
-      @shutdown.wait(timeout.to_f)
+      init_executor
     end
 
     # Post a task to be execute at the specified time. The given time may be either
@@ -69,8 +42,8 @@ def wait_for_termination(timeout)
     # @raise [ArgumentError] if the intended execution time is not in the future
     # @raise [ArgumentError] if no block is given
     def post(intended_time, &block)
-      @mutex.synchronize do
-        return false if shutdown?
+      mutex.synchronize do
+        return false unless running?
         raise ArgumentError.new('no block given') unless block_given?
         time = TimerSet.calculate_schedule_time(intended_time).to_f
 
@@ -88,12 +61,12 @@ def post(intended_time, &block)
     # but no new tasks will be accepted. Has no additional effect if the
     # thread pool is not running.
     def shutdown
-      @mutex.synchronize do
-        unless @shutdown.set?
-          @queue.clear
-          @thread.kill if @thread
-          @shutdown.set
-        end
+      mutex.synchronize do
+        break unless running?
+        stop_event.set
+        @queue.clear
+        @thread.kill if @thread
+        stopped_event.set
       end
       true
     end
@@ -146,7 +119,7 @@ def <=>(other)
     #
     # @!visibility private
     def check_processing_thread!
-      @mutex.synchronize do
+      mutex.synchronize do
         return if shutdown? || @queue.empty?
         if @thread && @thread.status == 'sleep'
           @thread.wakeup
@@ -165,7 +138,7 @@ def check_processing_thread!
     #
     # @!visibility private
     def next_task
-      @mutex.synchronize do
+      mutex.synchronize do
         unless @queue.empty? || @queue.peek.time > Time.now.to_f
           @queue.pop
         else
@@ -182,7 +155,7 @@ def next_task
     #
     # @!visibility private
     def next_sleep_interval
-      @mutex.synchronize do
+      mutex.synchronize do
         if @queue.empty?
           nil
         else