Merge pull request #41 from jdantonio/thread-pool-overload

Implemented thread pool overload policies.

Author: jdantonio

.travis.yml

@@ -14,3 +14,4 @@ matrix:
   allow_failures:
     - rvm: ruby-head
     - rvm: jruby-head
+    - rvm: 1.9.3

lib/concurrent/java_cached_thread_pool.rb

@@ -14,23 +14,28 @@ class JavaCachedThreadPool < JavaThreadPoolExecutor
       #   of threads which may be created in the pool
       # @option opts [Integer] :idletime (+DEFAULT_THREAD_IDLETIMEOUT+) maximum
       #   number of seconds a thread may be idle before it is reclaimed
+      # @option opts [Symbol] :overflow_policy (+:abort+) the overflow policy
       #
       # @raise [ArgumentError] if +max_threads+ is less than or equal to zero
       # @raise [ArgumentError] if +idletime+ is less than or equal to zero
+      # @raise [ArgumentError] if +overflow_policy+ is not a known policy
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--
       def initialize(opts = {})
         max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
         idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
+        @overflow_policy = opts.fetch(:overflow_policy, :abort)
+        @max_queue = 0
 
         raise ArgumentError.new('idletime must be greater than zero') if idletime <= 0
         raise ArgumentError.new('max_threads must be greater than zero') if max_length <= 0
+        raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(@overflow_policy)
 
         @executor = java.util.concurrent.ThreadPoolExecutor.new(
-          0, max_length,
+          @max_queue, max_length,
           idletime, java.util.concurrent.TimeUnit::SECONDS,
           java.util.concurrent.SynchronousQueue.new,
-          java.util.concurrent.ThreadPoolExecutor::AbortPolicy.new)
+          OVERFLOW_POLICIES[@overflow_policy].new)
       end
     end
   end

lib/concurrent/java_fixed_thread_pool.rb

@@ -1,7 +1,6 @@
 if defined? java.util
 
   require 'concurrent/java_thread_pool_executor'
-  require 'concurrent/utilities'
 
   module Concurrent
 
@@ -10,15 +9,25 @@ class JavaFixedThreadPool < JavaThreadPoolExecutor
 
       # Create a new thread pool.
       #
+      # @param [Hash] opts the options defining pool behavior.
+      # @option opts [Symbol] :overflow_policy (+:abort+) the overflow policy
+      #
+      # @raise [ArgumentError] if +num_threads+ is less than or equal to zero
+      # @raise [ArgumentError] if +overflow_policy+ is not a known policy
+      #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool-int-
-      def initialize(num_threads = Concurrent::processor_count)
+      def initialize(num_threads, opts = {})
+        @overflow_policy = opts.fetch(:overflow_policy, :abort)
+        @max_queue = 0
+
         raise ArgumentError.new('number of threads must be greater than zero') if num_threads < 1
+        raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(@overflow_policy)
 
         @executor = java.util.concurrent.ThreadPoolExecutor.new(
           num_threads, num_threads,
-          0, java.util.concurrent.TimeUnit::SECONDS,
+          @max_queue, java.util.concurrent.TimeUnit::SECONDS,
           java.util.concurrent.LinkedBlockingQueue.new,
-          java.util.concurrent.ThreadPoolExecutor::AbortPolicy.new)
+          OVERFLOW_POLICIES[@overflow_policy].new)
       end
     end
   end

lib/concurrent/java_thread_pool_executor.rb

@@ -2,6 +2,8 @@
 
   module Concurrent
 
+    RejectedExecutionError = Class.new(StandardError) unless defined? RejectedExecutionError
+
     # @!macro thread_pool_executor
     class JavaThreadPoolExecutor
 
@@ -30,31 +32,34 @@ class JavaThreadPoolExecutor
 
       attr_reader :max_queue
 
+      attr_reader :overflow_policy
+
       # Create a new thread pool.
       #
       # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
       def initialize(opts = {})
         min_length = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i
-        @max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
+        max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
         idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
         @max_queue = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i
-        overflow_policy = opts.fetch(:overflow_policy, :abort)
+        @overflow_policy = opts.fetch(:overflow_policy, :abort)
 
-        raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0
-        raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(overflow_policy)
+        raise ArgumentError.new('max_threads must be greater than zero') if max_length <= 0
+        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.keys.include?(@overflow_policy)
 
-        if min_length == 0 && max_queue == 0
+        if min_length == 0 && @max_queue == 0
           queue = java.util.concurrent.SynchronousQueue.new
-        elsif max_queue == 0
+        elsif @max_queue == 0
           queue = java.util.concurrent.LinkedBlockingQueue.new
         else
-          queue = java.util.concurrent.LinkedBlockingQueue.new(max_queue)
+          queue = java.util.concurrent.LinkedBlockingQueue.new(@max_queue)
         end
 
         @executor = java.util.concurrent.ThreadPoolExecutor.new(
-          min_length, @max_length,
+          min_length, max_length,
           idletime, java.util.concurrent.TimeUnit::SECONDS,
-          queue, OVERFLOW_POLICIES[overflow_policy].new)
+          queue, OVERFLOW_POLICIES[@overflow_policy].new)
       end
 
       def min_length
@@ -91,14 +96,14 @@ def queue_length
       end
 
       def remaining_capacity
-        @executor.getQueue.remainingCapacity
+        @max_queue == 0 ? -1 : @executor.getQueue.remainingCapacity
       end
 
       # Is the thread pool running?
       #
       # @return [Boolean] +true+ when running, +false+ when shutting down or shutdown
       def running?
-        ! (shutdown? || terminated?)
+        ! (@executor.isShutdown || @executor.isTerminated || @executor.isTerminating)
       end
 
       # Is the thread pool shutdown?
@@ -133,10 +138,14 @@ def wait_for_termination(timeout)
       # @raise [ArgumentError] if no task is given
       def post(*args)
         raise ArgumentError.new('no block given') unless block_given?
-        @executor.submit{ yield(*args) }
-        return true
+        if running?
+          @executor.submit{ yield(*args) }
+          true
+        else
+          false
+        end
       rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
-        return false
+        raise RejectedExecutionError
       end
 
       # Submit a task to the thread pool for asynchronous processing.
@@ -147,16 +156,15 @@ def post(*args)
       def <<(task)
         @executor.submit(&task)
       rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
-        # do nothing
-      ensure
-        return self
+        raise RejectedExecutionError
       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
+        @executor.getQueue.clear
         return nil
       end
 
@@ -166,17 +174,9 @@ def shutdown
       # not running.
       def kill
         @executor.shutdownNow
+        @executor.getQueue.clear
         return nil
       end
-
-      protected
-
-      # Were all tasks completed before shutdown?
-      #
-      # @return [Boolean] +true+ if shutdown and all tasks completed else +false+
-      def terminated?
-        @executor.isTerminated
-      end
     end
   end
 end

lib/concurrent/ruby_cached_thread_pool.rb

@@ -15,16 +15,20 @@ class RubyCachedThreadPool < RubyThreadPoolExecutor
     #
     # @raise [ArgumentError] if +max_threads+ is less than or equal to zero
     # @raise [ArgumentError] if +idletime+ is less than or equal to zero
+    # @raise [ArgumentError] if +overflow_policy+ is not a known policy
     def initialize(opts = {})
       max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
       idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
+      overflow_policy = opts.fetch(:overflow_policy, :abort)
 
       raise ArgumentError.new('idletime must be greater than zero') if idletime <= 0
       raise ArgumentError.new('max_threads must be greater than zero') if max_length <= 0
+      raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(overflow_policy)
 
       opts = opts.merge(
         min_threads: 0,
         max_threads: max_length,
+        num_threads: overflow_policy,
         idletime: idletime
       )
       super(opts)

lib/concurrent/ruby_fixed_thread_pool.rb

@@ -8,13 +8,21 @@ class RubyFixedThreadPool < RubyThreadPoolExecutor
     # Create a new thread pool.
     #
     # @param [Integer] num_threads the number of threads to allocate
+    # @param [Hash] opts the options defining pool behavior.
+    # @option opts [Symbol] :overflow_policy (+:abort+) the overflow policy
     #
     # @raise [ArgumentError] if +num_threads+ is less than or equal to zero
+    # @raise [ArgumentError] if +overflow_policy+ is not a known policy
     def initialize(num_threads, opts = {})
+      overflow_policy = opts.fetch(:overflow_policy, :abort)
+
       raise ArgumentError.new('number of threads must be greater than zero') if num_threads < 1
+      raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(overflow_policy)
+
       opts = opts.merge(
         min_threads: num_threads,
         max_threads: num_threads,
+        num_threads: overflow_policy,
         idletime: 0
       )
       super(opts)

lib/concurrent/ruby_thread_pool_executor.rb

@@ -5,6 +5,8 @@
 
 module Concurrent
 
+  RejectedExecutionError = Class.new(StandardError) unless defined? RejectedExecutionError
+
   # @!macro thread_pool_executor
   class RubyThreadPoolExecutor
 
@@ -37,6 +39,8 @@ class RubyThreadPoolExecutor
 
     attr_reader :max_queue
 
+    attr_reader :overflow_policy
+
     # Create a new thread pool.
     #
     # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
@@ -48,6 +52,7 @@ def initialize(opts = {})
       @overflow_policy = opts.fetch(:overflow_policy, :abort)
 
       raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0
+      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
@@ -59,7 +64,6 @@ def initialize(opts = {})
       @completed_task_count = 0
       @largest_length = 0
 
-      #@busy = 0
       @gc_interval = opts.fetch(:gc_interval, 1).to_i # undocumented
       @last_gc_time = Time.now.to_f - [1.0, (@gc_interval * 2.0)].max
     end
@@ -72,9 +76,11 @@ def length
     alias_method :current_length, :length
 
     def queue_length
+      @queue.length
     end
 
     def remaining_capacity
+      @mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }
     end
 
     # Is the thread pool running?
@@ -118,6 +124,7 @@ def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
       @mutex.synchronize do
         break false unless @state == :running
+        return handle_overflow(*args, &task) if @max_queue != 0 && @queue.length >= @max_queue
         @scheduled_task_count += 1
         @queue << [args, task]
         if Time.now.to_f - @gc_interval >= @last_gc_time
@@ -145,6 +152,7 @@ def <<(task)
     def shutdown
       @mutex.synchronize do
         break unless @state == :running
+        @queue.clear
         if @pool.empty?
           @state = :shutdown
           @terminator.set
@@ -162,24 +170,16 @@ def shutdown
     def kill
       @mutex.synchronize do
         break if @state == :shutdown
-        @state = :shutdown
         @queue.clear
+        @state = :shutdown
         drain_pool
         @terminator.set
       end
     end
 
-    ## @!visibility private
-    #def on_start_task # :nodoc:
-      #@mutex.synchronize do
-        #@busy += 1
-      #end
-    #end
-
     # @!visibility private
     def on_end_task # :nodoc:
       @mutex.synchronize do
-        #@busy -= 1
         @completed_task_count += 1 #if success
         break unless @state == :running
       end
@@ -198,6 +198,23 @@ def on_worker_exit(worker) # :nodoc:
 
     protected
 
+    # @!visibility private
+    def handle_overflow(*args) # :nodoc:
+      case @overflow_policy
+      when :abort
+        raise RejectedExecutionError
+      when :discard
+        false
+      when :caller_runs
+        begin
+          yield(*args)
+        rescue
+          # let it fail
+        end
+        true
+      end
+    end
+
     # @!visibility private
     def prune_pool # :nodoc:
       @pool.delete_if do |worker|

lib/concurrent/ruby_thread_pool_worker.rb

@@ -49,7 +49,6 @@ def run(thread = Thread.current) # :nodoc:
         end
 
         begin
-          #@parent.on_start_task
           task.last.call(*task.first)
         rescue => ex
           # let it fail