Implemented thread pool overload policies.

Author: jdantonio

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
 
@@ -37,13 +39,14 @@ class JavaThreadPoolExecutor
       # @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)
 
-        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
           queue = java.util.concurrent.SynchronousQueue.new
@@ -54,7 +57,7 @@ def initialize(opts = {})
         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)
       end
@@ -100,7 +103,7 @@ def remaining_capacity
       #
       # @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?
@@ -135,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.
@@ -149,17 +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.getQueue.clear
         @executor.shutdown
+        @executor.getQueue.clear
         return nil
       end
 
@@ -168,19 +173,10 @@ def shutdown
       # will be accepted. Has no additional effect if the thread pool is
       # not running.
       def kill
-        @executor.getQueue.clear
         @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
 
@@ -50,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
@@ -121,7 +124,7 @@ def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
       @mutex.synchronize do
         break false unless @state == :running
-        return false if @max_queue != 0 && @queue.length >= @max_queue
+        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
@@ -195,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|

spec/concurrent/cached_thread_pool_shared.rb

@@ -4,7 +4,12 @@
 share_examples_for :cached_thread_pool do
 
   let!(:max_threads){ 5 }
-  subject { described_class.new(max_threads: max_threads) }
+  subject do
+    described_class.new(
+      max_threads: max_threads,
+      overflow_policy: :discard
+    )
+  end
 
   after(:each) do
     subject.kill
@@ -106,14 +111,19 @@
 
   context 'worker creation and caching' do
 
-    subject{ described_class.new(idletime: 1, max_threads: 5) }
+    subject do
+      described_class.new(
+        idletime: 1,
+        max_threads: 5,
+        overflow_policy: :discard
+      )
+    end
 
     it 'never creates more than :max_threads threads' do
-      pool = described_class.new(max_threads: 5)
-      100.times{ pool << proc{ sleep(1) } }
+      100.times{ subject << proc{ sleep(1) } }
       sleep(0.1)
-      pool.length.should eq 5
-      pool.kill
+      subject.length.should eq 5
+      subject.kill
     end
   end
 end

spec/concurrent/java_cached_thread_pool_spec.rb

@@ -8,14 +8,57 @@ module Concurrent
 
     describe JavaCachedThreadPool do
 
-      subject { described_class.new(max_threads: 5) }
+      subject { described_class.new(max_threads: 5, overflow_policy: :discard) }
 
       after(:each) do
         subject.kill
         sleep(0.1)
       end
 
       it_should_behave_like :cached_thread_pool
+
+      context '#initialize' do
+
+        it 'sets :min_length correctly' do
+          subject = JavaCachedThreadPool.new
+          subject.min_length.should eq 0
+        end
+
+        it 'sets :max_length correctly' do
+          subject = JavaCachedThreadPool.new(max_threads: 5)
+          subject.max_length.should eq 5
+        end
+
+        it 'sets :idletime correctly' do
+          subject = JavaCachedThreadPool.new(idletime: 10)
+          subject.idletime.should eq 10
+        end
+
+        it 'sets :max_queue correctly' do
+          subject = JavaCachedThreadPool.new
+          subject.max_queue.should eq 0
+        end
+
+        it 'sets :overflow_policy correctly' do
+          clazz = java.util.concurrent.ThreadPoolExecutor::DiscardPolicy
+          policy = clazz.new
+          clazz.should_receive(:new).at_least(:once).with(any_args).and_return(policy)
+
+          subject = JavaCachedThreadPool.new(overflow_policy: :discard)
+          subject.overflow_policy.should eq :discard
+        end
+
+        it 'defaults :overflow_policy to :abort' do
+          subject = JavaCachedThreadPool.new
+          subject.overflow_policy.should eq :abort
+        end
+
+        it 'raises an exception if given an invalid :overflow_policy' do
+          expect {
+            JavaCachedThreadPool.new(overflow_policy: :bogus)
+          }.to raise_error(ArgumentError)
+        end
+      end
     end
   end
 end