Introduce safe_initialization! class marker instead of ensure_ivar_visibility!

* ensures full memory barier is inserted only when needed
* ensures that it's inserted only once
* testable correctness with @FinalIvar convention

Author: pitr-ch

ext/com/concurrent_ruby/ext/SynchronizationLibrary.java

@@ -87,14 +87,12 @@ public IRubyObject initialize(ThreadContext context) {
 
         @JRubyMethod(name = "full_memory_barrier", visibility = Visibility.PRIVATE)
         public IRubyObject fullMemoryBarrier(ThreadContext context) {
-            if (UnsafeHolder.U == null) {
-                // We are screwed
-                throw new UnsupportedOperationException();
-            } else if (UnsafeHolder.SUPPORTS_FENCES)
+            if (UnsafeHolder.U == null || !UnsafeHolder.SUPPORTS_FENCES) {
+                throw new UnsupportedOperationException(
+                        "concurrent-ruby requires java with sun.mics.Unsafe fences support, such as Java 8. " +
+                                "Current version is: " + System.getProperty("java.version"));
+            } else {
                 UnsafeHolder.fullFence();
-            else {
-                // TODO (pitr 06-Sep-2015): enforce Java 8
-                throw new UnsupportedOperationException();
             }
             return context.nil;
         }

lib/concurrent/agent.rb

@@ -224,7 +224,7 @@ def initialize(initial, opts = {})
     #
     # @return [Object] the current value
     def value
-      @current.value
+      @current.value # TODO (pitr 12-Sep-2015): broken unsafe read?
     end
 
     alias_method :deref, :value

lib/concurrent/atom.rb

@@ -60,6 +60,7 @@ module Concurrent
   class Atom < Synchronization::Object
     include Concern::Observable
 
+    safe_initialization!
     private *attr_volatile_with_cas(:value)
     public :value
 
@@ -78,7 +79,7 @@ class Atom < Synchronization::Object
     def initialize(value, opts = {})
       @Validator     = opts.fetch(:validator, -> v { true })
       self.observers = Collection::CopyOnNotifyObserverSet.new
-      super(value) # ensures visibility
+      super(value)
     end
 
     # @!method value

lib/concurrent/atomic/read_write_lock.rb

@@ -41,6 +41,8 @@ class ReadWriteLock < Synchronization::Object
     # @!visibility private
     MAX_WRITERS    = RUNNING_WRITER - MAX_READERS - 1
 
+    safe_initialization!
+
     # Implementation notes:
     # A goal is to make the uncontended path for both readers/writers lock-free
     # Only if there is reader-writer or writer-writer contention, should locks be used
@@ -54,10 +56,10 @@ class ReadWriteLock < Synchronization::Object
 
     # Create a new `ReadWriteLock` in the unlocked state.
     def initialize
+      super()
       @Counter   = AtomicFixnum.new(0) # single integer which represents lock state
       @ReadLock  = Synchronization::Lock.new
       @WriteLock = Synchronization::Lock.new
-      super() # ensures visibility
     end
 
     # Execute a block operation within a read lock.

lib/concurrent/atomic/reentrant_read_write_lock.rb

@@ -99,13 +99,15 @@ class ReentrantReadWriteLock < Synchronization::Object
     # @!visibility private
     WRITE_LOCK_MASK = MAX_WRITERS
 
+    safe_initialization!
+
     # Create a new `ReentrantReadWriteLock` in the unlocked state.
     def initialize
+      super()
       @Counter    = AtomicFixnum.new(0)       # single integer which represents lock state
       @ReadQueue  = Synchronization::Lock.new # used to queue waiting readers
       @WriteQueue = Synchronization::Lock.new # used to queue waiting writers
       @HeldCount  = ThreadLocalVar.new(0)     # indicates # of R & W locks held by this thread
-      super() # ensures visibility
     end
 
     # Execute a block operation within a read lock.

lib/concurrent/channel/buffered_channel.rb

@@ -14,7 +14,7 @@ def initialize(size)
       end
 
       def probe_set_size
-        @probe_set.size
+        @probe_set.size # TODO (pitr 12-Sep-2015): unsafe?
       end
 
       def buffer_queue_size

lib/concurrent/delay.rb

@@ -74,7 +74,7 @@ def initialize(opts = {}, &block)
     #
     # @!macro delay_note_regarding_blocking
     def value(timeout = nil)
-      if @executor
+      if @executor # TODO (pitr 12-Sep-2015): broken unsafe read?
         super
       else
         # this function has been optimized for performance and

lib/concurrent/edge/future.rb

@@ -126,6 +126,7 @@ def post_on(executor, *args, &job)
 
     # Represents an event which will happen in future (will be completed). It has to always happen.
     class Event < Synchronization::LockableObject
+      safe_initialization!
       include Concern::Deprecation
 
       # @!visibility private
@@ -167,13 +168,15 @@ def to_sym
       COMPLETED = Completed.new
 
       def initialize(promise, default_executor)
+        super()
         @Promise         = promise
         @DefaultExecutor = default_executor
         @Touched         = AtomicBoolean.new(false)
         @Callbacks       = LockFreeStack.new
-        @Waiters         = LockFreeStack.new # TODO replace with AtomicFixnum, avoid aba problem
+        # TODO (pitr 12-Sep-2015): replace with AtomicFixnum, avoid aba problem
+        # TODO (pitr 12-Sep-2015): look at java.util.concurrent solution
+        @Waiters         = LockFreeStack.new
         @State           = AtomicReference.new PENDING
-        super() # ensures visibility
       end
 
       # @return [:pending, :completed]
@@ -877,9 +880,11 @@ def hide_completable
     # @abstract
     # @!visibility private
     class AbstractPromise < Synchronization::Object
+      safe_initialization!
+
       def initialize(future)
-        @Future = future
         super()
+        @Future = future
       end
 
       def future
@@ -1373,10 +1378,12 @@ def initialize(default_executor, intended_time)
 
     # @note proof of concept
     class Channel < Synchronization::Object
+      safe_initialization!
+
       # TODO make lock free
       def initialize
-        @ProbeSet = Concurrent::Channel::WaitableList.new
         super()
+        @ProbeSet = Concurrent::Channel::WaitableList.new
       end
 
       def probe_set_size

lib/concurrent/edge/lock_free_linked_set.rb

@@ -55,7 +55,7 @@ def add(item)
 
           node = Node.new item, curr
 
-          if pred.Successor_reference.compare_and_set curr, node, false, false
+          if pred.successor_reference.compare_and_set curr, node, false, false
             return true
           end
         end
@@ -88,7 +88,7 @@ def contains?(item)
 
         while curr < item
           curr = curr.next_node
-          marked = curr.Successor_reference.marked?
+          marked = curr.successor_reference.marked?
         end
 
         curr == item && !marked
@@ -109,11 +109,11 @@ def remove(item)
           return false if curr != item
 
           succ = curr.next_node
-          removed = curr.Successor_reference.compare_and_set succ, succ, false, true
+          removed = curr.successor_reference.compare_and_set succ, succ, false, true
 
           next_node unless removed
 
-          pred.Successor_reference.compare_and_set curr, succ, false, false
+          pred.successor_reference.compare_and_set curr, succ, false, false
 
           return true
         end
@@ -134,9 +134,9 @@ def each
 
         until curr.last?
           curr = curr.next_node
-          marked = curr.Successor_reference.marked?
+          marked = curr.successor_reference.marked?
 
-          yield curr.Data unless marked
+          yield curr.data unless marked
         end
 
         self

lib/concurrent/edge/lock_free_linked_set/node.rb

@@ -6,25 +6,36 @@ class LockFreeLinkedSet
       class Node < Synchronization::Object
         include Comparable
 
-        attr_reader :Data, :Successor_reference, :Key
+        safe_initialization!
 
         def initialize(data = nil, successor = nil)
-          @Successor_reference = AtomicMarkableReference.new(successor || Tail.new)
-          @Data = data
-          @Key = key_for data
+          super()
+          @SuccessorReference = AtomicMarkableReference.new(successor || Tail.new)
+          @Data                = data
+          @Key                 = key_for data
+        end
+
+        def data
+          @Data
+        end
+
+        def successor_reference
+          @SuccessorReference
+        end
 
-          super() # ensures visibility
+        def key
+          @Key
         end
 
         # Check to see if the node is the last in the list.
         def last?
-          @Successor_reference.value.is_a? Tail
+          @SuccessorReference.value.is_a? Tail
         end
 
         # Next node in the list. Note: this is not the AtomicMarkableReference
         # of the next node, this is the actual Node itself.
         def next_node
-          @Successor_reference.value
+          @SuccessorReference.value
         end
 
         # This method provides a unqiue key for the data which will be used for
@@ -47,7 +58,7 @@ def <=>(other)
       # a self-loop.
       class Tail < Node
         def initialize(_data = nil, _succ = nil)
-          @Successor_reference = AtomicMarkableReference.new self
+          @SuccessorReference = AtomicMarkableReference.new self
         end
 
         # Always greater than other nodes. This means that traversal will end