Improved initialization synchronization.

Author: jdantonio

ext/com/concurrent_ruby/ext/SynchronizationLibrary.java

@@ -55,14 +55,14 @@ public IRubyObject initialize(ThreadContext context, IRubyObject[] args, Block b
             }
         }
 
-        @JRubyMethod(name = "synchronize", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "synchronize", visibility = Visibility.PROTECTED)
         public IRubyObject rubySynchronize(ThreadContext context, Block block) {
             synchronized (this) {
                 return block.yield(context, null);
             }
         }
 
-        @JRubyMethod(name = "ns_wait", optional = 1, visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ns_wait", optional = 1, visibility = Visibility.PROTECTED)
         public IRubyObject nsWait(ThreadContext context, IRubyObject[] args) {
             Ruby runtime = context.runtime;
             if (args.length > 1) {
@@ -94,19 +94,19 @@ public IRubyObject nsWait(ThreadContext context, IRubyObject[] args) {
             return this;
         }
 
-        @JRubyMethod(name = "ns_signal", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ns_signal", visibility = Visibility.PROTECTED)
         public IRubyObject nsSignal(ThreadContext context) {
             notify();
             return this;
         }
 
-        @JRubyMethod(name = "ns_broadcast", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ns_broadcast", visibility = Visibility.PROTECTED)
         public IRubyObject nsBroadcast(ThreadContext context) {
             notifyAll();
             return this;
         }
 
-        @JRubyMethod(name = "ensure_ivar_visibility!", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ensure_ivar_visibility!", visibility = Visibility.PROTECTED)
         public IRubyObject ensureIvarVisibilityBang(ThreadContext context) {
             return context.nil;
         }

lib/concurrent/actor/core.rb

@@ -47,58 +47,7 @@ class Core < Synchronization::Object
       #   any logging system
       # @param [Proc] block for class instantiation
       def initialize(opts = {}, &block)
-        super(&nil) # TODO use ns_initialize
-        synchronize do
-          @mailbox              = Array.new
-          @serialized_execution = SerializedExecution.new
-          @children             = Set.new
-
-          @context_class = Child! opts.fetch(:class), AbstractContext
-          allocate_context
-
-          @executor = Type! opts.fetch(:executor, Concurrent.global_io_executor), Executor
-          raise ArgumentError, 'ImmediateExecutor is not supported' if @executor.is_a? ImmediateExecutor
-
-          @reference = (Child! opts[:reference_class] || @context.default_reference_class, Reference).new self
-          @name      = (Type! opts.fetch(:name), String, Symbol).to_s
-
-          parent       = opts[:parent]
-          @parent_core = (Type! parent, Reference, NilClass) && parent.send(:core)
-          if @parent_core.nil? && @name != '/'
-            raise 'only root has no parent'
-          end
-
-          @path   = @parent_core ? File.join(@parent_core.path, @name) : @name
-          @logger = opts[:logger]
-
-          @parent_core.add_child reference if @parent_core
-
-          initialize_behaviours opts
-
-          @args       = opts.fetch(:args, [])
-          @block      = block
-          initialized = Type! opts[:initialized], Edge::CompletableFuture, NilClass
-
-          messages = []
-          messages << :link if opts[:link]
-          messages << :supervise if opts[:supervise]
-
-          schedule_execution do
-            begin
-              build_context
-
-              messages.each do |message|
-                handle_envelope Envelope.new(message, nil, parent, reference)
-              end
-
-              initialized.success reference if initialized
-            rescue => ex
-              log ERROR, ex
-              @first_behaviour.terminate!
-              initialized.fail ex if initialized
-            end
-          end
-        end
+        super
       end
 
       # @return [Reference, nil] of parent actor
@@ -199,6 +148,60 @@ def build_context
         @context.send :initialize, *@args, &@block
       end
 
+      protected
+
+      def ns_initialize(opts, &block)
+        @mailbox              = Array.new
+        @serialized_execution = SerializedExecution.new
+        @children             = Set.new
+
+        @context_class = Child! opts.fetch(:class), AbstractContext
+        allocate_context
+
+        @executor = Type! opts.fetch(:executor, Concurrent.global_io_executor), Executor
+        raise ArgumentError, 'ImmediateExecutor is not supported' if @executor.is_a? ImmediateExecutor
+
+        @reference = (Child! opts[:reference_class] || @context.default_reference_class, Reference).new self
+        @name      = (Type! opts.fetch(:name), String, Symbol).to_s
+
+        parent       = opts[:parent]
+        @parent_core = (Type! parent, Reference, NilClass) && parent.send(:core)
+        if @parent_core.nil? && @name != '/'
+          raise 'only root has no parent'
+        end
+
+        @path   = @parent_core ? File.join(@parent_core.path, @name) : @name
+        @logger = opts[:logger]
+
+        @parent_core.add_child reference if @parent_core
+
+        initialize_behaviours opts
+
+        @args       = opts.fetch(:args, [])
+        @block      = block
+        initialized = Type! opts[:initialized], Edge::CompletableFuture, NilClass
+
+        messages = []
+        messages << :link if opts[:link]
+        messages << :supervise if opts[:supervise]
+
+        schedule_execution do
+          begin
+            build_context
+
+            messages.each do |message|
+              handle_envelope Envelope.new(message, nil, parent, reference)
+            end
+
+            initialized.success reference if initialized
+          rescue => ex
+            log ERROR, ex
+            @first_behaviour.terminate!
+            initialized.fail ex if initialized
+          end
+        end
+      end
+
       private
 
       def handle_envelope(envelope)
@@ -215,7 +218,7 @@ def initialize_behaviours(opts)
         end
         @behaviours           = {}
         @first_behaviour      = @behaviour_definition.reverse.
-            reduce(nil) { |last, (behaviour, args)| @behaviours[behaviour] = behaviour.new(self, last, *args) }
+          reduce(nil) { |last, (behaviour, args)| @behaviours[behaviour] = behaviour.new(self, last, *args) }
       end
     end
   end

lib/concurrent/atomic/count_down_latch.rb

@@ -58,7 +58,7 @@ def count
       synchronize { @count }
     end
 
-    private
+    protected
 
     def ns_initialize(count)
       @count = count

lib/concurrent/atomic/cyclic_barrier.rb

@@ -62,7 +62,6 @@ def wait(timeout = nil)
       end
     end
 
-
     # resets the barrier to its initial state
     # If there is at least one waiting thread, it will be woken up, the `wait`
     # method will return false and the barrier will be broken
@@ -83,7 +82,7 @@ def broken?
       synchronize { @generation.status != :waiting }
     end
 
-    private
+    protected
 
     def ns_generation_done(generation, status, continue = true)
       generation.status = status
@@ -101,6 +100,5 @@ def ns_initialize(parties, &block)
       @action  = block
       ns_next_generation
     end
-
   end
 end

lib/concurrent/atomic/event.rb

@@ -70,7 +70,7 @@ def wait(timeout = nil)
       end
     end
 
-    private
+    protected
 
     def ns_set
       unless @set

lib/concurrent/delay.rb

@@ -68,19 +68,9 @@ class Delay < Synchronization::Object
     # @raise [ArgumentError] if no block is given
     def initialize(opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
-
-      super()
-      init_obligation(self)
-      set_deref_options(opts)
-      @task_executor = Executor.executor_from_options(opts)
-
-      @task      = block
-      @state     = :pending
-      @computing = false
+      super(opts, &block)
     end
 
-    protected :synchronize
-
     # Return the value this object represents after applying the options
     # specified by the `#set_deref_options` method. If the delayed operation
     # raised an exception this method will return nil. The execption object
@@ -169,6 +159,18 @@ def reconfigure(&block)
       end
     end
 
+    protected
+
+    def ns_initialize(opts, &block)
+      init_obligation(self)
+      set_deref_options(opts)
+      @task_executor = Executor.executor_from_options(opts)
+
+      @task      = block
+      @state     = :pending
+      @computing = false
+    end
+
     private
 
     # @!visibility private

lib/concurrent/executor/per_thread_executor.rb

@@ -19,13 +19,6 @@ module Concurrent
   class PerThreadExecutor < Synchronization::Object
     include Executor
 
-    # Creates a new executor
-    def initialize
-      @running = Concurrent::AtomicBoolean.new(true)
-      @stopped = Concurrent::Event.new
-      @count = Concurrent::AtomicFixnum.new(0)
-    end
-
     # @!macro executor_method_post
     def self.post(*args)
       raise ArgumentError.new('no block given') unless block_given?
@@ -97,5 +90,13 @@ def kill
     def wait_for_termination(timeout = nil)
       @stopped.wait(timeout)
     end
+
+    protected
+
+    def ns_initialize
+      @running = Concurrent::AtomicBoolean.new(true)
+      @stopped = Concurrent::Event.new
+      @count = Concurrent::AtomicFixnum.new(0)
+    end
   end
 end

lib/concurrent/ivar.rb

@@ -57,7 +57,8 @@ class IVar < Synchronization::Object
     # @option opts [String] :copy_on_deref (nil) call the given `Proc` passing
     #   the internal value and returning the value returned from the proc
     def initialize(value = NO_VALUE, opts = {})
-      super(&nil)
+      #FIXME use ns_initialize
+      super(&:nil)
       init_obligation(self)
       self.observers = CopyOnWriteObserverSet.new
       set_deref_options(opts)
@@ -88,7 +89,7 @@ def add_observer(observer = nil, func = :update, &block)
         func = :call
       end
 
-      mutex.synchronize do
+      synchronize do
         if event.set?
           direct_notification = true
         else
@@ -151,7 +152,7 @@ def try_set(value = NO_VALUE, &block)
 
     # @!visibility private
     def complete(success, value, reason) # :nodoc:
-      mutex.synchronize do
+      synchronize do
         raise MultipleAssignmentError if [:fulfilled, :rejected].include? @state
         set_state(success, value, reason)
         event.set

lib/concurrent/timer_task.rb

@@ -267,11 +267,6 @@ def timeout_interval=(value)
       end
     end
 
-    if Concurrent.on_jruby?
-      #FIXME: JRuby seems to handle privacy different here
-      public :synchronize
-    end
-
     private :post, :<<
 
     protected