Improved initialization synchronization of ExecutorService subclasses.

Author: jdantonio

lib/concurrent/executor/executor_service.rb

@@ -145,7 +145,7 @@ def kill_execution
       # do nothing
     end
 
-    private
+    protected
 
     def ns_auto_terminate?
       !!@auto_terminate

lib/concurrent/executor/java_single_thread_executor.rb

@@ -18,7 +18,12 @@ class JavaSingleThreadExecutor < JavaExecutorService
       # @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 = {})
-        super()
+        super(opts)
+      end
+
+      protected
+      
+      def ns_initialize(opts)
         @executor = java.util.concurrent.Executors.newSingleThreadExecutor
         @fallback_policy = opts.fetch(:fallback_policy, :discard)
         raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICY_CLASSES.keys.include?(@fallback_policy)

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -52,32 +52,7 @@ class JavaThreadPoolExecutor < JavaExecutorService
       #
       # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
       def initialize(opts = {})
-        super()
-
-        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
-        @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)
-
-        raise ArgumentError.new('max_threads must be greater than zero') if max_length <= 0
-        raise ArgumentError.new('min_threads cannot be less than zero') if min_length < 0
-        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
-        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, FALLBACK_POLICY_CLASSES[@fallback_policy].new)
-
-        self.auto_terminate = opts.fetch(:auto_terminate, true)
+        super(opts)
       end
 
       # @!macro executor_module_method_can_overflow_question
@@ -155,6 +130,35 @@ def remaining_capacity
       def running?
         super && !@executor.isTerminating
       end
+
+      protected
+
+      def ns_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
+        @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)
+
+        raise ArgumentError.new('max_threads must be greater than zero') if max_length <= 0
+        raise ArgumentError.new('min_threads cannot be less than zero') if min_length < 0
+        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
+        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, FALLBACK_POLICY_CLASSES[@fallback_policy].new)
+
+        self.auto_terminate = opts.fetch(:auto_terminate, true)
+      end
     end
   end
 end

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -18,16 +18,19 @@ class RubySingleThreadExecutor < RubyExecutorService
     # @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 = {})
-      super()
+      super
+    end
+
+    protected
+
+    def ns_initialize(opts)
       @queue = Queue.new
       @thread = nil
       @fallback_policy = opts.fetch(:fallback_policy, :discard)
       raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
       self.auto_terminate = opts.fetch(:auto_terminate, true)
     end
 
-    protected
-
     # @!visibility private
     def execute(*args, &task)
       supervise

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -128,7 +128,7 @@ def worker_died(worker)
 
     protected
 
-    def ns_initialize(opts = {})
+    def ns_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
@@ -345,6 +345,5 @@ def run_task(pool, task, args)
         throw :stop
       end
     end
-
   end
 end

lib/concurrent/executor/serialized_execution.rb

@@ -61,7 +61,7 @@ def posts(posts)
       true
     end
 
-    private
+    protected
 
     def ns_initialize
       @being_executed = false

lib/concurrent/executor/timer_set.rb

@@ -24,12 +24,7 @@ class TimerSet < RubyExecutorService
     #     `:operation` returns the global operation pool, and `:immediate` returns a new
     #     `ImmediateExecutor` object.
     def initialize(opts = {})
-      super()
-      @queue          = PriorityQueue.new(order: :min)
-      @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
-      @timer_executor = SingleThreadExecutor.new
-      @condition      = Event.new
-      self.auto_terminate = opts.fetch(:auto_terminate, true)
+      super(opts)
     end
 
     # Post a task to be execute run after a given delay (in seconds). If the
@@ -99,7 +94,7 @@ def self.calculate_delay!(delay)
       end
     end
 
-    private
+    protected
 
     # A struct for encapsulating a task and its intended execution time.
     # It facilitates proper prioritization by overriding the comparison
@@ -117,6 +112,14 @@ def <=>(other)
 
     private_constant :Task
 
+    def ns_initialize(opts)
+      @queue          = PriorityQueue.new(order: :min)
+      @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
+      @timer_executor = SingleThreadExecutor.new
+      @condition      = Event.new
+      self.auto_terminate = opts.fetch(:auto_terminate, true)
+    end
+
     # @!visibility private
     def shutdown_execution
       @queue.clear

lib/concurrent/timer_task.rb

@@ -267,11 +267,16 @@ def timeout_interval=(value)
       end
     end
 
+    if Concurrent.on_jruby?
+      #FIXME: JRuby seems to handle privacy different here
+      public :synchronize
+    end
+
     private :post, :<<
 
     protected
 
-    def ns_initialize(opts = {}, &task)
+    def ns_initialize(opts, &task)
       init_mutex(self)
       set_deref_options(opts)