Introduce ThreadPoolExecutor without a Queue

This adds the ability for a ThreadPoolExecutor to have a queue
depth of 0.

This is useful if you'd like to perform the rejection handler if no
available threads are availabile.

This is analogous to Java's use of SynchronousQueue in
ThreadPoolExecutor.

See https://stackoverflow.com/a/10186825/143733 for more details

Author: fzakaria

lib/concurrent-ruby/concurrent/executor/java_thread_pool_executor.rb

@@ -21,12 +21,18 @@ class JavaThreadPoolExecutor < JavaExecutorService
       # @!macro thread_pool_executor_constant_default_thread_timeout
       DEFAULT_THREAD_IDLETIMEOUT = 60
 
+      # @!macro thread_pool_executor_constant_default_synchronous
+      DEFAULT_SYNCHRONOUS = false
+
       # @!macro thread_pool_executor_attr_reader_max_length
       attr_reader :max_length
 
       # @!macro thread_pool_executor_attr_reader_max_queue
       attr_reader :max_queue
 
+      # @!macro thread_pool_executor_attr_reader_synchronous
+      attr_reader :synchronous
+
       # @!macro thread_pool_executor_method_initialize
       def initialize(opts = {})
         super(opts)
@@ -94,16 +100,22 @@ def ns_initialize(opts)
         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
+        @synchronous     = opts.fetch(:synchronous, DEFAULT_SYNCHRONOUS)
         @fallback_policy = opts.fetch(:fallback_policy, :abort)
 
+        raise ArgumentError.new("`synchronous` cannot be set unless `max_queue` is 0") if @synchronous && @max_queue > 0
         raise ArgumentError.new("`max_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if max_length < DEFAULT_MIN_POOL_SIZE
         raise ArgumentError.new("`max_threads` cannot be greater than #{DEFAULT_MAX_POOL_SIZE}") if max_length > DEFAULT_MAX_POOL_SIZE
         raise ArgumentError.new("`min_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if min_length < DEFAULT_MIN_POOL_SIZE
         raise ArgumentError.new("`min_threads` cannot be more than `max_threads`") if min_length > max_length
         raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICY_CLASSES.include?(@fallback_policy)
 
         if @max_queue == 0
-          queue = java.util.concurrent.LinkedBlockingQueue.new
+          if @synchronous
+            queue = java.util.concurrent.SynchronousQueue.new
+          else
+            queue = java.util.concurrent.LinkedBlockingQueue.new
+          end
         else
           queue = java.util.concurrent.LinkedBlockingQueue.new(@max_queue)
         end

lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb

@@ -3,6 +3,7 @@
 require 'concurrent/concern/logging'
 require 'concurrent/executor/ruby_executor_service'
 require 'concurrent/utility/monotonic_time'
+require 'concurrent/mvar'
 
 module Concurrent
 
@@ -23,6 +24,9 @@ class RubyThreadPoolExecutor < RubyExecutorService
     # @!macro thread_pool_executor_constant_default_thread_timeout
     DEFAULT_THREAD_IDLETIMEOUT = 60
 
+    # @!macro thread_pool_executor_constant_default_synchronous
+    DEFAULT_SYNCHRONOUS = false
+
     # @!macro thread_pool_executor_attr_reader_max_length
     attr_reader :max_length
 
@@ -35,6 +39,9 @@ class RubyThreadPoolExecutor < RubyExecutorService
     # @!macro thread_pool_executor_attr_reader_max_queue
     attr_reader :max_queue
 
+    # @!macro thread_pool_executor_attr_reader_synchronous
+    attr_reader :synchronous
+
     # @!macro thread_pool_executor_method_initialize
     def initialize(opts = {})
       super(opts)
@@ -114,9 +121,11 @@ def ns_initialize(opts)
       @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
+      @synchronous     = opts.fetch(:synchronous, DEFAULT_SYNCHRONOUS)
       @fallback_policy = opts.fetch(:fallback_policy, :abort)
-      raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
 
+      raise ArgumentError.new("`synchronous` cannot be set unless `max_queue` is 0") if @synchronous && @max_queue > 0
+      raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
       raise ArgumentError.new("`max_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if @max_length < DEFAULT_MIN_POOL_SIZE
       raise ArgumentError.new("`max_threads` cannot be greater than #{DEFAULT_MAX_POOL_SIZE}") if @max_length > DEFAULT_MAX_POOL_SIZE
       raise ArgumentError.new("`min_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if @min_length < DEFAULT_MIN_POOL_SIZE
@@ -145,8 +154,11 @@ def ns_limited_queue?
     def ns_execute(*args, &task)
       ns_reset_if_forked
 
-      if ns_assign_worker(*args, &task) || ns_enqueue(*args, &task)
+      assigned_worker = ns_assign_worker(*args, &task)
+      if assigned_worker
         @scheduled_task_count += 1
+      elsif !@synchronous
+        ns_enqueue(*args, &task)
       else
         handle_fallback(*args, &task)
       end
@@ -233,7 +245,7 @@ def ns_add_busy_worker
     #
     # @!visibility private
     def ns_ready_worker(worker, success = true)
-      task_and_args = @queue.shift
+        task_and_args = @queue.shift
       if task_and_args
         worker << task_and_args
       else

lib/concurrent-ruby/concurrent/executor/thread_pool_executor.rb

@@ -73,7 +73,8 @@ class ThreadPoolExecutor < ThreadPoolExecutorImplementation
     #   @option opts [Symbol] :fallback_policy (:abort) the policy for handling new
     #     tasks that are received when the queue size has reached
     #     `max_queue` or the executor has shut down
-    #
+    #   @options opts [Boolean] :synchronous (DEFAULT_SYNCHRONOUS) whether or not a value of 0
+    #     for :max_queue means the queue must perform direct hand-off rather than unbounded.
     #   @raise [ArgumentError] if `:max_threads` is less than one
     #   @raise [ArgumentError] if `:min_threads` is less than zero
     #   @raise [ArgumentError] if `:fallback_policy` is not one of the values specified

spec/concurrent/executor/thread_pool_executor_shared.rb

@@ -137,6 +137,55 @@
     end
   end
 
+  context '#synchronous' do
+
+    subject do
+      described_class.new(
+          min_threads: 1,
+          max_threads: 2,
+          max_queue: 0,
+          synchronous: true,
+          fallback_policy: :abort
+      )
+    end
+
+    it 'cannot be set unless `max_queue` is zero' do
+      expect {
+        described_class.new(
+            min_threads: 2,
+            max_threads: 5,
+            max_queue: 1,
+            fallback_policy: :discard,
+            synchronous: true
+        )
+      }.to raise_error(ArgumentError)
+    end
+
+    it 'executes fallback policy once max_threads has been reached' do
+      latch = Concurrent::CountDownLatch.new(1)
+      (subject.max_length).times do
+        subject.post {
+          latch.wait
+        }
+      end
+
+      expect(subject.queue_length).to eq 0
+
+      # verify nothing happening
+      20.times {
+        expect {
+          subject.post {
+            sleep
+          }
+        }.to raise_error(Concurrent::RejectedExecutionError)
+      }
+
+      # release
+      latch.count_down
+    end
+
+  end
+
   context '#queue_length', :truffle_bug => true do # only actually fails for RubyThreadPoolExecutor
 
     let!(:expected_max){ 10 }