Ensure proper synchronization of Actor instances

Author: pitr-ch

lib/concurrent/actor.rb

@@ -2,6 +2,7 @@
 require 'concurrent/executor/serialized_execution'
 require 'concurrent/ivar'
 require 'concurrent/logging'
+require 'concurrent/atomic/synchronization'
 
 module Concurrent
   # TODO https://github.com/celluloid/celluloid/wiki/Supervision-Groups

lib/concurrent/actor/core.rb

@@ -10,6 +10,7 @@ module Actor
     class Core
       include TypeCheck
       include Concurrent::Logging
+      include Synchronization
 
       # @!attribute [r] reference
       #   @return [Reference] reference to this actor which can be safely passed around
@@ -42,53 +43,51 @@ class Core
       #   can be used to hook actor instance to any logging system
       # @param [Proc] block for class instantiation
       def initialize(opts = {}, &block)
-        # @mutex = Mutex.new
-        # @mutex.lock
-        # FIXME make initialization safe!
-
-        @mailbox              = Array.new
-        @serialized_execution = SerializedExecution.new
-        @executor             = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
-        @children             = Set.new
-        @context_class        = Child! opts.fetch(:class), AbstractContext
-        allocate_context
-        @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
+        synchronize do
+          @mailbox              = Array.new
+          @serialized_execution = SerializedExecution.new
+          @executor             = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
+          @children             = Set.new
+          @context_class        = Child! opts.fetch(:class), AbstractContext
+          allocate_context
+          @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]
+          @path   = @parent_core ? File.join(@parent_core.path, @name) : @name
+          @logger = opts[:logger]
 
-        @parent_core.add_child reference if @parent_core
+          @parent_core.add_child reference if @parent_core
 
-        initialize_behaviours opts
+          initialize_behaviours opts
 
-        @args       = opts.fetch(:args, [])
-        @block      = block
-        initialized = Type! opts[:initialized], IVar, NilClass
+          @args       = opts.fetch(:args, [])
+          @block      = block
+          initialized = Type! opts[:initialized], IVar, NilClass
 
-        messages = []
-        messages << :link if opts[:link]
-        messages << :supervise if opts[:supervise]
+          messages = []
+          messages << :link if opts[:link]
+          messages << :supervise if opts[:supervise]
 
-        schedule_execution do
-          begin
-            build_context
+          schedule_execution do
+            begin
+              build_context
 
-            messages.each do |message|
-              handle_envelope Envelope.new(message, nil, parent, reference)
-            end
+              messages.each do |message|
+                handle_envelope Envelope.new(message, nil, parent, reference)
+              end
 
-            initialized.set true if initialized
-          rescue => ex
-            log ERROR, ex
-            @first_behaviour.terminate!
-            initialized.fail ex if initialized
+              initialized.set true if initialized
+            rescue => ex
+              log ERROR, ex
+              @first_behaviour.terminate!
+              initialized.fail ex if initialized
+            end
           end
         end
       end
@@ -148,13 +147,15 @@ def log(level, message = nil, &block)
       # sets Actress.current
       def schedule_execution
         @serialized_execution.post(@executor) do
-          begin
-            Thread.current[:__current_actor__] = reference
-            yield
-          rescue => e
-            log FATAL, e
-          ensure
-            Thread.current[:__current_actor__] = nil
+          synchronize do
+            begin
+              Thread.current[:__current_actor__] = reference
+              yield
+            rescue => e
+              log FATAL, e
+            ensure
+              Thread.current[:__current_actor__] = nil
+            end
           end
         end
 

lib/concurrent/atomic/synchronization.rb

@@ -0,0 +1,51 @@
+module Concurrent
+
+  # Safe synchronization under JRuby, prevents reading uninitialized @mutex variable.
+  # @note synchronized needs to be called in #initialize for this module to work properly
+  # @example usage
+  #     class AClass
+  #       include Synchronized
+  #
+  #       def initialize
+  #         synchronize do
+  #           # body of the constructor ...
+  #         end
+  #       end
+  #
+  #       def a_method
+  #         synchronize do
+  #           # body of a_method ...
+  #         end
+  #       end
+  #     end
+  module Synchronization
+
+    engine = defined?(RUBY_ENGINE) && RUBY_ENGINE
+
+    case engine
+    when 'jruby'
+      require 'jruby'
+
+      def synchronize
+        JRuby.reference0(self).synchronized { yield }
+      end
+
+    when 'rbx'
+
+      def synchronize
+        Rubinius.lock(self)
+        yield
+      ensure
+        Rubinius.unlock(self)
+      end
+
+    else
+
+      def synchronize
+        @mutex ||= Mutex.new
+        @mutex.synchronize { yield }
+      end
+
+    end
+  end
+end

lib/concurrent/atomics.rb

@@ -8,3 +8,4 @@
 require 'concurrent/atomic/count_down_latch'
 require 'concurrent/atomic/event'
 require 'concurrent/atomic/thread_local_var'
+require 'concurrent/atomic/synchronization'