wip: RubyThreadPoolExecutor now uses Exector for stop events.

Author: jdantonio

lib/concurrent/executor/executor.rb

@@ -8,6 +8,35 @@ module Concurrent
 
   module Executor
 
+    # Submit a task to the thread pool for asynchronous processing.
+    #
+    # @param [Array] args zero or more arguments to be passed to the task
+    #
+    # @yield the asynchronous task to perform
+    #
+    # @return [Boolean] `true` if the task is queued, `false` if the thread pool
+    #   is not running
+    #
+    # @raise [ArgumentError] if no task is given
+    def post(*args, &task)
+      raise ArgumentError.new('no block given') unless block_given?
+      mutex.synchronize do
+        break false unless running?
+        execute(*args, &task)
+        true
+      end
+    end
+
+    # Submit a task to the thread pool for asynchronous processing.
+    #
+    # @param [Proc] task the asynchronous task to perform
+    #
+    # @return [self] returns itself
+    def <<(task)
+      post(&task)
+      self
+    end
+
     # Is the thread pool running?
     #
     # @return [Boolean] `true` when running, `false` when shutting down or shutdown
@@ -41,9 +70,6 @@ def wait_for_termination(timeout)
       @stopped_event.wait(timeout.to_i)
     end
 
-    def post(*args, &task)
-    end
-
     protected
 
     attr_reader :mutex, :stop_event, :stopped_event
@@ -55,6 +81,7 @@ def init_executor
     end
 
     def execute(*args, &task)
+      raise NotImplementedError
     end
   end
 

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -17,26 +17,6 @@ def initialize(opts = {})
       init_executor
     end
 
-    # Submit a task to the thread pool for asynchronous processing.
-    #
-    # @param [Array] args zero or more arguments to be passed to the task
-    #
-    # @yield the asynchronous task to perform
-    #
-    # @return [Boolean] `true` if the task is queued, `false` if the thread pool
-    #   is not running
-    #
-    # @raise [ArgumentError] if no task is given
-    def post(*args, &task)
-      raise ArgumentError.new('no block given') unless block_given?
-      mutex.synchronize do
-        break false unless running?
-        supervise
-        @queue << [args, task]
-        true
-      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.
@@ -65,6 +45,11 @@ def kill
 
     protected
 
+    def execute(*args, &task)
+      supervise
+      @queue << [args, task]
+    end
+
     def alive?
       @thread && @thread.alive?
     end

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -87,11 +87,11 @@ def initialize(opts = {})
       raise ArgumentError.new('min_threads cannot be less than zero') if @min_length < 0
       raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(@overflow_policy)
 
-      @state = :running
+      init_executor
+
       @pool = []
-      @terminator = Event.new
+      @stopped_event = Event.new
       @queue = Queue.new
-      @mutex = Mutex.new
       @scheduled_task_count = 0
       @completed_task_count = 0
       @largest_length = 0
@@ -104,8 +104,8 @@ def initialize(opts = {})
     #
     # @return [Integer] the length
     def length
-      @mutex.synchronize do
-        @state != :shutdown ? @pool.length : 0
+      mutex.synchronize do
+        running? ? @pool.length : 0
       end
     end
     alias_method :current_length, :length
@@ -122,29 +122,15 @@ def queue_length
     #
     # @return [Integer] the remaining_capacity
     def remaining_capacity
-      @mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }
-    end
-
-    # Is the thread pool running?
-    #
-    # @return [Boolean] `true` when running, `false` when shutting down or shutdown
-    def running?
-      @mutex.synchronize { @state == :running }
+      mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }
     end
 
     # Returns an array with the status of each thread in the pool
     #
     # This method is deprecated and will be removed soon.
     def status
       warn '[DEPRECATED] `status` is deprecated and will be removed soon.'
-      @mutex.synchronize { @pool.collect { |worker| worker.status } }
-    end
-
-    # Is the thread pool shutdown?
-    #
-    # @return [Boolean] `true` when shutdown, `false` when shutting down or running
-    def shutdown?
-      @mutex.synchronize { @state != :running }
+      mutex.synchronize { @pool.collect { |worker| worker.status } }
     end
 
     # Block until thread pool shutdown is complete or until `timeout` seconds have
@@ -157,7 +143,7 @@ def shutdown?
     #
     # @return [Boolean] `true` if shutdown complete or false on `timeout`
     def wait_for_termination(timeout)
-      return @terminator.wait(timeout.to_i)
+      return @stopped_event.wait(timeout.to_i)
     end
 
     # Submit a task to the thread pool for asynchronous processing.
@@ -172,8 +158,8 @@ def wait_for_termination(timeout)
     # @raise [ArgumentError] if no task is given
     def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
-      @mutex.synchronize do
-        break false unless @state == :running
+      mutex.synchronize do
+        break false unless running?
         return handle_overflow(*args, &task) if @max_queue != 0 && @queue.length >= @max_queue
         @scheduled_task_count += 1
         @queue << [args, task]
@@ -190,14 +176,13 @@ def post(*args, &task)
     # 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 @state == :running
+      mutex.synchronize do
+        break unless running?
         @queue.clear
+        stop_event.set
         if @pool.empty?
-          @state = :shutdown
-          @terminator.set
+          @stopped_event.set
         else
-          @state = :shuttingdown
           @pool.length.times{ @queue << :stop }
         end
       end
@@ -208,34 +193,34 @@ def shutdown
     # will be accepted. Has no additional effect if the thread pool is
     # not running.
     def kill
-      @mutex.synchronize do
-        break if @state == :shutdown
+      mutex.synchronize do
+        return if shutdown?
+        stop_event.set
         @queue.clear
-        @state = :shutdown
         drain_pool
-        @terminator.set
+        stopped_event.set
       end
     end
 
     # Run on task completion.
     #
     # @!visibility private
     def on_end_task # :nodoc:
-      @mutex.synchronize do
+      mutex.synchronize do
         @completed_task_count += 1 #if success
-        break unless @state == :running
+        break unless running?
       end
     end
 
     # Run when a thread worker exits.
     #
     # @!visibility private
     def on_worker_exit(worker) # :nodoc:
-      @mutex.synchronize do
+      mutex.synchronize do
         @pool.delete(worker)
-        if @pool.empty? && @state != :running
-          @state = :shutdown
-          @terminator.set
+        if @pool.empty? && ! running?
+          stop_event.set
+          stopped_event.set
         end
       end
     end