Add ensure_ivar_visibility! and volatile attributes

Author: pitr-ch

lib/concurrent/synchronization.rb

@@ -1,6 +1,7 @@
 require 'concurrent/utility/engine'
 require 'concurrent/synchronization/abstract_object'
 require 'concurrent/native_extensions' # JavaObject
+require 'concurrent/synchronization/java_object' # JavaObject
 require 'concurrent/synchronization/mutex_object'
 require 'concurrent/synchronization/monitor_object'
 require 'concurrent/synchronization/rbx_object'
@@ -21,6 +22,7 @@ module Synchronization
                      end
     private_constant :Implementation
 
+    # @see AbstractObject
     class Object < Implementation
     end
   end

lib/concurrent/synchronization/abstract_object.rb

@@ -1,9 +1,7 @@
 module Concurrent
-  # TODO rename to Synchronization
-  # TODO add newCondition
   module Synchronization
     # Safe synchronization under any Ruby implementation.
-    # It provides methods like {#synchronize}, {#wait}, {#signal} and {#broadcast}.
+    # It provides methods like {#synchronize}, {#ns_wait}, {#ns_signal} and {#ns_broadcast}.
     # Provides a single layer which can improve its implementation over time without changes needed to
     # the classes using it. Use {Synchronization::Object} not this abstract class.
     #
@@ -118,6 +116,40 @@ def ns_signal
       def ns_broadcast
         raise NotImplementedError
       end
+
+      # Allows to construct immutable objects where all fields are visible after initialization, not requiring
+      # further synchronization on access.
+      # @example
+      #   class AClass
+      #     attr_reader :val
+      #     def initialize(val)
+      #       @val = val # final value, after assignment it's not changed (just convention, not enforced)
+      #       ensure_ivar_visibility!
+      #       # now it can be shared as Java's final field
+      #     end
+      #   end
+      def ensure_ivar_visibility!
+        raise NotImplementedError
+      end
+
+      # creates methods for reading and writing to a instance variable with volatile (Java semantic) instance variable
+      # return [Array<Symbol>] names of defined method names
+      def self.attr_volatile(*names)
+        names.each do |name|
+          ivar = :"@volatile_#{name}"
+          class_eval <<-RUBY, __FILE__, __LINE__ + 1
+            def #{name}
+              #{ivar}
+            end
+
+            def #{name}=(value)
+              #{ivar} = value
+            end
+          RUBY
+        end
+        names.map { |n| [n, :"#{n}="] }.flatten
+      end
+
     end
   end
 end

lib/concurrent/synchronization/java_object.rb

@@ -0,0 +1,14 @@
+module Concurrent
+  module Synchronization
+
+    if Concurrent.on_jruby?
+      require 'jruby'
+
+      class JavaObject < AbstractObject
+        def ensure_ivar_visibility!
+          # relying on undocumented behavior of JRuby, ivar access is volatile
+        end
+      end
+    end
+  end
+end

lib/concurrent/synchronization/java_pure_object.rb

@@ -33,6 +33,10 @@ def ns_signal
           JRuby.reference0(self).notify
           self
         end
+
+        def ensure_ivar_visibility!
+          # relying on undocumented behavior of JRuby, ivar access is volatile
+        end
       end
     end
   end

lib/concurrent/synchronization/mutex_object.rb

@@ -31,6 +31,11 @@ def ns_wait(timeout = nil)
         @__condition__do_not_use_directly.wait @__lock__do_not_use_directly, timeout
         self
       end
+
+      def ensure_ivar_visibility!
+        # relying on undocumented behavior of CRuby, GVL acquire has lock which ensures visibility of ivars
+        # https://github.com/ruby/ruby/blob/ruby_2_2/thread_pthread.c#L204-L211
+      end
     end
   end
 end

lib/concurrent/synchronization/rbx_object.rb

@@ -44,6 +44,29 @@ def ns_broadcast
           @waiters.shift << true until @waiters.empty?
           self
         end
+
+        def ensure_ivar_visibility!
+          # Rubinius instance variables are not volatile so we need to insert barrier
+          Rubinius.memory_barrier
+        end
+
+        def self.attr_volatile *names
+          names.each do |name|
+            ivar = :"@volatile_#{name}"
+            class_eval <<-RUBY, __FILE__, __LINE__ + 1
+            def #{name}
+              Rubinius.memory_barrier
+              #{ivar}
+            end
+
+            def #{name}=(value)
+              #{ivar} = value
+              Rubinius.memory_barrier
+            end
+            RUBY
+          end
+          names.map { |n| [n, :"#{n}="] }.flatten
+        end
       end
     end
   end

spec/concurrent/synchronized_object_spec.rb

@@ -0,0 +1,109 @@
+module Concurrent
+
+  describe Synchronization::Object do
+
+    class AClass < Synchronization::Object
+      attr_volatile :volatile
+      attr_accessor :not_volatile
+
+      def initialize(value = nil)
+        super()
+        @Final = value
+        ensure_ivar_visibility!
+      end
+
+      def final
+        @Final
+      end
+
+      def count
+        synchronize { @count += 1 }
+      end
+
+      def wait(timeout = nil)
+        synchronize { ns_wait(timeout) }
+      end
+
+      private
+
+      def ns_initialize
+        @count = 0
+      end
+    end
+
+    subject { AClass.new }
+
+    describe '#wait' do
+
+      it 'waiting thread is sleeping' do
+        t = Thread.new do
+          Thread.abort_on_exception = true
+          subject.wait
+        end
+        sleep 0.1
+        expect(t.status).to eq 'sleep'
+      end
+
+      it 'sleeping thread can be killed' do
+        t = Thread.new do
+          Thread.abort_on_exception = true
+          subject.wait rescue nil
+        end
+        sleep 0.1
+        t.kill
+        sleep 0.1
+        expect(t.status).to eq false
+        expect(t.alive?).to eq false
+      end
+    end
+
+    describe '#synchronize' do
+      it 'allows only one thread to execute count' do
+        threads = 10.times.map { Thread.new(subject) { 100.times { subject.count } } }
+        threads.each(&:join)
+        expect(subject.count).to eq 1001
+      end
+    end
+
+    describe 'signaling' do
+      pending 'for now pending, tested pretty well by Event'
+    end
+
+    specify 'final field always visible' do
+      store = AClass.new 'asd'
+      t1    = Thread.new { 1000000000.times { |i| store = AClass.new i.to_s } }
+      t2    = Thread.new { 10.times { expect(store.final).not_to be_nil; Thread.pass } }
+      t2.join
+      t1.kill
+    end
+
+    describe 'attr volatile' do
+      specify 'older writes are always visible' do
+        store              = AClass.new
+        store.not_volatile = 0
+        store.volatile     = 0
+
+        t1 = Thread.new do
+          Thread.abort_on_exception = true
+          1000000000.times do |i|
+            store.not_volatile = i
+            store.volatile     = i
+          end
+        end
+
+        t2 = Thread.new do
+          10.times do
+            volatile     = store.volatile
+            not_volatile = store.not_volatile
+            expect(not_volatile).to be >= volatile
+            Thread.pass
+          end
+        end
+
+        t2.join
+        t1.kill
+      end
+    end
+
+  end
+end