Improved performance of Java thread pools

Author: jdantonio

examples/benchmark_thread_pool_implementations.rb

@@ -0,0 +1,23 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'benchmark'
+require 'concurrent/executors'
+
+COUNT = 100_000
+
+EXECUTORS = [
+  Concurrent::JavaThreadPoolExecutor
+]
+
+def test_executor(executor_class, count)
+  executor = executor_class.new
+  count.times { executor.post{} }
+end
+
+Benchmark.bmbm do |x|
+  EXECUTORS.each do |executor_class|
+    x.report(executor_class.to_s) { test_executor(executor_class, COUNT) }
+  end
+end

lib/concurrent/executor/executor_service.rb

@@ -250,13 +250,13 @@ class JavaExecutorService < AbstractExecutorService
 
       def initialize(*args, &block)
         super
+        ns_make_executor_runnable
       end
 
       def post(*args, &task)
         raise ArgumentError.new('no block given') unless block_given?
         return handle_fallback(*args, &task) unless running?
-        executor_submit = @executor.java_method(:submit, [Runnable.java_class])
-        executor_submit.call { yield(*args) }
+        @executor.submit_runnable Job.new(args, task)
         true
       rescue Java::JavaUtilConcurrent::RejectedExecutionException
         raise RejectedExecutionError
@@ -304,6 +304,27 @@ def ns_shuttingdown?
       def ns_shutdown?
         @executor.isShutdown || @executor.isTerminated
       end
+
+      def ns_make_executor_runnable
+        if !defined?(@executor.submit_runnable)
+          @executor.class.class_eval do
+            java_alias :submit_runnable, :submit, [java.lang.Runnable.java_class]
+          end
+        end
+      end
+
+      class Job
+        include Runnable
+        def initialize(args, block)
+          @args = args
+          @block = block
+        end
+
+        def run
+          @block.call(*@args)
+        end
+      end
+      private_constant :Job
     end
   end
 end

lib/concurrent/executor/java_cached_thread_pool.rb

@@ -17,8 +17,12 @@ class JavaCachedThreadPool < JavaThreadPoolExecutor
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--
       def initialize(opts = {})
-        super()
+        super(opts)
+      end
+
+      protected
 
+      def ns_initialize(opts)
         @fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
         warn '[DEPRECATED] :overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)
         @max_queue = 0

lib/concurrent/executor/java_fixed_thread_pool.rb

@@ -24,8 +24,6 @@ def initialize(num_threads, opts = {})
             max_threads: num_threads
         }.merge(opts)
         super(opts)
-
-        self.auto_terminate = opts.fetch(:auto_terminate, true)
       end
     end
   end

lib/concurrent/executor/java_single_thread_executor.rb

@@ -23,6 +23,10 @@ def initialize(opts = {})
 
       protected
 
+      def ns_initialize(opts)
+        super(opts)
+      end
+
       def ns_initialize(opts)
         @executor = java.util.concurrent.Executors.newSingleThreadExecutor
         @fallback_policy = opts.fetch(:fallback_policy, :discard)