Edge - Future tweaks

- remove all synchronization from path when there is no thread waiting
- improve AllPromise iterations
- zipped futures apply as multiple arguments to block not as a single array

Author: pitr-ch

examples/benchmark_new_futures.rb

@@ -8,48 +8,51 @@
 warmup *= 10 if Concurrent.on_jruby?
 
 Benchmark.ips(time, warmup) do |x|
-  x.report('graph-old') do
-    head    = Concurrent::Promise.execute { 1 }
-    branch1 = head.then(&:succ)
-    branch2 = head.then(&:succ).then(&:succ)
-    branch3 = head.then(&:succ).then(&:succ).then(&:succ)
-    Concurrent::Promise.zip(branch1, branch2, branch3).then { |(a, b, c)| a + b + c }.value!
-  end
-  x.report('graph-new') do
-    head    = Concurrent.future { 1 }
-    branch1 = head.then(&:succ)
-    branch2 = head.then(&:succ).then(&:succ)
-    branch3 = head.then(&:succ).then(&:succ).then(&:succ)
-    Concurrent.zip(branch1, branch2, branch3).then { |(a, b, c)| a + b + c }.value!
-  end
+  x.report('status-old') { f = Concurrent::Promise.execute { nil }; 100.times { f.complete? } }
+  x.report('status-new') { f = Concurrent.future(:fast) { nil }; 100.times { f.completed? } }
   x.compare!
 end
 
 Benchmark.ips(time, warmup) do |x|
-  x.report('status-old') { f = Concurrent::Promise.execute { nil }; 10.times { f.complete? } }
-  x.report('status-new') { f = Concurrent.future { nil }; 10.times { f.completed? } }
+  of = Concurrent::Promise.execute { 1 }
+  nf = Concurrent.future(:fast) { 1 }
+  x.report('value-old') { of.value! }
+  x.report('value-new') { nf.value! }
   x.compare!
 end
 
 Benchmark.ips(time, warmup) do |x|
-  of = Concurrent::Promise.execute { 1 }
-  nf = Concurrent.future { 1 }
-  x.report('value-old') { of.value! }
-  x.report('value-new') { nf.value! }
+  x.report('graph-old') do
+    head = Concurrent::Promise.execute { 1 }
+    10.times do
+      branch1 = head.then(&:succ)
+      branch2 = head.then(&:succ).then(&:succ)
+      head    = Concurrent::Promise.zip(branch1, branch2).then { |a, b| a + b }
+    end
+    head.value!
+  end
+  x.report('graph-new') do
+    head = Concurrent.future(:fast) { 1 }
+    10.times do
+      branch1 = head.then(&:succ)
+      branch2 = head.then(&:succ).then(&:succ)
+      head    = (branch1 & branch2).then { |a, b| a + b }
+    end
+    head.value!
+  end
   x.compare!
 end
 
 Benchmark.ips(time, warmup) do |x|
   x.report('immediate-old') { Concurrent::Promise.execute { nil }.value! }
-  x.report('immediate-new') { Concurrent.future { nil }.value! }
+  x.report('immediate-new') { Concurrent.future(:fast) { nil }.value! }
   x.compare!
 end
 
 Benchmark.ips(time, warmup) do |x|
   of = Concurrent::Promise.execute { 1 }
-  nf = Concurrent.future { 1 }
-  x.report('then-old') { of.then(&:succ).then(&:succ).value! }
-  x.report('then-new') { nf.then(&:succ).then(&:succ).value! }
+  nf = Concurrent.future(:fast) { 1 }
+  x.report('then-old') { 100.times.reduce(nf) { |nf, _| nf.then(&:succ) }.value! }
+  x.report('then-new') { 100.times.reduce(nf) { |nf, _| nf.then(&:succ) }.value! }
   x.compare!
 end
-

lib/concurrent/edge/future.rb

@@ -83,6 +83,7 @@ def initialize(promise, default_executor = :io)
         @DefaultExecutor = default_executor
         @Touched         = AtomicBoolean.new(false)
         @Callbacks       = LockFreeStack.new
+        @Waiters         = LockFreeStack.new
         @State           = AtomicReference.new :pending
         super()
         ensure_ivar_visibility!
@@ -174,7 +175,8 @@ def inspect
       # @api private
       def complete(raise = true)
         if complete_state
-          synchronize { ns_broadcast }
+          # go to synchronized block only if there were waiting threads
+          synchronize { ns_broadcast } if @Waiters.clear
           call_callbacks
         else
           Concurrent::MultipleAssignmentError.new('multiple assignment') if raise
@@ -222,8 +224,19 @@ def touched
       private
 
       def wait_until_complete(timeout)
-        unless completed?
-          synchronize { ns_wait_until(timeout) { completed? } }
+        lock = Synchronization::Lock.new
+
+        while true
+          last_waiter = @Waiters.peek # waiters' state before completion
+          break if completed?
+
+          # synchronize so it cannot be signaled before it waits
+          synchronize do
+            # ok only if completing thread did not start signaling
+            next unless @Waiters.compare_and_push last_waiter, lock
+            ns_wait_until(timeout) { completed? }
+            break
+          end
         end
         self
       end
@@ -244,12 +257,11 @@ def pr_callback_on_completion(callback)
         callback.call
       end
 
-      def pr_notify_blocked(promise)
+      def pr_callback_notify_blocked(promise)
         promise.on_done self
       end
 
       def call_callback(method, *args)
-        # all methods has to be pure
         self.send method, *args
       end
 
@@ -273,7 +285,7 @@ def to_s
         end
       end
 
-      Failed  = ImmutableStruct.new :reason do
+      Failed = ImmutableStruct.new :reason do
         def value
           nil
         end
@@ -395,9 +407,15 @@ def on_failure!(&callback)
         add_callback :pr_callback_on_failure, callback
       end
 
+      # @api private
+      def apply_value(value, block)
+        block.call value
+      end
+
       # @api private
       def complete(success, value, reason, raise = true)
         if complete_state success, value, reason
+          @Waiters.clear
           synchronize { ns_broadcast }
           call_callbacks success, value, reason
         else
@@ -414,6 +432,7 @@ def add_callback(method, *args)
         else
           @Callbacks.push [method, *args]
           state = self.state
+          # take back if it was completed in the meanwhile
           call_callbacks success?(state), state.value, state.reason if completed?(state)
         end
         self
@@ -439,6 +458,10 @@ def call_callbacks(success, value, reason)
         end
       end
 
+      def call_callback(method, success, value, reason, *args)
+        self.send method, success, value, reason, *args
+      end
+
       def pr_async_callback_on_success(success, value, reason, executor, callback)
         pr_with_async(executor, success, value, reason, callback) do |success, value, reason, callback|
           pr_callback_on_success success, value, reason, callback
@@ -452,7 +475,7 @@ def pr_async_callback_on_failure(success, value, reason, executor, callback)
       end
 
       def pr_callback_on_success(success, value, reason, callback)
-        callback.call value if success
+        apply_value value, callback if success
       end
 
       def pr_callback_on_failure(success, value, reason, callback)
@@ -463,7 +486,7 @@ def pr_callback_on_completion(success, value, reason, callback)
         callback.call success, value, reason
       end
 
-      def pr_notify_blocked(success, value, reason, promise)
+      def pr_callback_notify_blocked(success, value, reason, promise)
         super(promise)
       end
 
@@ -506,11 +529,11 @@ def try_fail(reason = StandardError.new)
       end
 
       def evaluate_to(*args, &block)
-        promise.evaluate_to(*args, &block)
+        promise.evaluate_to(*args, block)
       end
 
       def evaluate_to!(*args, &block)
-        promise.evaluate_to!(*args, &block)
+        promise.evaluate_to!(*args, block)
       end
     end
 
@@ -609,8 +632,8 @@ def try_fail(reason = StandardError.new)
       public :evaluate_to
 
       # @return [Future]
-      def evaluate_to!(*args, &block)
-        evaluate_to(*args, &block).wait!
+      def evaluate_to!(*args, block)
+        evaluate_to(*args, block).wait!
       end
     end
 
@@ -625,7 +648,7 @@ def initialize(future, blocked_by_futures, countdown, &block)
         @Countdown = AtomicFixnum.new countdown
 
         super(future)
-        blocked_by.each { |f| f.add_callback :pr_notify_blocked, self }
+        blocked_by.each { |future| future.add_callback :pr_callback_notify_blocked, self }
       end
 
       # @api private
@@ -705,7 +728,9 @@ def initialize(blocked_by_future, default_executor = :io, executor = default_exe
 
       def on_completable(done_future)
         if done_future.success?
-          Concurrent.post_on(@Executor, done_future, @Task) { |done_future, task| evaluate_to done_future.value, &task }
+          Concurrent.post_on(@Executor, done_future, @Task) do |done_future, task|
+            evaluate_to { done_future.apply_value done_future.value, task }
+          end
         else
           complete false, nil, done_future.reason
         end
@@ -722,7 +747,7 @@ def initialize(blocked_by_future, default_executor = :io, executor = default_exe
 
       def on_completable(done_future)
         if done_future.failed?
-          Concurrent.post_on(@Executor, done_future, @Task) { |done_future, task| evaluate_to done_future.reason, &task }
+          Concurrent.post_on(@Executor, done_future.reason, @Task) { |reason, task| evaluate_to reason, &task }
         else
           complete true, done_future.value, nil
         end
@@ -757,12 +782,12 @@ def blocked_by
 
       def process_on_done(future)
         countdown = super(future)
-        value = future.value
+        value     = future.value
         if countdown.nonzero?
           case value
           when Future
             @BlockedBy.push value
-            value.add_callback :pr_notify_blocked, self
+            value.add_callback :pr_callback_notify_blocked, self
             @Countdown.value
           when Event
             raise TypeError, 'cannot flatten to Event'
@@ -797,26 +822,57 @@ def clear_blocked_by!
 
     # used internally to support #with_default_executor
     class AllPromise < BlockedPromise
+
+      class ArrayFuture < Future
+        def apply_value(value, block)
+          block.call(*value)
+        end
+      end
+
       private
 
       def initialize(blocked_by_futures, default_executor = :io)
-        klass = blocked_by_futures.any? { |f| f.is_a?(Future) } ? Future : Event
+        klass = Event
+        blocked_by_futures.each do |f|
+          if f.is_a?(Future)
+            if klass == Event
+              klass = Future
+            elsif klass == Future
+              klass = ArrayFuture
+              break
+            end
+          end
+        end
+
         # noinspection RubyArgCount
         super(klass.new(self, default_executor), blocked_by_futures, blocked_by_futures.size)
       end
 
       def on_completable(done_future)
-        results = blocked_by.select { |f| f.is_a?(Future) }.map(&:result)
-        if results.empty?
-          complete
-        else
-          if results.all? { |success, _, _| success }
-            params = results.map { |_, value, _| value }
-            complete(true, params.size == 1 ? params.first : params, nil)
+        all_success = true
+        reason      = nil
+
+        values = blocked_by.each_with_object([]) do |future, values|
+          next unless future.is_a?(Future)
+          success, value, reason = future.result
+
+          unless success
+            all_success = false
+            reason      = reason
+            break
+          end
+          values << value
+        end
+
+        if all_success
+          if values.empty?
+            complete
           else
-            # TODO what about other reasons?
-            complete false, nil, results.find { |success, _, _| !success }.last
+            complete(true, values.size == 1 ? values.first : values, nil)
           end
+        else
+          # TODO what about other reasons?
+          complete false, nil, reason
         end
       end
     end

lib/concurrent/edge/lock_free_stack.rb

@@ -52,8 +52,11 @@ def compare_and_clear(head)
       end
 
       def clear
-        @Head.update { |_| EMPTY }
-        self
+        while true
+          head = @Head.get
+          return false if head == EMPTY
+          return true if @Head.compare_and_set head, EMPTY
+        end
       end
 
       include Enumerable