Fix callback yielded arguments and delay propagation

Author: pitr-ch

lib/concurrent/edge/promises.rb

@@ -727,7 +727,18 @@ def waiting_threads
         @Waiters.each.to_a
       end
 
+      # # @!visibility private
+      def add_callback_notify_blocked(promise, index)
+        add_callback :callback_notify_blocked, promise, index
+      end
+
       # @!visibility private
+      def add_callback_clear_delayed_node(node)
+        add_callback(:callback_clear_delayed_node, node)
+      end
+
+      private
+
       def add_callback(method, *args)
         state = internal_state
         if resolved?(state)
@@ -741,7 +752,9 @@ def add_callback(method, *args)
         self
       end
 
-      private
+      def callback_clear_delayed_node(state, node)
+        node.value = nil
+      end
 
       # @return [Boolean]
       def wait_until_resolved(timeout)
@@ -1202,7 +1215,7 @@ def callback_on_rejection(state, args, callback)
       end
 
       def callback_on_resolution(state, args, callback)
-        callback.call state.result, *args
+        callback.call *state.result, *args
       end
 
     end
@@ -1326,7 +1339,7 @@ def touch
 
       alias_method :inspect, :to_s
 
-      def delayed
+      def delayed_because
         nil
       end
 
@@ -1381,55 +1394,45 @@ class BlockedPromise < InnerPromise
       private_class_method :new
 
       def self.new_blocked_by1(blocker, *args, &block)
-        blocker_delayed = blocker.promise.delayed
-        delayed         = blocker_delayed ? LockFreeStack.new.push(blocker_delayed) : nil
-        promise         = new(delayed, 1, *args, &block)
-        blocker.add_callback :callback_notify_blocked, promise, 0
+        blocker_delayed = blocker.promise.delayed_because
+        promise         = new(blocker_delayed, 1, *args, &block)
+        blocker.add_callback_notify_blocked promise, 0
         promise
       end
 
       def self.new_blocked_by2(blocker1, blocker2, *args, &block)
-        blocker_delayed1 = blocker1.promise.delayed
-        blocker_delayed2 = blocker2.promise.delayed
-        # TODO (pitr-ch 23-Dec-2016): use arrays when we know it will not grow (only flat adds delay)
-        delayed          = if blocker_delayed1
-                             if blocker_delayed2
-                               LockFreeStack.of2(blocker_delayed1, blocker_delayed2)
-                             else
-                               LockFreeStack.of1(blocker_delayed1)
-                             end
+        blocker_delayed1 = blocker1.promise.delayed_because
+        blocker_delayed2 = blocker2.promise.delayed_because
+        delayed          = if blocker_delayed1 && blocker_delayed2
+                             # TODO (pitr-ch 23-Dec-2016): use arrays when we know it will not grow (only flat adds delay)
+                             LockFreeStack.of2(blocker_delayed1, blocker_delayed2)
                            else
-                             blocker_delayed2 ? LockFreeStack.of1(blocker_delayed2) : nil
+                             blocker_delayed1 || blocker_delayed2
                            end
         promise          = new(delayed, 2, *args, &block)
-        blocker1.add_callback :callback_notify_blocked, promise, 0
-        blocker2.add_callback :callback_notify_blocked, promise, 1
+        blocker1.add_callback_notify_blocked promise, 0
+        blocker2.add_callback_notify_blocked promise, 1
         promise
       end
 
       def self.new_blocked_by(blockers, *args, &block)
-        delayed = blockers.reduce(nil, &method(:add_delayed))
+        delayed = blockers.reduce(nil) { |d, f| add_delayed d, f.promise.delayed_because }
         promise = new(delayed, blockers.size, *args, &block)
-        blockers.each_with_index { |f, i| f.add_callback :callback_notify_blocked, promise, i }
+        blockers.each_with_index { |f, i| f.add_callback_notify_blocked promise, i }
         promise
       end
 
-      def self.add_delayed(delayed, blocker)
-        blocker_delayed = blocker.promise.delayed
-        if blocker_delayed
-          delayed = unless delayed
-                      LockFreeStack.of1(blocker_delayed)
-                    else
-                      delayed.push(blocker_delayed)
-                    end
+      def self.add_delayed(delayed1, delayed2)
+        if delayed1 && delayed2
+          delayed1.push delayed2
+          delayed1
+        else
+          delayed1 || delayed2
         end
-        delayed
       end
 
       def initialize(delayed, blockers_count, future)
         super(future)
-        # noinspection RubyArgCount
-        @Touched   = AtomicBoolean.new false
         @Delayed   = delayed
         # noinspection RubyArgCount
         @Countdown = AtomicFixnum.new blockers_count
@@ -1442,16 +1445,12 @@ def on_blocker_resolution(future, index)
         on_resolvable(future, index) if resolvable
       end
 
-      def delayed
+      def delayed_because
         @Delayed
       end
 
       def touch
-        clear_propagate_touch if @Touched.make_true
-      end
-
-      def touched?
-        @Touched.value
+        clear_and_propagate_touch
       end
 
       # for inspection only
@@ -1463,13 +1462,11 @@ def blocked_by
 
       private
 
-      def clear_propagate_touch
-        @Delayed.clear_each { |o| propagate_touch o } if @Delayed
-      end
+      def clear_and_propagate_touch(stack_or_element = @Delayed)
+        return if stack_or_element.nil?
 
-      def propagate_touch(stack_or_element = @Delayed)
         if stack_or_element.is_a? LockFreeStack
-          stack_or_element.each { |element| propagate_touch element }
+          stack_or_element.clear_each { |element| clear_and_propagate_touch element }
         else
           stack_or_element.touch unless stack_or_element.nil? # if still present
         end
@@ -1572,8 +1569,28 @@ def initialize(default_executor, fulfilled, value, reason)
 
     class AbstractFlatPromise < BlockedPromise
 
+      def initialize(delayed_because, blockers_count, event_or_future)
+        delayed = LockFreeStack.of1(self)
+        super(delayed, blockers_count, event_or_future)
+        # noinspection RubyArgCount
+        @Touched        = AtomicBoolean.new false
+        @DelayedBecause = delayed_because || LockFreeStack.new
+
+        event_or_future.add_callback_clear_delayed_node delayed.peek
+      end
+
+      def touch
+        if @Touched.make_true
+          clear_and_propagate_touch @DelayedBecause
+        end
+      end
+
       private
 
+      def touched?
+        @Touched.value
+      end
+
       def on_resolvable(resolved_future, index)
         resolve_with resolved_future.internal_state
       end
@@ -1583,11 +1600,12 @@ def resolvable?(countdown, future, index)
       end
 
       def add_delayed_of(future)
+        delayed = future.promise.delayed_because
         if touched?
-          propagate_touch future.promise.delayed
+          clear_and_propagate_touch delayed
         else
-          BlockedPromise.add_delayed @Delayed, future
-          clear_propagate_touch if touched?
+          BlockedPromise.add_delayed @DelayedBecause, delayed
+          clear_and_propagate_touch @DelayedBecause if touched?
         end
       end
 
@@ -1615,7 +1633,7 @@ def process_on_blocker_resolution(future, index)
           case value
           when Future, Event
             add_delayed_of value
-            value.add_callback :callback_notify_blocked, self, nil
+            value.add_callback_notify_blocked self, nil
             countdown
           else
             resolve_with RESOLVED
@@ -1651,7 +1669,7 @@ def process_on_blocker_resolution(future, index)
           case value
           when Future
             add_delayed_of value
-            value.add_callback :callback_notify_blocked, self, nil
+            value.add_callback_notify_blocked self, nil
             countdown
           when Event
             evaluate_to(lambda { raise TypeError, 'cannot flatten to Event' })
@@ -1684,7 +1702,7 @@ def process_on_blocker_resolution(future, index)
         case value
         when Future
           add_delayed_of value
-          value.add_callback :callback_notify_blocked, self, nil
+          value.add_callback_notify_blocked self, nil
         else
           resolve_with internal_state
         end
@@ -1851,17 +1869,17 @@ def resolvable?(countdown, future, index)
     class DelayPromise < InnerPromise
 
       def initialize(default_executor)
-        super event = Event.new(self, default_executor)
-        @Delayed = LockFreeStack.new.push self
-        # TODO (pitr-ch 20-Dec-2016): implement directly without callback?
-        event.on_resolution!(@Delayed.peek) { |stack_node| stack_node.value = nil }
+        event    = Event.new(self, default_executor)
+        @Delayed = LockFreeStack.of1(self)
+        super event
+        event.add_callback_clear_delayed_node @Delayed.peek
       end
 
       def touch
         @Future.resolve_with RESOLVED
       end
 
-      def delayed
+      def delayed_because
         @Delayed
       end
 

spec/concurrent/edge/promises_spec.rb

@@ -244,28 +244,71 @@ def behaves_as_delay(delay, value)
 
   describe 'Future' do
     it 'has sync and async callbacks' do
-      callbacks_tester = ->(future) do
-        queue = Queue.new
-        future.on_resolution_using(:io) { |result| queue.push("async on_resolution #{ result.inspect }") }
-        future.on_resolution! { |result| queue.push("sync on_resolution #{ result.inspect }") }
-        future.on_fulfillment_using(:io) { |value| queue.push("async on_fulfillment #{ value.inspect }") }
-        future.on_fulfillment! { |value| queue.push("sync on_fulfillment #{ value.inspect }") }
-        future.on_rejection_using(:io) { |reason| queue.push("async on_rejection #{ reason.inspect }") }
-        future.on_rejection! { |reason| queue.push("sync on_rejection #{ reason.inspect }") }
-        future.wait
-        [queue.pop, queue.pop, queue.pop, queue.pop].sort
+      callbacks_tester = ->(event_or_future) do
+        queue     = Queue.new
+        push_args = -> *args { queue.push args }
+
+        event_or_future.on_resolution! &push_args
+        event_or_future.on_resolution!(1, &push_args)
+        if event_or_future.is_a? Concurrent::Promises::Future
+          event_or_future.on_fulfillment! &push_args
+          event_or_future.on_fulfillment!(2, &push_args)
+          event_or_future.on_rejection! &push_args
+          event_or_future.on_rejection!(3, &push_args)
+        end
+
+        event_or_future.on_resolution &push_args
+        event_or_future.on_resolution(4, &push_args)
+        if event_or_future.is_a? Concurrent::Promises::Future
+          event_or_future.on_fulfillment &push_args
+          event_or_future.on_fulfillment(5, &push_args)
+          event_or_future.on_rejection &push_args
+          event_or_future.on_rejection(6, &push_args)
+        end
+        event_or_future.on_resolution_using(:io, &push_args)
+        event_or_future.on_resolution_using(:io, 7, &push_args)
+        if event_or_future.is_a? Concurrent::Promises::Future
+          event_or_future.on_fulfillment_using(:io, &push_args)
+          event_or_future.on_fulfillment_using(:io, 8, &push_args)
+          event_or_future.on_rejection_using(:io, &push_args)
+          event_or_future.on_rejection_using(:io, 9, &push_args)
+        end
+
+        event_or_future.wait
+        Array.new(event_or_future.is_a?(Concurrent::Promises::Future) ? 12 : 6) { queue.pop }
       end
-      callback_results = callbacks_tester.call(future { :value })
-      expect(callback_results).to eq ["async on_fulfillment :value",
-                                      "async on_resolution [true, :value, nil]",
-                                      "sync on_fulfillment :value",
-                                      "sync on_resolution [true, :value, nil]"]
 
-      callback_results = callbacks_tester.call(future { raise 'error' })
-      expect(callback_results).to eq ["async on_rejection #<RuntimeError: error>",
-                                      "async on_resolution [false, nil, #<RuntimeError: error>]",
-                                      "sync on_rejection #<RuntimeError: error>",
-                                      "sync on_resolution [false, nil, #<RuntimeError: error>]"]
+      callback_results = callbacks_tester.call(fulfilled_future(:v))
+      expect(callback_results).to contain_exactly([true, :v, nil],
+                                                  [true, :v, nil, 1],
+                                                  [:v],
+                                                  [:v, 2],
+                                                  [true, :v, nil],
+                                                  [true, :v, nil, 4],
+                                                  [:v],
+                                                  [:v, 5],
+                                                  [true, :v, nil],
+                                                  [true, :v, nil, 7],
+                                                  [:v],
+                                                  [:v, 8])
+
+      err              = StandardError.new 'boo'
+      callback_results = callbacks_tester.call(rejected_future(err))
+      expect(callback_results).to contain_exactly([false, nil, err],
+                                                  [false, nil, err, 1],
+                                                  [err],
+                                                  [err, 3],
+                                                  [false, nil, err],
+                                                  [false, nil, err, 4],
+                                                  [err],
+                                                  [err, 6],
+                                                  [false, nil, err],
+                                                  [false, nil, err, 7],
+                                                  [err],
+                                                  [err, 9])
+
+      callback_results = callbacks_tester.call(resolved_event)
+      expect(callback_results).to contain_exactly([], [1], [], [4], [], [7])
     end
 
     [:wait, :wait!, :value, :value!, :reason, :result].each do |method_with_timeout|
@@ -387,6 +430,8 @@ def behaves_as_delay(delay, value)
 
       it 'propagates requests for values to delayed futures' do
         expect(future { delay { 1 } }.flat.value!(0.1)).to eq 1
+        expect(Array.new(3) { |i| Concurrent::Promises.delay { i } }.
+            inject { |a, b| a.then { b }.flat }.value!(0.2)).to eq 2
       end
     end