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
@@ -111,6 +111,9 @@ trait Cell[K <: Key[V], V] {
 
   def removeCompleteCallbacks(cell: Cell[K, V]): Unit
   def removeNextCallbacks(cell: Cell[K, V]): Unit
+
+  private[cell] def incIncomingCallbacks(): Int
+  private[cell] def decIncomingCallbacks(): Int
 }
 
 object Cell {
@@ -189,6 +192,7 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
    * Assumes that dependencies need to be kept until a final result is known.
    */
   private val state = new AtomicReference[AnyRef](State.empty[K, V](updater))
+  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
@@ -439,10 +443,10 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
           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
@@ -479,16 +483,12 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
         res
 
       case (pre: State[K, V], newVal: Try[V]) =>
-        pre.nextCallbacks.values.foreach { callbacks =>
-          callbacks.foreach(callback => callback.execute())
-        }
-        pre.completeCallbacks.values.foreach { callbacks =>
-          callbacks.foreach(callback => callback.execute())
-        }
+        triggerCompleteCallbacks(pre)
 
         val depsCells = pre.completeDeps
         val nextDepsCells = pre.nextDeps
 
+        // Other cells do not need to call us any more
         if (depsCells.nonEmpty)
           depsCells.foreach(_.removeCompleteCallbacks(this))
         if (nextDepsCells.nonEmpty)
@@ -502,6 +502,29 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
     res
   }
 
+  // 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 = {
+    pre.nextCallbacks.values.foreach { callbacks =>
+      callbacks.foreach(callback => callback.execute())
+    }
+    pre.completeCallbacks.values.foreach { callbacks =>
+      callbacks.foreach(callback => callback.execute())
+    }
+  }
+
+  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())
+        }
+    }
+  }
+
   @tailrec
   override private[cell] final def removeDep(cell: Cell[K, V]): Unit = {
     state.get() match {
@@ -515,7 +538,7 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
         else if (newDeps.isEmpty)
           nodepslatch.countDown()
 
-      case _ => /* do nothing */
+      case _ => /* `this` has been completed and therefore does not have any deps stored. */
     }
   }
 
@@ -532,7 +555,7 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
         else if (newNextDeps.isEmpty)
           nonextdepslatch.countDown()
 
-      case _ => /* do nothing */
+      case _ => /* `this` has been completed and therefore does not have any deps stored. */
     }
   }
 
@@ -666,4 +689,18 @@ private class CellImpl[K <: Key[V], V](pool: HandlerPool, val key: K, updater: U
     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,12 @@ private[cell] abstract class CompleteCallbackRunnable[K <: Key[V], V](
   def run(): Unit = {
     require(!started) // can't complete it twice
     started = true
+
+    if (dependentCell != null) // dependentCell == null for oncomplete callbacks
+      dependentCell.incIncomingCallbacks()
     callback(Success(otherCell.getResult()))
+    if (dependentCell != null)
+      dependentCell.decIncomingCallbacks()
   }
 }
 
@@ -166,7 +172,11 @@ private[cell] abstract class NextCallbackRunnable[K <: Key[V], V](
   extends CallbackRunnable[K, V] {
 
   def run(): Unit = {
+    if (dependentCell != null) // dependetCell == null for onnext callbacks
+      dependentCell.incIncomingCallbacks()
     callback(Success(otherCell.getResult()))
+    if (dependentCell != null)
+      dependentCell.decIncomingCallbacks()
   }
 }