Documented fixed thread pools.

Author: jdantonio

lib/concurrent/cached_thread_pool.rb

@@ -2,19 +2,17 @@
 
 module Concurrent
 
-  # @!class CachedThreadPool
-
   if defined? java.util
     require 'concurrent/java_cached_thread_pool'
     # @!macro [attach] cached_thread_pool
     #   A thread pool that dynamically grows and shrinks to fit the current workload.
     #   New threads are created as needed, existing threads are reused, and threads
     #   that remain idle for too long are killed and removed from the pool. These
-    #   pools are articularly suited to programs that run a high volume of short-lived
-    #   tasks.
+    #   pools are particularly suited to applications that perform a high volume of
+    #   short-lived tasks.
     #  
     #   On creation a +CachedThreadPool+ has zero running threads. New threads are
-    #   spawned on the pool as new operations are +#post+. The size of the pool
+    #   created on the pool as new operations are +#post+. The size of the pool
     #   will grow until +#max_threads+ threads are in the pool or until the number
     #   of threads exceeds the number of running and pending operations. When a new
     #   operation is post to the pool the first available idle thread will be tasked
@@ -26,14 +24,22 @@ module Concurrent
     #   efficient at reclaiming unused resources.
     #  
     #   The API and behavior of this class are based on Java's +CachedThreadPool+
+    #
+    #   @note When running on the JVM (JRuby) this class will inherit from +JavaCachedThreadPool+.
+    #     On all other platforms it will inherit from +RubyCachedThreadPool+.
+    #
+    #   @see Concurrent::RubyCachedThreadPool
+    #   @see Concurrent::JavaCachedThreadPool
     #  
     #   @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
     #   @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
     #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
     #   @see http://stackoverflow.com/questions/17957382/fixedthreadpool-vs-cachedthreadpool-the-lesser-of-two-evils
-    CachedThreadPool = Class.new(JavaCachedThreadPool)
+    class CachedThreadPool < JavaCachedThreadPool
+    end
   else
     # @!macro cached_thread_pool
-    CachedThreadPool = Class.new(RubyCachedThreadPool)
+    class CachedThreadPool < RubyCachedThreadPool
+    end
   end
 end

lib/concurrent/fixed_thread_pool.rb

@@ -4,8 +4,31 @@ module Concurrent
 
   if defined? java.util
     require 'concurrent/java_fixed_thread_pool'
-    FixedThreadPool = Class.new(JavaFixedThreadPool)
+    # @!macro [attach] fixed_thread_pool
+    #
+    #   A thread pool with a set number of threads. The number of threads in the pool
+    #   is set on construction and remains constant. When all threads are busy new
+    #   tasks +#post+ to the thread pool are enqueued until a thread becomes available.
+    #   Should a thread crash for any reason the thread will immediately be removed
+    #   from the pool and replaced.
+    #  
+    #   The API and behavior of this class are based on Java's +FixedThreadPool+
+    #
+    #   @note When running on the JVM (JRuby) this class will inherit from +JavaFixedThreadPool+.
+    #     On all other platforms it will inherit from +RubyFixedThreadPool+.
+    #
+    #   @see Concurrent::RubyFixedThreadPool
+    #   @see Concurrent::JavaFixedThreadPool
+    #  
+    #   @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
+    #   @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
+    #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
+    #   @see http://stackoverflow.com/questions/17957382/fixedthreadpool-vs-fixedthreadpool-the-lesser-of-two-evils
+    class FixedThreadPool < JavaFixedThreadPool
+    end
   else
-    FixedThreadPool = Class.new(RubyFixedThreadPool)
+    # @!macro fixed_thread_pool
+    class FixedThreadPool < RubyFixedThreadPool
+    end
   end
 end

lib/concurrent/java_fixed_thread_pool.rb

@@ -4,11 +4,11 @@
 
   module Concurrent
 
-    # @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
-    # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
-    # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
+    # @!macro fixed_thread_pool
     class JavaFixedThreadPool
 
+      # Create a new thread pool.
+      #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool-int-
       def initialize(num_threads = Concurrent::processor_count)
         @num_threads = num_threads.to_i
@@ -17,22 +17,50 @@ def initialize(num_threads = Concurrent::processor_count)
         @executor = java.util.concurrent.Executors.newFixedThreadPool(@num_threads)
       end
 
+      # Is the thread pool running?
+      #
+      # @return [Boolean] +true+ when running, +false+ when shutting down or shutdown
       def running?
         ! (shutdown? || terminated?)
       end
 
+      # Is the thread pool shutdown?
+      #
+      # @return [Boolean] +true+ when shutdown, +false+ when shutting down or running
       def shutdown?
         @executor.isShutdown
       end
 
+      # Were all tasks completed before shutdown?
+      #
+      # @return [Boolean] +true+ if shutdown and all tasks completed else +false+
       def terminated?
         @executor.isTerminated
       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)
         @executor.awaitTermination(timeout.to_i, java.util.concurrent.TimeUnit::SECONDS)
       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)
         raise ArgumentError.new('no block given') unless block_given?
         @executor.submit{ yield(*args) }
@@ -41,24 +69,40 @@ def post(*args)
         return false
       end
 
-      def <<(block)
-        @executor.submit(&block)
+      # Submit a task to the thread pool for asynchronous processing.
+      #
+      # @param [Proc] task the asynchronous task to perform
+      #
+      # @return [self] returns itself
+      def <<(task)
+        @executor.submit(&task)
       rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
         # do nothing
       ensure
         return self
       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
         @executor.shutdown
         return nil
       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
         @executor.shutdownNow
         return nil
       end
 
+      # The number of threads currently in the pool.
+      #
+      # @return [Integer] a non-zero value when the pool is running,
+      #   zero when the pool is shutdown
       def length
         running? ? @num_threads : 0
       end

lib/concurrent/ruby_cached_thread_pool.rb

@@ -140,8 +140,8 @@ def kill
       @mutex.synchronize do
         break if @state == :shutdown
         @state = :shutdown
-          @idle.each{|worker| worker.kill }
-          @busy.each{|worker| worker.kill }
+        @idle.each{|worker| worker.kill }
+        @busy.each{|worker| worker.kill }
         @terminator.set
       end
     end
@@ -156,6 +156,8 @@ def length
       end
     end
     alias_method :size, :length
+    alias_method :current_size, :length
+    alias_method :current_length, :length
 
     # @!visibility private
     def on_worker_exit(worker) # :nodoc:

lib/concurrent/ruby_fixed_thread_pool.rb

@@ -5,9 +5,21 @@
 
 module Concurrent
 
+  # @!macro fixed_thread_pool
+  #
+  # @note To prevent deadlocks and race conditions, no threads will be allocated
+  #   on construction. Threads will be allocated once the first task is post to
+  #   the pool. Additionally, threads that crash will be removed from the pool and
+  #   replaced. Thus the +#length+ and +#current_length+ may occasionally be
+  #   different.
   class RubyFixedThreadPool
 
-    def initialize(num_threads, opts = {})
+    # Create a new thread pool.
+    #
+    # @param [Integer] num_threads the number of threads to allocate
+    #
+    # @raise [ArgumentError] if +num_threads+ is less than or equal to zero
+    def initialize(num_threads)
       @num_threads = num_threads.to_i
       raise ArgumentError.new('number of threads must be greater than zero') if @num_threads < 1
 
@@ -18,34 +30,67 @@ def initialize(num_threads, opts = {})
       @mutex = Mutex.new
     end
 
+    # Is the thread pool running?
+    #
+    # @return [Boolean] +true+ when running, +false+ when shutting down or shutdown
     def running?
       return @state == :running
     end
 
+    # Is the thread pool shutdown?
+    #
+    # @return [Boolean] +true+ when shutdown, +false+ when shutting down or running
     def shutdown?
       return @state != :running
     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 @terminator.wait(timeout.to_i)
     end
 
-    def post(*args, &block)
-      raise ArgumentError.new('no block given') if block.nil?
+    # Submit a task to the thread pool for asynchronous processing.
+    #
+    # @param [Array] args zero or more arguments to be passed to the block
+    #
+    # @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 block is given
+    def post(*args, &task)
+      raise ArgumentError.new('no block given') if task.nil?
       @mutex.synchronize do
         break false unless @state == :running
-        @queue << [args, block]
+        @queue << [args, task]
         clean_pool
         fill_pool
         true
       end
     end
 
-    def <<(block)
-      self.post(&block)
+    # Submit a task to the thread pool for asynchronous processing.
+    #
+    # @param [Proc] task the asynchronous task to perform
+    #
+    # @return [self] returns itself
+    def <<(task)
+      self.post(&task)
       return self
     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 @state == :running
@@ -59,6 +104,10 @@ def shutdown
       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
+    # not running.
     def kill
       @mutex.synchronize do
         break if @state == :shutdown
@@ -69,21 +118,30 @@ def kill
       end
     end
 
+    # The number of threads allocated for the pool.
+    #
+    # @return [Integer] the number of threads allocated for a running pool,
+    #   zero when the pool is shutdown
     def length
       @mutex.synchronize do
         @state == :running ? @num_threads : 0
       end
     end
     alias_method :size, :length
 
+    # The number of threads currently in the pool.
+    #
+    # @return [Integer] the number of threads currently operating when the
+    #   pool is running, zero when the pool is shutdown
     def current_length
       @mutex.synchronize do
         @state == :running ? @pool.length : 0
       end
     end
     alias_method :current_size, :current_length
 
-    def create_worker_thread
+    # @!visibility private
+    def create_worker_thread # :nodoc:
       wrkr = Worker.new(@queue, self)
       Thread.new(wrkr, self) do |worker, parent|
         Thread.current.abort_on_exception = false
@@ -93,34 +151,40 @@ def create_worker_thread
       return wrkr
     end
 
-    def fill_pool
+    # @!visibility private
+    def fill_pool # :nodoc:
       return unless @state == :running
       while @pool.length < @num_threads
         @pool << create_worker_thread
       end
     end
 
-    def clean_pool
+    # @!visibility private
+    def clean_pool # :nodoc:
       @pool.reject! {|worker| worker.dead? } 
     end
 
-    def drain_pool
+    # @!visibility private
+    def drain_pool # :nodoc:
       @pool.each {|worker| worker.kill }
       @pool.clear
     end
 
-    def on_start_task(worker)
+    # @!visibility private
+    def on_start_task(worker) # :nodoc:
     end
 
-    def on_end_task(worker)
+    # @!visibility private
+    def on_end_task(worker) # :nodoc:
       @mutex.synchronize do
         break unless @state == :running
         clean_pool
         fill_pool
       end
     end
 
-    def on_worker_exit(worker)
+    # @!visibility private
+    def on_worker_exit(worker) # :nodoc:
       @mutex.synchronize do
         @pool.delete(worker)
         if @pool.empty? && @state != :running