Use lock-free-queue in throttling and improve API

Author: pitr-ch

lib/concurrent-edge.rb

@@ -8,6 +8,7 @@
 
 require 'concurrent/edge/atomic_markable_reference'
 require 'concurrent/edge/lock_free_linked_set'
+require 'concurrent/edge/lock_free_queue'
 
 require 'concurrent/edge/promises'
 require 'concurrent/edge/cancellation'

lib/concurrent/edge/lock_free_queue.rb

@@ -0,0 +1,116 @@
+module Concurrent
+
+  class LockFreeQueue < Synchronization::Object
+
+    class Node < Synchronization::Object
+      attr_atomic :successor
+
+      def initialize(item, successor)
+        super()
+        # published through queue, no need to be volatile or final
+        @Item          = item
+        self.successor = successor
+      end
+
+      def item
+        @Item
+      end
+    end
+
+    safe_initialization!
+
+    attr_atomic :head, :tail
+
+    def initialize
+      super()
+      dummy_node = Node.new(:dummy, nil)
+
+      self.head = dummy_node
+      self.tail = dummy_node
+    end
+
+    def push(item)
+      # allocate a new node with the item embedded
+      new_node = Node.new(item, nil)
+
+      # keep trying until the operation succeeds
+      while true
+        current_tail_node      = tail
+        current_tail_successor = current_tail_node.successor
+
+        # if our stored tail is still the current tail
+        if current_tail_node == tail
+          # if that tail was really the last node
+          if current_tail_successor.nil?
+            # if we can update the previous successor of tail to point to this new node
+            if current_tail_node.compare_and_set_successor(nil, new_node)
+              # then update tail to point to this node as well
+              compare_and_set_tail(current_tail_node, new_node)
+              # and return
+              return true
+              # else, start the loop over
+            end
+          else
+            # in this case, the tail ref we had wasn't the real tail
+            # so we try to set its successor as the real tail, then start the loop again
+            compare_and_set_tail(current_tail_node, current_tail_successor)
+          end
+        end
+      end
+    end
+
+    def pop
+      # retry until some value can be returned
+      while true
+        # the value in @head is just a dummy node that always sits in that position,
+        # the real 'head' is in its successor
+        current_dummy_node = head
+        current_tail_node  = tail
+
+        current_head_node = current_dummy_node.successor
+
+        # if our local head is still consistent with the head node, continue
+        # otherwise, start over
+        if current_dummy_node == head
+          # if either the queue is empty, or falling behind
+          if current_dummy_node == current_tail_node
+            # if there's nothing after the 'dummy' head node
+            if current_head_node.nil?
+              # just return nil
+              return nil
+            else
+              # here the head element succeeding head is not nil, but the head and tail are equal
+              # so tail is falling behind, update it, then start over
+              compare_and_set_tail(current_tail_node, current_head_node)
+            end
+
+            # the queue isn't empty
+            # if we can set the dummy head to the 'real' head, we're free to return the value in that real head, success
+          elsif compare_and_set_head(current_dummy_node, current_head_node)
+            # grab the item from the popped node
+            item = current_head_node.item
+
+            # return it, success!
+            return item
+          end
+        end
+      end
+    end
+
+    # approximate
+    def size
+      successor = head.successor
+      count     = 0
+
+      while true
+        break if successor.nil?
+
+        current_node = successor
+        successor    = current_node.successor
+        count        += 1
+      end
+
+      count
+    end
+  end
+end

lib/concurrent/edge/promises.rb

@@ -1967,50 +1967,47 @@ class Promises::Throttle < Synchronization::Object
     def initialize(max)
       super()
       self.can_run = max
-      # TODO (pitr-ch 10-Jun-2016): lock-free queue is needed
-      @Queue       = Queue.new
+      @Queue       = LockFreeQueue.new
     end
 
-    def limit(future = nil, &block)
-      if future
-        # future.chain { block.call(new_trigger & future).on_resolution! { done } }.flat
-        block.call(new_trigger & future).on_resolution! { done }
+    def throttle(future = nil, &throttled_future)
+      if block_given?
+        trigger = future ? (new_trigger & future) : new_trigger
+        throttled_future.call(trigger).on_resolution! { done }
       else
-        if block_given?
-          block.call(new_trigger).on_resolution! { done }
-        else
-          new_trigger
-        end
+        new_trigger
       end
     end
 
-    # TODO (pitr-ch 10-Oct-2016): maybe just then?
-    def then_limit(&block)
-      limit { |trigger| trigger.then &block }
+    def then_throttle(&task)
+      throttle { |trigger| trigger.then &task }
     end
 
+    private
+
     def done
       while true
         current_can_run = can_run
         if compare_and_set_can_run current_can_run, current_can_run + 1
-          @Queue.pop.resolve if current_can_run < 0
+          if current_can_run <= 0
+            Thread.pass until (trigger = @Queue.pop)
+            trigger.resolve
+          end
           return self
         end
       end
     end
 
-    private
-
     def new_trigger
       while true
         current_can_run = can_run
         if compare_and_set_can_run current_can_run, current_can_run - 1
           if current_can_run > 0
             return Promises.resolved_event
           else
-            e = Promises.resolvable_event
-            @Queue.push e
-            return e
+            event = Promises.resolvable_event
+            @Queue.push event
+            return event
           end
         end
       end
@@ -2020,7 +2017,7 @@ def new_trigger
   class Promises::AbstractEventFuture < Synchronization::Object
 
     def throttle(throttle, &throttled_future)
-      throttle.limit(self, &throttled_future)
+      throttle.throttle(self, &throttled_future)
     end
 
     def then_throttle(throttle, &block)

spec/concurrent/edge/promises_spec.rb

@@ -507,21 +507,21 @@ def behaves_as_delay(delay, value)
 
       expect(Concurrent::Promises.zip(
           *12.times.map do |i|
-            max_tree.limit { |trigger| trigger.then &testing }
-          end).value!.all? { |v| v < 3 }).to be_truthy
+            max_tree.throttle { |trigger| trigger.then &testing }
+          end).value!.all? { |v| v <= 3 }).to be_truthy
 
       expect(Concurrent::Promises.zip(
           *12.times.map do |i|
             Concurrent::Promises.
                 fulfilled_future(i).
                 throttle(max_tree) { |trigger| trigger.then &testing }
-          end).value!.all? { |v| v < 3 }).to be_truthy
+          end).value!.all? { |v| v <= 3 }).to be_truthy
     end
 
     specify do
       max_five = Concurrent::Promises::Throttle.new 5
       jobs     = 20.times.map do |i|
-        max_five.limit do |trigger|
+        max_five.throttle do |trigger|
           # trigger is an event, has same chain-able capabilities as current promise
           trigger.then do
             # at any given time there max 5 simultaneous executions of this block
@@ -537,7 +537,7 @@ def behaves_as_delay(delay, value)
     specify do
       max_five = Concurrent::Promises::Throttle.new 5
       jobs     = 20.times.map do |i|
-        max_five.then_limit do
+        max_five.then_throttle do
           # at any given time there max 5 simultaneous executions of this block
           the_work = i * 2
         end # returns promise