Unify overflow/fallback handling between all four Executors

Author: rkday

lib/concurrent/executor/executor.rb

@@ -5,7 +5,11 @@
 module Concurrent
 
   module Executor
-    
+    # The policy defining how rejected tasks (tasks received once the queue size reaches
+    # the configured `max_queue`) are handled. Must be one of the values specified in
+    # `OVERFLOW_POLICIES`.
+    attr_reader :overflow_policy
+
     # @!macro [attach] executor_module_method_can_overflow_question
     #
     #   Does the task queue have a maximum size?
@@ -17,6 +21,31 @@ def can_overflow?
       false
     end
 
+    # Handler which executes the `overflow_policy` once the queue size
+    # reaches `max_queue`.
+    #
+    # @param [Array] args the arguments to the task which is being handled.
+    #
+    # @!visibility private
+    def handle_overflow(*args)
+      case @overflow_policy
+      when :abort
+        raise RejectedExecutionError
+      when :discard
+        false
+      when :caller_runs
+        begin
+          yield(*args)
+        rescue => ex
+          # let it fail
+          log DEBUG, ex
+        end
+        true
+      else
+        fail "Unknown overflow policy #{@overflow_policy}"
+      end
+    end
+
     # @!macro [attach] executor_module_method_serialized_question
     #
     #   Does this executor guarantee serialization of its operations?
@@ -63,6 +92,9 @@ module RubyExecutor
     include Executor
     include Logging
 
+    # The set of possible overflow policies that may be set at thread pool creation.
+    OVERFLOW_POLICIES          = [:abort, :discard, :caller_runs]
+
     # @!macro [attach] executor_method_post
     #
     #   Submit a task to the executor for asynchronous processing.
@@ -78,16 +110,8 @@ module RubyExecutor
     def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
       mutex.synchronize do
-        unless running?
-          # The executor is shut down - figure out how to reject this task
-          if self.respond_to?(:handle_overflow, true)
-            # Reject this task in the same way we'd reject an overflow
-            return handle_overflow(*args, &task)
-          else
-            # No handle_overflow method defined - just return false
-            return false
-          end
-        end
+        # If the executor is shut down, reject this task
+        return handle_overflow(*args, &task) unless running?
         execute(*args, &task)
         true
       end
@@ -219,16 +243,20 @@ module JavaExecutor
       include Executor
       java_import 'java.lang.Runnable'
 
+      # The set of possible overflow policies that may be set at thread pool creation.
+      OVERFLOW_POLICIES = {
+        abort: java.util.concurrent.ThreadPoolExecutor::AbortPolicy,
+        discard: java.util.concurrent.ThreadPoolExecutor::DiscardPolicy,
+        caller_runs: java.util.concurrent.ThreadPoolExecutor::CallerRunsPolicy
+      }.freeze
+
       # @!macro executor_method_post
       def post(*args)
         raise ArgumentError.new('no block given') unless block_given?
-        if running?
-          executor_submit = @executor.java_method(:submit, [Runnable.java_class])
-          executor_submit.call { yield(*args) }
-          true
-        else
-          false
-        end
+        return handle_overflow(*args, &task) unless running?
+        executor_submit = @executor.java_method(:submit, [Runnable.java_class])
+        executor_submit.call { yield(*args) }
+        true
       rescue Java::JavaUtilConcurrent::RejectedExecutionException
         raise RejectedExecutionError
       end

lib/concurrent/executor/java_single_thread_executor.rb

@@ -10,11 +10,16 @@ class JavaSingleThreadExecutor
 
       # Create a new thread pool.
       #
+      # @option opts [Symbol] :overflow_policy (:discard) the policy for handling new
+      #   tasks that are received when the queue size has reached `max_queue`
+      #
       # @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
       def initialize(opts = {})
         @executor = java.util.concurrent.Executors.newSingleThreadExecutor
+        @overflow_policy = opts.fetch(:overflow_policy, :discard)
+        raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(@overflow_policy)
         set_shutdown_hook
       end
     end

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -20,13 +20,6 @@ class JavaThreadPoolExecutor
       # before being reclaimed.
       DEFAULT_THREAD_IDLETIMEOUT = 60
 
-      # The set of possible overflow policies that may be set at thread pool creation.
-      OVERFLOW_POLICIES = {
-        abort: java.util.concurrent.ThreadPoolExecutor::AbortPolicy,
-        discard: java.util.concurrent.ThreadPoolExecutor::DiscardPolicy,
-        caller_runs: java.util.concurrent.ThreadPoolExecutor::CallerRunsPolicy
-      }.freeze
-
       # The maximum number of threads that may be created in the pool.
       attr_reader :max_length
 

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -9,12 +9,17 @@ class RubySingleThreadExecutor
 
     # Create a new thread pool.
     #
+    # @option opts [Symbol] :overflow_policy (:discard) the policy for handling new
+    #   tasks that are received when the queue size has reached `max_queue`
+    #
     # @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
     def initialize(opts = {})
       @queue = Queue.new
       @thread = nil
+      @overflow_policy = opts.fetch(:overflow_policy, :discard)
+      raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(@overflow_policy)
       init_executor
     end
 

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -23,9 +23,6 @@ class RubyThreadPoolExecutor
     # before being reclaimed.
     DEFAULT_THREAD_IDLETIMEOUT = 60
 
-    # The set of possible overflow policies that may be set at thread pool creation.
-    OVERFLOW_POLICIES          = [:abort, :discard, :caller_runs]
-
     # The maximum number of threads that may be created in the pool.
     attr_reader :max_length
 
@@ -49,11 +46,6 @@ class RubyThreadPoolExecutor
     # accordance with the configured `overflow_policy`.
     attr_reader :max_queue
 
-    # The policy defining how rejected tasks (tasks received once the queue size reaches
-    # the configured `max_queue`) are handled. Must be one of the values specified in
-    # `OVERFLOW_POLICIES`.
-    attr_reader :overflow_policy
-
     # Create a new thread pool.
     #
     # @param [Hash] opts the options which configure the thread pool
@@ -224,29 +216,6 @@ def ensure_capacity?
       capacity
     end
 
-    # Handler which executes the `overflow_policy` once the queue size
-    # reaches `max_queue`.
-    #
-    # @param [Array] args the arguments to the task which is being handled.
-    #
-    # @!visibility private
-    def handle_overflow(*args)
-      case @overflow_policy
-      when :abort
-        raise RejectedExecutionError
-      when :discard
-        false
-      when :caller_runs
-        begin
-          yield(*args)
-        rescue => ex
-          # let it fail
-          log DEBUG, ex
-        end
-        true
-      end
-    end
-
     # Scan all threads in the pool and reclaim any that are dead or
     # have been idle too long. Will check the last time the pool was
     # pruned and only run if the configured garbage collection