Merge pull request #335 from ianks/lock-free-linked-set

LockFreeLinkedSet: add initial implementation

Author: jdantonio

lib/concurrent-edge.rb

@@ -9,3 +9,4 @@
 require 'concurrent/edge/future'
 require 'concurrent/edge/lock_free_stack'
 require 'concurrent/edge/atomic_markable_reference'
+require 'concurrent/edge/lock_free_linked_set'

lib/concurrent/edge/lock_free_linked_set.rb

@@ -0,0 +1,146 @@
+require 'concurrent/edge/lock_free_linked_set/node'
+require 'concurrent/edge/lock_free_linked_set/window'
+
+module Concurrent
+  module Edge
+    # This class implements a lock-free linked set. The general idea of this
+    # implementation is this: each node has a successor which is an Atomic
+    # Markable Reference. This is used to ensure that all modifications to the
+    # list are atomic, preserving the structure of the linked list under _any_
+    # circumstance in a multithreaded application.
+    #
+    # One interesting aspect of this algorithm occurs with removing a node.
+    # Instead of physically removing a node when remove is called, a node is
+    # logically removed, by 'marking it.' By doing this, we prevent calls to
+    # `remove` from traversing the list twice to perform a physical removal.
+    # Instead, we have have calls to `add` and `remove` clean up all marked
+    # nodes they encounter while traversing the list.
+    #
+    # This algorithm is a variation of the Nonblocking Linked Set found in
+    # 'The Art of Multiprocessor Programming' by Herlihy and Shavit.
+    class LockFreeLinkedSet
+      include Enumerable
+
+      # @!macro [attach] lock_free_linked_list_method_initialize
+      #
+      # @param [Fixnum] initial_size the size of the linked_list to initialize
+      def initialize(initial_size = 0, val = nil)
+        @head = Head.new
+
+        initial_size.times do
+          val = block_given? ? yield : val
+          add val
+        end
+      end
+
+      # @!macro [attach] lock_free_linked_list_method_add
+      #
+      #   Atomically adds the item to the set if it does not yet exist. Note:
+      #   internally the set uses `Object#hash` to compare equality of items,
+      #   meaning that Strings and other objects will be considered equal
+      #   despite being different objects.
+      #
+      #   @param [Object] item the item you wish to insert
+      #
+      #   @return [Boolean] `true` if successful. A `false` return indicates
+      #   that the item was already in the set.
+      def add(item)
+        loop do
+          window = Window.find @head, item
+
+          pred, curr = window.pred, window.curr
+
+          # Item already in set
+          return false if curr == item
+
+          node = Node.new item, curr
+
+          if pred.Successor_reference.compare_and_set curr, node, false, false
+            return true
+          end
+        end
+      end
+
+      # @!macro [attach] lock_free_linked_list_method_<<
+      #
+      #   Atomically adds the item to the set if it does not yet exist.
+      #
+      #   @param [Object] item the item you wish to insert
+      #
+      #   @return [Oject] the set on which the :<< method was invoked
+      def <<(item)
+        add item
+        self
+      end
+
+      # @!macro [attach] lock_free_linked_list_method_contains
+      #
+      #   Atomically checks to see if the set contains an item. This method
+      #   compares equality based on the `Object#hash` method, meaning that the
+      #   hashed contents of an object is what determines equality instead of
+      #   `Object#object_id`
+      #
+      #   @param [Object] item the item you to check for presence in the set
+      #
+      #   @return [Boolean] whether or not the item is in the set
+      def contains?(item)
+        curr = @head
+
+        while curr < item
+          curr = curr.next_node
+          marked = curr.Successor_reference.marked?
+        end
+
+        curr == item && !marked
+      end
+
+      # @!macro [attach] lock_free_linked_list_method_remove
+      #
+      #   Atomically attempts to remove an item, comparing using `Object#hash`.
+      #
+      #   @param [Object] item the item you to remove from the set
+      #
+      #   @return [Boolean] whether or not the item was removed from the set
+      def remove(item)
+        loop do
+          window = Window.find @head, item
+          pred, curr = window.pred, window.curr
+
+          return false if curr != item
+
+          succ = curr.next_node
+          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
+
+          return true
+        end
+      end
+
+      # @!macro [attach] lock_free_linked_list_method_each
+      #
+      #   An iterator to loop through the set.
+      #
+      #   @param [Object] item the item you to remove from the set
+      #   @yeild [Object] each item in the set
+      #
+      #   @return [Object] self: the linked set on which each was called
+      def each
+        return to_enum unless block_given?
+
+        curr = @head
+
+        until curr.last?
+          curr = curr.next_node
+          marked = curr.Successor_reference.marked?
+
+          yield curr.Data unless marked
+        end
+
+        self
+      end
+    end
+  end
+end

lib/concurrent/edge/lock_free_linked_set/node.rb

@@ -0,0 +1,72 @@
+require 'concurrent/edge/atomic_markable_reference'
+
+module Concurrent
+  module Edge
+    class LockFreeLinkedSet
+      class Node < Synchronization::Object
+        include Comparable
+
+        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!
+        end
+
+        # Check to see if the node is the last in the list.
+        def last?
+          @Successor_reference.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
+        end
+
+        # This method provides a unqiue key for the data which will be used for
+        # ordering. This is configurable, and changes depending on how you wish
+        # the nodes to be ordered.
+        def key_for(data)
+          data.hash
+        end
+
+        # We use `Object#hash` as a way to enforce ordering on the nodes. This
+        # can be configurable in the future; for example, you could enforce a
+        # split-ordering on the nodes in the set.
+        def <=>(other)
+          @Key <=> other.hash
+        end
+      end
+
+      # Internal sentinel node for the Tail. It is always greater than all
+      # other nodes, and  it is self-referential; meaning its successor is
+      # a self-loop.
+      class Tail < Node
+        def initialize(_data = nil, _succ = nil)
+          @Successor_reference = AtomicMarkableReference.new self
+        end
+
+        # Always greater than other nodes. This means that traversal will end
+        # at the tail node since we are comparing node size in the traversal.
+        def <=>(_other)
+          1
+        end
+      end
+
+
+      # Internal sentinel node for the Head of the set. Head is always smaller
+      # than any other node.
+      class Head < Node
+        def <=>(_other)
+          -1
+        end
+      end
+    end
+  end
+end

lib/concurrent/edge/lock_free_linked_set/window.rb

@@ -0,0 +1,49 @@
+module Concurrent
+  module Edge
+    class LockFreeLinkedSet
+      class Window
+        attr_accessor :pred, :curr
+
+        def initialize(pred, curr)
+          @pred, @curr = pred, curr
+        end
+
+        # This method is used to find a 'window' for which `add` and `remove`
+        # methods can use to know where to add and remove from the list. However,
+        # it has another responsibilility, which is to physically unlink any
+        # nodes marked for removal in the set. This prevents adds/removes from
+        # having to retraverse the list to physically unlink nodes.
+        def self.find(head, item)
+          loop do
+            break_inner_loops = false
+            pred = head
+            curr = pred.next_node
+
+            loop do
+              succ, marked = curr.Successor_reference.get
+
+              # Remove sequence of marked nodes
+              while marked
+                removed = pred.Successor_reference.compare_and_set curr, succ, false, false
+
+                # If could not remove node, try again
+                break_inner_loops = true && break unless removed
+
+                curr = succ
+                succ, marked = curr.Successor_reference.get
+              end
+
+              break if break_inner_loops
+
+              # We have found a window
+              return new pred, curr if curr >= item
+
+              pred = curr
+              curr = succ
+            end
+          end
+        end
+      end
+    end
+  end
+end