Merge pull request #39 from jdantonio/thread-pool-executor

 `RubyThreadPoolExecutor` class

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,50 +1,36 @@
 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
-
-      # The maximum number of threads that may be created in the pool
-      # (unless overridden during construction).
-      DEFAULT_MAX_POOL_SIZE = java.lang.Integer::MAX_VALUE # 2147483647
-
-      # The maximum number of seconds a thread in the pool may remain idle before
-      # being reclaimed (unless overridden during construction).
-      DEFAULT_THREAD_IDLETIME = 60
-
-      # The maximum number of threads that may be created in the pool.
-      attr_reader :max_length
+    class JavaCachedThreadPool < JavaThreadPoolExecutor
 
       # Create a new thread pool.
       #
+      # @param [Hash] opts the options defining pool behavior.
+      # @option opts [Integer] :max_threads (+DEFAULT_MAX_POOL_SIZE+) maximum number
+      #   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
+      #
+      # @raise [ArgumentError] if +max_threads+ is less than or equal to zero
+      # @raise [ArgumentError] if +idletime+ is less than or equal to zero
+      #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--
       def initialize(opts = {})
-        idletime = (opts[:thread_idletime] || opts[:idletime] || DEFAULT_THREAD_IDLETIME).to_i
-        raise ArgumentError.new('idletime must be greater than zero') if idletime <= 0
+        max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
+        idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
 
-        @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
+        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
 
-        #@executor = java.util.concurrent.Executors.newCachedThreadPool
         @executor = java.util.concurrent.ThreadPoolExecutor.new(
-          0, @max_length,
+          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,37 +1,24 @@
 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.
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool-int-
       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
+        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,
+          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
@@ -48,13 +108,6 @@ def shutdown?
         @executor.isShutdown
       end
 
-      # Were all tasks completed before shutdown?
-      #
-      # @return [Boolean] +true+ if shutdown and all tasks completed else +false+
-      def terminated?
-        @executor.isTerminated
-      end
-
       # Block until thread pool shutdown is complete or until +timeout+ seconds have
       # passed.
       #
@@ -115,6 +168,15 @@ def kill
         @executor.shutdownNow
         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/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)