Add volatile cas fields, few classes converted

fix usages of #ensure_ivar_visibility!

Author: pitr-ch

lib/concurrent/atom.rb

@@ -57,9 +57,12 @@ module Concurrent
   #
   # @see http://clojure.org/atoms Clojure Atoms
   # @see http://clojure.org/state Values and Change - Clojure's approach to Identity and State
-  class Atom < Synchronization::LockableObject
+  class Atom < Synchronization::Object
     include Concern::Observable
 
+    private *attr_volatile_with_cas(:value)
+    public :value
+
     # Create a new atom with the given initial value.
     #
     # @param [Object] value The initial value
@@ -73,20 +76,16 @@ class Atom < Synchronization::LockableObject
     #
     # @raise [ArgumentError] if the validator is not a `Proc` (when given)
     def initialize(value, opts = {})
-      super()
-      synchronize do
-        @validator = opts.fetch(:validator, ->(v){ true })
-        @value = Concurrent::AtomicReference.new(value)
-        self.observers = Collection::CopyOnNotifyObserverSet.new
-      end
+      @Validator     = opts.fetch(:validator, -> (v) { true })
+      self.observers = Collection::CopyOnNotifyObserverSet.new
+      super(value) # ensures visibility
     end
 
-    # The current value of the atom.
+    # @!method value
+    #   The current value of the atom.
     #
-    # @return [Object] The current value.
-    def value
-      @value.value
-    end
+    #   @return [Object] The current value.
+
     alias_method :deref, :value
 
     # Atomically swaps the value of atom using the given block. The current
@@ -122,7 +121,7 @@ def swap(*args)
       raise ArgumentError.new('no block given') unless block_given?
 
       loop do
-        old_value = @value.value
+        old_value = value
         begin
           new_value = yield(old_value, *args)
           break old_value unless valid?(new_value)
@@ -143,7 +142,7 @@ def swap(*args)
     #
     # @return [Boolean] True if the value is changed else false.
     def compare_and_set(old_value, new_value)
-      if valid?(new_value) && @value.compare_and_set(old_value, new_value)
+      if valid?(new_value) && compare_and_set_value(old_value, new_value)
         observers.notify_observers(Time.now, old_value, new_value)
         true
       else
@@ -160,9 +159,9 @@ def compare_and_set(old_value, new_value)
     # @return [Object] The final value of the atom after all operations and
     #   validations are complete.
     def reset(new_value)
-      old_value = @value.value
+      old_value = value
       if valid?(new_value)
-        @value.set(new_value)
+        self.value = new_value
         observers.notify_observers(Time.now, old_value, new_value)
         new_value
       else
@@ -178,7 +177,7 @@ def reset(new_value)
     # @return [Boolean] false if the validator function returns false or raises
     #   an exception else true
     def valid?(new_value)
-      @validator.call(new_value)
+      @Validator.call(new_value)
     rescue
       false
     end

lib/concurrent/atomic/read_write_lock.rb

@@ -41,10 +41,10 @@ class ReadWriteLock < Synchronization::Object
     # @!visibility private
     MAX_WRITERS    = RUNNING_WRITER - MAX_READERS - 1
 
-    # Implementation notes: 
+    # 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
-    # Internal state is represented by a single integer ("counter"), and updated 
+    # Internal state is represented by a single integer ("counter"), and updated
     #  using atomic compare-and-swap operations
     # When the counter is 0, the lock is free
     # Each reader increments the counter by 1 when acquiring a read lock
@@ -57,8 +57,7 @@ def initialize
       @Counter   = AtomicFixnum.new(0) # single integer which represents lock state
       @ReadLock  = Synchronization::Lock.new
       @WriteLock = Synchronization::Lock.new
-      ensure_ivar_visibility!
-      super()
+      super() # ensures visibility
     end
 
     # Execute a block operation within a read lock.

lib/concurrent/atomic/reentrant_read_write_lock.rb

@@ -105,7 +105,7 @@ def initialize
       @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
-      ensure_ivar_visibility!
+      super() # ensures visibility
     end
 
     # Execute a block operation within a read lock.

lib/concurrent/edge/atomic_markable_reference.rb

@@ -11,11 +11,11 @@ module Edge
     #   @api Edge
     class AtomicMarkableReference < ::Concurrent::Synchronization::Object
 
+      private *attr_volatile_with_cas(:reference)
+
       # @!macro [attach] atomic_markable_reference_method_initialize
       def initialize(value = nil, mark = false)
-        super()
-        @Reference = AtomicReference.new ImmutableArray[value, mark]
-        ensure_ivar_visibility!
+        super(ImmutableArray[value, mark]) # ensures visibility
       end
 
       # @!macro [attach] atomic_markable_reference_method_compare_and_set
@@ -36,7 +36,7 @@ def initialize(value = nil, mark = false)
       def compare_and_set(expected_val, new_val, expected_mark, new_mark)
         # Memoize a valid reference to the current AtomicReference for
         # later comparison.
-        current = @Reference.get
+        current = reference
         curr_val, curr_mark = current
 
         # Ensure that that the expected marks match.
@@ -56,7 +56,7 @@ def compare_and_set(expected_val, new_val, expected_mark, new_mark)
 
         prospect = ImmutableArray[new_val, new_mark]
 
-        @Reference.compare_and_set current, prospect
+        compare_and_set_reference current, prospect
       end
       alias_method :compare_and_swap, :compare_and_set
 
@@ -66,7 +66,7 @@ def compare_and_set(expected_val, new_val, expected_mark, new_mark)
       #
       #   @return [ImmutableArray] the current reference and marked values
       def get
-        @Reference.get
+        reference
       end
 
       # @!macro [attach] atomic_markable_reference_method_value
@@ -75,7 +75,7 @@ def get
       #
       #   @return [Object] the current value of the reference
       def value
-        @Reference.get[0]
+        reference[0]
       end
 
       # @!macro [attach] atomic_markable_reference_method_mark
@@ -84,7 +84,7 @@ def value
       #
       #   @return [Boolean] the current marked value
       def mark
-        @Reference.get[1]
+        reference[1]
       end
       alias_method :marked?, :mark
 
@@ -98,7 +98,7 @@ def mark
       #
       #   @return [ImmutableArray] both the new value and the new mark
       def set(new_val, new_mark)
-        @Reference.set ImmutableArray[new_val, new_mark]
+        self.reference = ImmutableArray[new_val, new_mark]
       end
 
       # @!macro [attach] atomic_markable_reference_method_update
@@ -115,7 +115,7 @@ def set(new_val, new_mark)
       # @return [ImmutableArray] the new value and new mark
       def update
         loop do
-          old_val, old_mark = @Reference.get
+          old_val, old_mark = reference
           new_val, new_mark = yield old_val, old_mark
 
           if compare_and_set old_val, new_val, old_mark, new_mark
@@ -139,7 +139,7 @@ def update
       #
       # @raise [Concurrent::ConcurrentUpdateError] if the update fails
       def try_update!
-        old_val, old_mark = @Reference.get
+        old_val, old_mark = reference
         new_val, new_mark = yield old_val, old_mark
 
         unless compare_and_set old_val, new_val, old_mark, new_mark
@@ -165,7 +165,7 @@ def try_update!
       # @return [ImmutableArray] the new value and marked state, or nil if
       # the update failed
       def try_update
-        old_val, old_mark = @Reference.get
+        old_val, old_mark = reference
         new_val, new_mark = yield old_val, old_mark
 
         return unless compare_and_set old_val, new_val, old_mark, new_mark

lib/concurrent/edge/future.rb

@@ -173,8 +173,7 @@ def initialize(promise, default_executor)
         @Callbacks       = LockFreeStack.new
         @Waiters         = LockFreeStack.new # TODO replace with AtomicFixnum, avoid aba problem
         @State           = AtomicReference.new PENDING
-        super()
-        ensure_ivar_visibility!
+        super() # ensures visibility
       end
 
       # @return [:pending, :completed]
@@ -879,9 +878,8 @@ def hide_completable
     # @!visibility private
     class AbstractPromise < Synchronization::Object
       def initialize(future)
-        super()
         @Future = future
-        ensure_ivar_visibility!
+        super()
       end
 
       def future
@@ -1377,9 +1375,8 @@ def initialize(default_executor, intended_time)
     class Channel < Synchronization::Object
       # TODO make lock free
       def initialize
-        super
         @ProbeSet = Concurrent::Channel::WaitableList.new
-        ensure_ivar_visibility!
+        super()
       end
 
       def probe_set_size

lib/concurrent/edge/lock_free_linked_set/node.rb

@@ -9,13 +9,11 @@ class Node < Synchronization::Object
         attr_reader :Data, :Successor_reference, :Key
 
         def initialize(data = nil, successor = nil)
-          super()
-
           @Successor_reference = AtomicMarkableReference.new(successor || Tail.new)
           @Data = data
           @Key = key_for data
 
-          ensure_ivar_visibility!
+          super() # ensures visibility
         end
 
         # Check to see if the node is the last in the list.

lib/concurrent/edge/lock_free_stack.rb

@@ -21,51 +21,51 @@ def next_node
 
       EMPTY = Empty[nil, nil]
 
+      private *attr_volatile_with_cas(:head)
+
       def initialize
-        super
-        @Head = AtomicReference.new EMPTY
-        ensure_ivar_visibility!
+        super(EMPTY)
       end
 
       def empty?
-        @Head.get.equal? EMPTY
+        head.equal? EMPTY
       end
 
       def compare_and_push(head, value)
-        @Head.compare_and_set head, Node[value, head]
+        compare_and_set_head head, Node[value, head]
       end
 
       def push(value)
         while true
-          head = @Head.get
-          return self if @Head.compare_and_set head, Node[value, head]
+          current_head = head
+          return self if compare_and_set_head current_head, Node[value, current_head]
         end
       end
 
       def peek
-        @Head.get
+        head
       end
 
       def compare_and_pop(head)
-        @Head.compare_and_set head, head.next_node
+        compare_and_set_head head, head.next_node
       end
 
       def pop
         while true
-          head = @Head.get
-          return head.value if @Head.compare_and_set head, head.next_node
+          current_head = head
+          return current_head.value if compare_and_set_head current_head, current_head.next_node
         end
       end
 
       def compare_and_clear(head)
-        @Head.compare_and_set head, EMPTY
+        compare_and_set_head head, EMPTY
       end
 
       def clear
         while true
-          head = @Head.get
-          return false if head == EMPTY
-          return true if @Head.compare_and_set head, EMPTY
+          current_head = head
+          return false if current_head == EMPTY
+          return true if compare_and_set_head current_head, EMPTY
         end
       end
 

lib/concurrent/executor/safe_task_executor.rb

@@ -9,23 +9,22 @@ module Concurrent
   class SafeTaskExecutor < Synchronization::LockableObject
 
     def initialize(task, opts = {})
-      super()
-      @task = task
+      @task            = task
       @exception_class = opts.fetch(:rescue_exception, false) ? Exception : StandardError
-      ensure_ivar_visibility!
+      super() # ensures visibility
     end
 
     # @return [Array]
     def execute(*args)
       synchronize do
         success = false
-        value = reason = nil
+        value   = reason = nil
 
         begin
-          value = @task.call(*args)
+          value   = @task.call(*args)
           success = true
         rescue @exception_class => ex
-          reason = ex
+          reason  = ex
           success = false
         end
 

lib/concurrent/synchronization/condition.rb

@@ -9,7 +9,6 @@ class Condition < LockableObject
 
       def initialize(lock)
         @Lock = lock
-        ensure_ivar_visibility!
         super()
       end
 

lib/concurrent/synchronization/lockable_object.rb

@@ -19,7 +19,10 @@ module Synchronization
     private_constant :LockableObjectImplementation
 
     class LockableObject < LockableObjectImplementation
-      def self.allow_only_direct_descendants! # FIXME interne dedime docela dost :/
+
+      # TODO make private for c-r
+
+      def self.allow_only_direct_descendants! # FIXME we inherit too much ourselves :/
         this = self
         singleton_class.send :define_method, :inherited do |child|
           # super child