Merge pull request #38 from jdantonio/better-thread-pool

Refactored thread pools to more closely mirror Java ThreadPoolExecutor.

Author: jdantonio

lib/concurrent/cached_thread_pool.rb

@@ -13,7 +13,7 @@ module Concurrent
     #  
     #   On creation a +CachedThreadPool+ has zero running threads. New threads are
     #   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
+    #   will grow until +#max_length+ 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
     #   with the new operation.

lib/concurrent/java_abstract_thread_pool.rb

@@ -5,6 +5,35 @@ module Concurrent
     # @!macro cached_thread_pool
     module JavaAbstractThreadPool
 
+      def min_length
+        @executor.getCorePoolSize
+      end
+
+      def max_length
+        @executor.getMaximumPoolSize
+      end
+
+      def length
+        @executor.getPoolSize
+      end
+      alias_method :current_length, :length
+
+      def largest_length
+        @executor.getLargestPoolSize
+      end
+
+      def scheduled_task_count
+        @executor.getTaskCount
+      end
+
+      def completed_task_count
+        @executor.getCompletedTaskCount
+      end
+
+      def idletime
+        @executor.getKeepAliveTime(java.util.concurrent.TimeUnit::SECONDS)
+      end
+
       # Is the thread pool running?
       #
       # @return [Boolean] +true+ when running, +false+ when shutting down or shutdown
@@ -86,15 +115,6 @@ 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? ? 1 : 0
-      end
-      alias_method :size, :length
     end
   end
 end

lib/concurrent/java_cached_thread_pool.rb

@@ -8,21 +8,44 @@ module Concurrent
     class JavaCachedThreadPool
       include JavaAbstractThreadPool
 
+      # The maximum number of threads that may be created in the pool
+      # (unless overridden during construction).
+      DEFAULT_MAX_POOL_SIZE = java.lang.Integer::MAX_VALUE # 2147483647
+
+      # The maximum number of seconds a thread in the pool may remain idle before
+      # being reclaimed (unless overridden during construction).
+      DEFAULT_THREAD_IDLETIME = 60
+
+      # The maximum number of threads that may be created in the pool.
+      attr_reader :max_length
+
       # Create a new thread pool.
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--
       def initialize(opts = {})
-        @executor = java.util.concurrent.Executors.newCachedThreadPool
-      end
+        idletime = (opts[:thread_idletime] || opts[:idletime] || DEFAULT_THREAD_IDLETIME).to_i
+        raise ArgumentError.new('idletime must be greater than zero') if idletime <= 0
 
-      # 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? ? 1 : 0
+        @max_length = opts[:max_threads] || opts[:max] || DEFAULT_MAX_POOL_SIZE
+        raise ArgumentError.new('maximum_number of threads must be greater than zero') if @max_length <= 0
+
+        #@executor = java.util.concurrent.Executors.newCachedThreadPool
+        @executor = java.util.concurrent.ThreadPoolExecutor.new(
+          0, @max_length,
+          idletime, java.util.concurrent.TimeUnit::SECONDS,
+          java.util.concurrent.SynchronousQueue.new,
+          java.util.concurrent.ThreadPoolExecutor::AbortPolicy.new)
+
+        #p = java.util.concurrent.Executors.newCachedThreadPool
+        #p.getCorePoolSize #=> 0
+        #p.getMaximumPoolSize #=> 2147483647
+        #p.getKeepAliveTime(java.util.concurrent.TimeUnit::SECONDS) #=> 60
+        #p.getQueue #=> #<Java::JavaUtilConcurrent::SynchronousQueue:0x68ec7913>
+
+        #p.getActiveCount #=> 0
+        #p.getQueue.size #=> 0
+        #p.getRejectedExecutionHandler #=> #<Java::JavaUtilConcurrent::ThreadPoolExecutor::AbortPolicy:0x57f897a7>
       end
-      alias_method :size, :length
     end
   end
 end

lib/concurrent/java_fixed_thread_pool.rb

@@ -16,17 +16,23 @@ def initialize(num_threads = Concurrent::processor_count)
         @num_threads = num_threads.to_i
         raise ArgumentError.new('number of threads must be greater than zero') if @num_threads < 1
 
-        @executor = java.util.concurrent.Executors.newFixedThreadPool(@num_threads)
-      end
+        #@executor = java.util.concurrent.Executors.newFixedThreadPool(@num_threads)
+        @executor = java.util.concurrent.ThreadPoolExecutor.new(
+          @num_threads, @num_threads,
+          0, java.util.concurrent.TimeUnit::SECONDS,
+          java.util.concurrent.LinkedBlockingQueue.new,
+          java.util.concurrent.ThreadPoolExecutor::AbortPolicy.new)
 
-      # 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
+        #p = java.util.concurrent.Executors.newFixedThreadPool(10)
+        #p.getCorePoolSize #=> 10
+        #p.getMaximumPoolSize #=> 10
+        #p.getKeepAliveTime(java.util.concurrent.TimeUnit::SECONDS) #=> 0
+        #p.getQueue #=> #<Java::JavaUtilConcurrent::LinkedBlockingQueue:0x97dabf4>
+
+        #p.getActiveCount #=> 0
+        #p.getQueue.size #=> 0
+        #p.getRejectedExecutionHandler #=> #<Java::JavaUtilConcurrent::ThreadPoolExecutor::AbortPolicy:0x7e41986c>
       end
-      alias_method :size, :length
     end
   end
 end

lib/concurrent/ruby_cached_thread_pool.rb

@@ -17,7 +17,17 @@ class RubyCachedThreadPool
     DEFAULT_THREAD_IDLETIME = 60
 
     # The maximum number of threads that may be created in the pool.
-    attr_accessor :max_threads
+    attr_reader :max_length
+
+    # The minimum number of threads that may be created in the pool.
+    attr_reader :min_length
+
+    attr_reader :largest_length
+
+    attr_reader :scheduled_task_count
+    attr_reader :completed_task_count
+
+    attr_reader :idletime
 
     # Create a new thread pool.
     #
@@ -33,8 +43,8 @@ def initialize(opts = {})
       @idletime = (opts[:thread_idletime] || opts[:idletime] || DEFAULT_THREAD_IDLETIME).to_i
       raise ArgumentError.new('idletime must be greater than zero') if @idletime <= 0
 
-      @max_threads = opts[:max_threads] || opts[:max] || DEFAULT_MAX_POOL_SIZE
-      raise ArgumentError.new('maximum_number of threads must be greater than zero') if @max_threads <= 0
+      @max_length = opts[:max_threads] || opts[:max] || DEFAULT_MAX_POOL_SIZE
+      raise ArgumentError.new('maximum_number of threads must be greater than zero') if @max_length <= 0
 
       @state = :running
       @pool = []
@@ -43,6 +53,10 @@ def initialize(opts = {})
 
       @busy = []
       @idle = []
+      @scheduled_task_count = 0
+      @completed_task_count = 0
+      @min_length = 0
+      @largest_length = 0
     end
 
     # Is the thread pool running?
@@ -86,18 +100,20 @@ def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
       @mutex.synchronize do
         break false unless @state == :running
+        @scheduled_task_count += 1
 
         if @idle.empty?
-          if @idle.length + @busy.length < @max_threads
+          if @idle.length + @busy.length < @max_length
             worker = create_worker_thread
           else
-            worker = @busy.shift
+            worker = @busy.pop
           end
         else
           worker = @idle.pop
         end
 
         @busy.push(worker)
+        @largest_length = [@idle.length + @busy.length, @largest_length].max
         worker.signal(*args, &task)
 
         prune_stale_workers
@@ -152,11 +168,9 @@ def kill
     #   zero when the pool is shutdown
     def length
       @mutex.synchronize do
-        @state == :running ? @busy.length + @idle.length : 0
+        @state != :shutdown ? @busy.length + @idle.length : 0
       end
     end
-    alias_method :size, :length
-    alias_method :current_size, :length
     alias_method :current_length, :length
 
     # @!visibility private
@@ -172,11 +186,14 @@ def on_worker_exit(worker) # :nodoc:
     end
 
     # @!visibility private
-    def on_end_task(worker) # :nodoc:
+    def on_end_task(worker, success) # :nodoc:
       @mutex.synchronize do
+        @completed_task_count += 1 #if success
         break unless @state == :running
-        @busy.delete(worker)
-        @idle.push(worker)
+        unless worker.tasks_remaining?
+          @busy.delete(worker)
+          @idle.push(worker)
+        end
       end
     end
 

lib/concurrent/ruby_cached_thread_pool/worker.rb

@@ -13,9 +13,11 @@ def initialize(parent)
         @resource = ConditionVariable.new
         @tasks = Queue.new
       end
-     
-      def idle?
-        return ! @idletime.nil?
+
+      def tasks_remaining?
+        return @mutex.synchronize do
+          ! @tasks.empty?
+        end
       end
 
       def dead?
@@ -65,7 +67,6 @@ def run(thread = Thread.current)
         loop do
           task = @mutex.synchronize do
             @resource.wait(@mutex, 60) if @tasks.empty?
-
             @tasks.pop(true)
           end
 
@@ -76,13 +77,12 @@ def run(thread = Thread.current)
             break
           end
 
-          #@parent.on_start_task(self)
           begin
             task.last.call(*task.first)
-          rescue
+          rescue => ex
             # let it fail
           ensure
-            @parent.on_end_task(self)
+            @parent.on_end_task(self, ex.nil?)
           end
         end
       end