Most executors now share behavior through the Executor and JavaExecutor mixin modules.

Author: jdantonio

lib/concurrent/executor/executor.rb

@@ -44,11 +44,18 @@ def running?
       ! stop_event.set?
     end
 
+    # Is the executor shuttingdown?
+    #
+    # @return [Boolean] `true` when not running and not shutdown, else `false`
+    def shuttingdown?
+      ! (running? || shutdown?)
+    end
+
     # Is the executor shutdown?
     #
     # @return [Boolean] `true` when shutdown, `false` when shutting down or running
     def shutdown?
-      stop_event.set?
+      stopped_event.set?
     end
 
     # Begin an orderly shutdown. Tasks already in the queue will be executed,
@@ -58,12 +65,23 @@ def shutdown
       mutex.synchronize do
         return unless running?
         stop_event.set
-        stop_execution
+        shutdown_execution
       end
       true
     end
 
+    # Begin an immediate shutdown. In-progress tasks will be allowed to
+    # complete but enqueued tasks will be dismissed and no new tasks
+    # will be accepted. Has no additional effect if the thread pool is
+    # not running.
     def kill
+      mutex.synchronize do
+        return if shutdown?
+        stop_event.set
+        kill_execution
+        stopped_event.set
+      end
+      true
     end
 
     # Block until executor shutdown is complete or until `timeout` seconds have
@@ -76,7 +94,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_f)
     end
 
     protected
@@ -93,9 +111,13 @@ def execute(*args, &task)
       raise NotImplementedError
     end
 
-    def stop_execution
+    def shutdown_execution
       stopped_event.set
     end
+
+    def kill_execution
+      # do nothing
+    end
   end
 
   if RUBY_PLATFORM == 'java'
@@ -138,14 +160,25 @@ def <<(task)
       #
       # @return [Boolean] `true` when running, `false` when shutting down or shutdown
       def running?
-        ! (@executor.isShutdown || @executor.isTerminated)
+        ! (shuttingdown? || shutdown?)
+      end
+
+      # Is the executor shuttingdown?
+      #
+      # @return [Boolean] `true` when not running and not shutdown, else `false`
+      def shuttingdown?
+        if @executor.respond_to? :isTerminating
+          @executor.isTerminating
+        else
+          false
+        end
       end
 
       # Is the executor shutdown?
       #
       # @return [Boolean] `true` when shutdown, `false` when shutting down or running
       def shutdown?
-        @executor.isShutdown
+        @executor.isShutdown || @executor.isTerminated
       end
 
       # Block until executor shutdown is complete or until `timeout` seconds have
@@ -166,7 +199,7 @@ def wait_for_termination(timeout)
       # executor is not running.
       def shutdown
         @executor.shutdown
-        return nil
+        nil
       end
 
       # Begin an immediate shutdown. In-progress tasks will be allowed to

lib/concurrent/executor/java_cached_thread_pool.rb

@@ -24,8 +24,7 @@ def initialize(opts = {})
         @executor = java.util.concurrent.Executors.newCachedThreadPool
         @executor.setRejectedExecutionHandler(OVERFLOW_POLICIES[@overflow_policy].new)
 
-        # without this the process may fail to exit
-        at_exit { self.kill }
+        set_shutdown_hook
       end
     end
   end

lib/concurrent/executor/java_fixed_thread_pool.rb

@@ -26,8 +26,7 @@ def initialize(num_threads, opts = {})
         @executor = java.util.concurrent.Executors.newFixedThreadPool(num_threads)
         @executor.setRejectedExecutionHandler(OVERFLOW_POLICIES[@overflow_policy].new)
 
-        # without this the process may fail to exit
-        at_exit { self.kill }
+        set_shutdown_hook
       end
     end
   end

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -17,21 +17,6 @@ def initialize(opts = {})
       init_executor
     end
 
-    # Begin an immediate shutdown. In-progress tasks will be allowed to
-    # complete but enqueued tasks will be dismissed and no new tasks
-    # will be accepted. Has no additional effect if the thread pool is
-    # not running.
-    def kill
-      mutex.synchronize do
-        return if shutdown?
-        stop_event.set
-        @queue.clear
-        @thread.kill if alive?
-        stopped_event.set unless alive?
-      end
-      true
-    end
-
     protected
 
     # @!visibility private
@@ -41,11 +26,17 @@ def execute(*args, &task)
     end
 
     # @!visibility private
-    def stop_execution
+    def shutdown_execution
       @queue << :stop
       stopped_event.set unless alive?
     end
 
+    # @!visibility private
+    def kill_execution
+      @queue.clear
+      @thread.kill if alive?
+    end
+
     # @!visibility private
     def alive?
       @thread && @thread.alive?

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -103,17 +103,15 @@ def initialize(opts = {})
     #
     # @return [Integer] the length
     def length
-      mutex.synchronize do
-        running? ? @pool.length : 0
-      end
+      mutex.synchronize{ running? ? @pool.length : 0 }
     end
     alias_method :current_length, :length
 
     # The number of tasks in the queue awaiting execution.
     #
     # @return [Integer] the queue_length
     def queue_length
-      @queue.length
+      mutex.synchronize{ running? ? @queue.length : 0 }
     end
 
     # Number of tasks that may be enqueued before reaching `max_queue` and rejecting
@@ -132,20 +130,6 @@ def status
       mutex.synchronize { @pool.collect { |worker| worker.status } }
     end
 
-    # Begin an immediate shutdown. In-progress tasks will be allowed to
-    # complete but enqueued tasks will be dismissed and no new tasks
-    # will be accepted. Has no additional effect if the thread pool is
-    # not running.
-    def kill
-      mutex.synchronize do
-        return if shutdown?
-        stop_event.set
-        @queue.clear
-        drain_pool
-        stopped_event.set
-      end
-    end
-
     # Run on task completion.
     #
     # @!visibility private
@@ -184,7 +168,7 @@ def execute(*args, &task)
     end
 
     # @!visibility private
-    def stop_execution
+    def shutdown_execution
       @queue.clear
       if @pool.empty?
         stopped_event.set
@@ -193,6 +177,12 @@ def stop_execution
       end
     end
 
+    # @!visibility private
+    def kill_execution
+      @queue.clear
+      drain_pool
+    end
+
     # Handler which executes the `overflow_policy` once the queue size
     # reaches `max_queue`.
     #

lib/concurrent/executor/ruby_thread_pool_worker.rb

@@ -3,25 +3,25 @@
 module Concurrent
 
   # @!visibility private
-  class RubyThreadPoolWorker # :nodoc:
+  class RubyThreadPoolWorker
 
     # @!visibility private
-    def initialize(queue, parent) # :nodoc:
+    def initialize(queue, parent)
       @queue = queue
       @parent = parent
       @mutex = Mutex.new
       @last_activity = Time.now.to_f
     end
 
     # @!visibility private
-    def dead? # :nodoc:
+    def dead?
       return @mutex.synchronize do
         @thread.nil? ? false : ! @thread.alive?
       end
     end
 
     # @!visibility private
-    def last_activity # :nodoc:
+    def last_activity
       @mutex.synchronize { @last_activity }
     end
 
@@ -33,15 +33,15 @@ def status
     end
 
     # @!visibility private
-    def kill # :nodoc:
+    def kill
       @mutex.synchronize do
         Thread.kill(@thread) unless @thread.nil?
         @thread = nil
       end
     end
 
     # @!visibility private
-    def run(thread = Thread.current) # :nodoc:
+    def run(thread = Thread.current)
       @mutex.synchronize do
         raise StandardError.new('already running') unless @thread.nil?
         @thread = thread

lib/concurrent/executor/safe_task_executor.rb

@@ -25,4 +25,4 @@ def execute
       [success, value, reason]
     end
   end
-end
+end

lib/concurrent/executor/timer_set.rb

@@ -99,7 +99,7 @@ def execute(time, &task)
     end
 
     # @!visibility private
-    def stop_execution
+    def shutdown_execution
       @queue.clear
       @thread.kill if @thread
       stopped_event.set