Removed JavaAbstractThreadPool in lieu of JavaThreadPoolExecutor.

* Created JavaThreadPoolExecutor; some specs failing.
* JavaThreadPoolExecutor specs passing.
* Added shared specs for global thread pool.
* Removed JavaAbstractThreadPool in lieu of JavaThreadPoolExecutor.

Author: jdantonio

lib/concurrent.rb

@@ -34,6 +34,7 @@
 require 'concurrent/fixed_thread_pool'
 require 'concurrent/immediate_executor'
 require 'concurrent/per_thread_executor'
+require 'concurrent/thread_pool_executor'
 require 'concurrent/uses_global_thread_pool'
 
 # Modern concurrency tools for Ruby. Inspired by Erlang, Clojure, Scala, Haskell,

lib/concurrent/immediate_executor.rb

@@ -2,13 +2,14 @@ module Concurrent
   class ImmediateExecutor
 
     def post(*args, &block)
+      raise ArgumentError.new('no block given') unless block_given?
       block.call(*args)
+      return true
     end
 
     def <<(block)
       post(&block)
       self
     end
-
   end
-end
+end

lib/concurrent/java_cached_thread_pool.rb

@@ -1,12 +1,11 @@
 if defined? java.util
 
-  require 'concurrent/java_abstract_thread_pool'
+  require 'concurrent/java_thread_pool_executor'
 
   module Concurrent
 
     # @!macro cached_thread_pool
-    class JavaCachedThreadPool
-      include JavaAbstractThreadPool
+    class JavaCachedThreadPool < JavaThreadPoolExecutor
 
       # The maximum number of threads that may be created in the pool
       # (unless overridden during construction).
@@ -29,22 +28,11 @@ def initialize(opts = {})
         @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
     end
   end

lib/concurrent/java_fixed_thread_pool.rb

@@ -1,13 +1,12 @@
 if defined? java.util
 
-  require 'concurrent/java_abstract_thread_pool'
+  require 'concurrent/java_thread_pool_executor'
   require 'concurrent/utilities'
 
   module Concurrent
 
     # @!macro fixed_thread_pool
-    class JavaFixedThreadPool
-      include JavaAbstractThreadPool
+    class JavaFixedThreadPool < JavaThreadPoolExecutor
 
       # Create a new thread pool.
       #
@@ -16,22 +15,11 @@ 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)
         @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)
-
-        #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
     end
   end

lib/concurrent/java_abstract_thread_pool.rb renamed to lib/concurrent/java_thread_pool_executor.rb

@@ -2,8 +2,60 @@
 
   module Concurrent
 
-    # @!macro cached_thread_pool
-    module JavaAbstractThreadPool
+    # @!macro thread_pool_executor
+    class JavaThreadPoolExecutor
+
+      # The maximum number of threads that will be created in the pool
+      # (unless overridden during construction).
+      DEFAULT_MAX_POOL_SIZE = java.lang.Integer::MAX_VALUE # 2147483647
+
+      # The minimum number of threads that will be created in the pool
+      # (unless overridden during construction).
+      DEFAULT_MIN_POOL_SIZE = 0
+
+      DEFAULT_MAX_QUEUE_SIZE = 0
+
+      # The maximum number of seconds a thread in the pool may remain idle before
+      # being reclaimed (unless overridden during construction).
+      DEFAULT_THREAD_IDLETIMEOUT = 60
+
+      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
+
+      attr_reader :max_queue
+
+      # 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
+        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)
+
+        if min_length == 0 && max_queue == 0
+          queue = java.util.concurrent.SynchronousQueue.new
+        elsif max_queue == 0
+          queue = java.util.concurrent.LinkedBlockingQueue.new
+        else
+          queue = java.util.concurrent.LinkedBlockingQueue.new(max_queue)
+        end
+
+        @executor = java.util.concurrent.ThreadPoolExecutor.new(
+          min_length, @max_length,
+          idletime, java.util.concurrent.TimeUnit::SECONDS,
+          queue, OVERFLOW_POLICIES[overflow_policy].new)
+      end
 
       def min_length
         @executor.getCorePoolSize
@@ -34,6 +86,14 @@ def idletime
         @executor.getKeepAliveTime(java.util.concurrent.TimeUnit::SECONDS)
       end
 
+      def queue_length
+        @executor.getQueue.size
+      end
+
+      def remaining_capacity
+        @executor.getQueue.remainingCapacity
+      end
+
       # Is the thread pool running?
       #
       # @return [Boolean] +true+ when running, +false+ when shutting down or shutdown

lib/concurrent/per_thread_executor.rb

@@ -3,6 +3,7 @@ module Concurrent
   class PerThreadExecutor
 
     def self.post(*args)
+      raise ArgumentError.new('no block given') unless block_given?
       Thread.new(*args) do
         Thread.current.abort_on_exception = false
         yield(*args)

lib/concurrent/ruby_fixed_thread_pool.rb

@@ -81,7 +81,7 @@ def wait_for_termination(timeout)
     #
     # @raise [ArgumentError] if no block is given
     def post(*args, &task)
-      raise ArgumentError.new('no block given') if task.nil?
+      raise ArgumentError.new('no block given') unless block_given?
       @mutex.synchronize do
         break false unless @state == :running
         @scheduled_task_count += 1

lib/concurrent/thread_pool_executor.rb

@@ -0,0 +1,30 @@
+#require 'concurrent/ruby_thread_pool_executor'
+
+module Concurrent
+
+  if defined? java.util
+    require 'concurrent/java_thread_pool_executor'
+    # @!macro [attach] thread_pool_executor
+    #
+    #   A thread pool...
+    #  
+    #   The API and behavior of this class are based on Java's +ThreadPoolExecutor+
+    #
+    #   @note When running on the JVM (JRuby) this class will inherit from +JavaThreadPoolExecutor+.
+    #     On all other platforms it will inherit from +RubyThreadPoolExecutor+.
+    #
+    #   @see Concurrent::RubyThreadPoolExecutor
+    #   @see Concurrent::JavaThreadPoolExecutor
+    #  
+    #   @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
+    #   @see http://stackoverflow.com/questions/17957382/fixedthreadpool-vs-fixedthreadpool-the-lesser-of-two-evils
+    class ThreadPoolExecutor < JavaThreadPoolExecutor
+    end
+  else
+    ## @!macro thread_pool_executor
+    #class ThreadPoolExecutor < RubyThreadPoolExecutor
+    #end
+  end
+end

spec/concurrent/cached_thread_pool_shared.rb

@@ -23,7 +23,7 @@
 
     it 'raises an exception when the pool size is less than one' do
       lambda {
-        described_class.new(max: 0)
+        described_class.new(max_threads: 0)
       }.should raise_error(ArgumentError)
     end
 
@@ -74,37 +74,24 @@
     end
   end
 
-  context '#length' do
-
-    it 'returns zero for a new thread pool' do
-      subject.length.should eq 0
-    end
-
-    it 'returns :max_length while running' do
-      10.times{ subject.post{ sleep(0.1) } }
-      sleep(1)
-      subject.length.should eq max_threads
-    end
-  end
-
   context '#largest_length' do
 
     it 'returns zero on creation' do
       subject.largest_length.should eq 0
     end
 
-    it 'returns :max_threads while running' do
+    it 'returns a non-zero number once tasks have been received' do
       10.times{ subject.post{ sleep(0.1) } }
-      sleep(1)
-      subject.largest_length.should eq max_threads
+      sleep(0.1)
+      subject.largest_length.should > 0
     end
 
-    it 'returns :max_threads once shutdown' do
+    it 'returns a non-zero number after shutdown if tasks have been received' do
       10.times{ subject.post{ sleep(0.1) } }
-      sleep(1)
+      sleep(0.1)
       subject.shutdown
       subject.wait_for_termination(1)
-      subject.largest_length.should eq max_threads
+      subject.largest_length.should > 0
     end
   end
 
@@ -119,10 +106,11 @@
 
   context 'garbage collection' do
 
-    subject{ described_class.new(idletime: 1) }
+    subject{ described_class.new(idletime: 1, max_threads: 5) }
 
     it 'removes from pool any thread that has been idle too long' do
       3.times { subject << proc{ sleep(0.1) } }
+      sleep(0.1)
       subject.length.should eq 3
       sleep(2)
       subject << proc{ nil }
@@ -131,7 +119,8 @@
 
     it 'removes from pool any dead thread' do
       3.times { subject << proc{ sleep(0.1); raise Exception } }
-      subject.length.should == 3
+      sleep(0.1)
+      subject.length.should eq 3
       sleep(2)
       subject << proc{ nil }
       subject.length.should < 3
@@ -157,7 +146,7 @@
     end
 
     it 'never creates more than :max_threads threads' do
-      pool = described_class.new(max: 5)
+      pool = described_class.new(max_threads: 5)
       100.times{ sleep(0.01); pool << proc{ sleep(1) } }
       sleep(0.1)
       pool.length.should eq 5

spec/concurrent/global_thread_pool_shared.rb

@@ -0,0 +1,33 @@
+require 'spec_helper'
+
+share_examples_for :global_thread_pool do
+
+  context '#post' do
+
+    it 'raises an exception if no block is given' do
+      lambda {
+        subject.post
+      }.should raise_error(ArgumentError)
+    end
+
+    it 'returns true when the block is added to the queue' do
+      subject.post{ nil }.should be_true
+    end
+
+    it 'calls the block with the given arguments' do
+      @expected = nil
+      subject.post(1, 2, 3) do |a, b, c|
+        @expected = a + b + c
+      end
+      sleep(0.1)
+      @expected.should eq 6
+    end
+
+    it 'aliases #<<' do
+      @expected = false
+      subject << proc { @expected = true }
+      sleep(0.1)
+      @expected.should be_true
+    end
+  end
+end