Shared #shutdown for RubySingleThreadExecutor, RubyThreadPoolExecutor, and TimerSet.

Author: jdantonio

lib/concurrent/executor/executor.rb

@@ -51,7 +51,16 @@ def shutdown?
       stop_event.set?
     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.
     def shutdown
+      mutex.synchronize do
+        return unless running?
+        stop_event.set
+        stop_execution
+      end
+      true
     end
 
     def kill
@@ -83,6 +92,10 @@ def init_executor
     def execute(*args, &task)
       raise NotImplementedError
     end
+
+    def stop_execution
+      stopped_event.set
+    end
   end
 
   if RUBY_PLATFORM == 'java'

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -17,18 +17,6 @@ def initialize(opts = {})
       init_executor
     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.
-    def shutdown
-      mutex.synchronize do
-        return unless running?
-        stop_event.set
-        @queue << :stop
-        stopped_event.set unless alive?
-      end
-    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
@@ -46,26 +34,37 @@ def kill
 
     protected
 
+    # @!visibility private
     def execute(*args, &task)
       supervise
       @queue << [args, task]
     end
 
+    # @!visibility private
+    def stop_execution
+      @queue << :stop
+      stopped_event.set unless alive?
+    end
+
+    # @!visibility private
     def alive?
       @thread && @thread.alive?
     end
 
+    # @!visibility private
     def supervise
       @thread = new_worker_thread unless alive?
     end
 
+    # @!visibility private
     def new_worker_thread
       Thread.new do
         Thread.current.abort_on_exception = false
         work
       end
     end
 
+    # @!visibility private
     def work
       loop do
         task = @queue.pop

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -132,23 +132,6 @@ def status
       mutex.synchronize { @pool.collect { |worker| worker.status } }
     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.
-    def shutdown
-      mutex.synchronize do
-        break unless running?
-        @queue.clear
-        stop_event.set
-        if @pool.empty?
-          stopped_event.set
-        else
-          @pool.length.times{ @queue << :stop }
-        end
-      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
@@ -200,6 +183,16 @@ def execute(*args, &task)
       grow_pool
     end
 
+    # @!visibility private
+    def stop_execution
+      @queue.clear
+      if @pool.empty?
+        stopped_event.set
+      else
+        @pool.length.times{ @queue << :stop }
+      end
+    end
+
     # Handler which executes the `overflow_policy` once the queue size
     # reaches `max_queue`.
     #

lib/concurrent/executor/timer_set.rb

@@ -49,19 +49,6 @@ def post(intended_time, &task)
       end
     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.
-    def shutdown
-      mutex.synchronize do
-        break unless running?
-        stop_event.set
-        @queue.clear
-        @thread.kill if @thread
-        stopped_event.set
-      end
-      true
-    end
     alias_method :kill, :shutdown
 
     # Calculate an Epoch time with milliseconds at which to execute a
@@ -102,6 +89,7 @@ def <=>(other)
       end
     end
 
+    # @!visibility private
     def execute(time, &task)
       if (time - Time.now.to_f) <= 0.01
         @executor.post(&task)
@@ -110,6 +98,13 @@ def execute(time, &task)
       end
     end
 
+    # @!visibility private
+    def stop_execution
+      @queue.clear
+      @thread.kill if @thread
+      stopped_event.set
+    end
+
     # Check the status of the processing thread. This thread is responsible
     # for monitoring the internal task queue and sending tasks to the
     # executor when it is time for them to be processed. If there is no