Add smart propagation prevention

Do not inform dependent cells about updates, if another update is in sight.

Author: JanKoelzer

core/src/main/scala/cell/Cell.scala

@@ -1,6 +1,6 @@
 package cell
 
-import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.atomic.{ AtomicInteger, AtomicReference }
 import java.util.concurrent.{ CountDownLatch, ExecutionException }
 
 import scala.annotation.tailrec
@@ -106,6 +106,9 @@ trait Cell[K <: Key[V], V] {
 
   private[cell] def removeCompleteCallbacks(cell: Cell[K, V]): Unit
   private[cell] def removeNextCallbacks(cell: Cell[K, V]): Unit
+
+  private[cell] def incIncomingCallbacks(): Int
+  private[cell] def decIncomingCallbacks(): Int
 }
 
 object Cell {
@@ -181,6 +184,7 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
    * Assumes that dependencies need to be kept until a final result is known.
    */
   private val state = new AtomicReference[AnyRef](State.empty[K, V](lattice))
+  private val numIncomingCallbacks = new AtomicInteger(0) // Should this be included into `state`?
 
   // `CellCompleter` and corresponding `Cell` are the same run-time object.
   override def cell: Cell[K, V] = this
@@ -428,10 +432,10 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
           if (!state.compareAndSet(current, newState)) {
             tryNewState(value)
           } else {
-            // CAS was successful, so there was a point in time where `newVal` was in the cell
-            current.nextCallbacks.values.foreach { callbacks =>
-              callbacks.foreach(callback => callback.execute())
-            }
+            // If we came here via a direct putNext (instead of Outcome of a whenNextCallback)
+            // this incoming change has not been counted. So we need to manually start outgoing callbacks.
+            // (This might lead to duplicate invocation.)
+            if (numIncomingCallbacks.get() == 0) triggerNextCallbacks() //
             true
           }
         } else true
@@ -468,28 +472,12 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
         res
 
       case (pre: State[K, V], newVal: Try[V]) =>
+        triggerCompleteCallbacks(pre)
+
         val depsCells = pre.completeDeps
         val nextDepsCells = pre.nextDeps
-        if (pre.completeCallbacks.isEmpty) {
-          // call callbacks
-          pre.nextCallbacks.values.foreach { callbacks =>
-            callbacks.foreach(callback => callback.execute())
-          }
-        } else {
-          // onNext callbacks with these cells should not be triggered, because they have
-          // onComplete callbacks which are triggered instead.
-          pre.completeCallbacks.values.foreach { callbacks =>
-            callbacks.foreach(callback => callback.execute())
-          }
-          // call (concurrent) callbacks
-          pre.nextCallbacks.values.foreach { callbacks =>
-            callbacks.foreach { callback =>
-              if (!pre.completeCallbacks.contains(callback.dependentCell))
-                callback.execute()
-            }
-          }
-        }
 
+        // Other cells do not need to call as any more
         if (depsCells.nonEmpty)
           depsCells.foreach(_.removeCompleteCallbacks(this))
         if (nextDepsCells.nonEmpty)
@@ -503,6 +491,42 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
     res
   }
 
+  private def triggerNextCallbacks(): Unit = {
+    state.get() match {
+      case _: Try[_] => /* Meanwhile, cell has been completed. No need to trigger `next` callbacks any more. */
+      case raw: State[_, _] =>
+        val current = raw.asInstanceOf[State[K, V]]
+        current.nextCallbacks.values.foreach { callbacks =>
+          callbacks.foreach(callback => callback.execute())
+        }
+    }
+  }
+
+  // We need to pass the list of callbacks here, because they
+  // might have already been deleted in `state` while the cell has
+  // been set to its final value.
+  private def triggerCompleteCallbacks(pre: State[K, V]): Unit = {
+    if (pre.completeCallbacks.isEmpty) {
+      // call callbacks
+      pre.nextCallbacks.values.foreach { callbacks =>
+        callbacks.foreach(callback => callback.execute())
+      }
+    } else {
+      // onNext callbacks with these cells should not be triggered, because they have
+      // onComplete callbacks which are triggered instead.
+      pre.completeCallbacks.values.foreach { callbacks =>
+        callbacks.foreach(callback => callback.execute())
+      }
+      // call (concurrent) callbacks
+      pre.nextCallbacks.values.foreach { callbacks =>
+        callbacks.foreach { callback =>
+          if (!pre.completeCallbacks.contains(callback.dependentCell))
+            callback.execute()
+        }
+      }
+    }
+  }
+
   @tailrec
   override private[cell] final def removeDep(cell: Cell[K, V]): Unit = {
     state.get() match {
@@ -516,7 +540,7 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
         else if (newDeps.isEmpty)
           nodepslatch.countDown()
 
-      case _ => /* do nothing */
+      case _ => /* `this` has been completed and therefore does not have any deps stored. */
     }
   }
 
@@ -533,7 +557,7 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
         else if (newNextDeps.isEmpty)
           nonextdepslatch.countDown()
 
-      case _ => /* do nothing */
+      case _ => /* `this` has been completed and therefore does not have any deps stored. */
     }
   }
 
@@ -643,4 +667,17 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, lattice: L
     case t => Failure(t)
   }
 
+  /** Called, when a CallbackRunnable r with r.dependentCell == this has been started. */
+  override private[cell] def incIncomingCallbacks(): Int = {
+    numIncomingCallbacks.incrementAndGet()
+  }
+
+  /** Called, when a CallbackRunnable r with r.dependentCell == this has been completed.
+    * Triggers all outgoing callbacks, if no more incoming callbacks are running.
+    */
+  override private[cell] def decIncomingCallbacks(): Int = {
+    val newValue = numIncomingCallbacks.decrementAndGet()
+    if (newValue == 0) triggerNextCallbacks()
+    newValue
+  }
 }

core/src/main/scala/cell/callbackRunnable.scala

@@ -65,6 +65,7 @@ private[cell] trait Dependency[K <: Key[V], V] {
 
 /**
  * To be run when `otherCell` gets its final update.
+ *
  * @param pool          The handler pool that runs the callback function
  * @param dependentCell The cell, that depends on `otherCell`.
  * @param otherCell     Cell that triggers this callback.
@@ -83,7 +84,10 @@ private[cell] abstract class CompleteCallbackRunnable[K <: Key[V], V](
   def run(): Unit = {
     require(!started) // can't complete it twice
     started = true
+
+    dependentCell.incIncomingCallbacks()
     callback(Success(otherCell.getResult()))
+    dependentCell.decIncomingCallbacks()
   }
 }
 
@@ -166,7 +170,9 @@ private[cell] abstract class NextCallbackRunnable[K <: Key[V], V](
   extends CallbackRunnable[K, V] {
 
   def run(): Unit = {
+    dependentCell.incIncomingCallbacks()
     callback(Success(otherCell.getResult()))
+    dependentCell.decIncomingCallbacks()
   }
 }