Improved handling of dynamic attach for Pulsecount and Subtree engines. It's no longer fixed to a single dynamic predecessor and doesn't grow the stack for successive shifts.

schlangster · schlangster · commit 126bbbaf1175 · 2014-06-17T14:25:40.000+02:00
diff --git a/include/react/engine/PulsecountEngine.h b/include/react/engine/PulsecountEngine.h
@@ -57,7 +57,8 @@ enum class ENodeState
 {
     unchanged,
     changed,
-    deferred
+    dyn_defer,
+    dyn_repeat
 };
 
 class Node : public IReactiveNode
diff --git a/include/react/engine/SubtreeEngine.h b/include/react/engine/SubtreeEngine.h
@@ -67,6 +67,10 @@ class Node : public IReactiveNode
     inline void SetDeferredFlag()    { flags_.Set<flag_deferred>(); }
     inline void ClearDeferredFlag()  { flags_.Clear<flag_deferred>(); }
 
+    inline bool IsRepeated() const   { return flags_.Test<flag_repeated>(); }
+    inline void SetRepeatedFlag()    { flags_.Set<flag_repeated>(); }
+    inline void ClearRepeatedFlag()  { flags_.Clear<flag_repeated>(); }
+
     inline bool IsInitial() const   { return flags_.Test<flag_initial>(); }
     inline void SetInitialFlag()    { flags_.Set<flag_initial>(); }
     inline void ClearInitialFlag()  { flags_.Clear<flag_initial>(); }
@@ -96,9 +100,9 @@ class Node : public IReactiveNode
 
     NodeVector<Node>    Successors;
     ShiftMutexT         ShiftMutex;
-    uint16_t            Level       { 0 };
-    uint16_t            NewLevel    { 0 };
-    uint16_t            WaitCount   { 0 };
+    uint16_t            Level       = 0;
+    uint16_t            NewLevel    = 0;
+    uint16_t            WaitCount   = 0;
 
 private:
     enum EFlags : uint16_t
@@ -107,6 +111,7 @@ class Node : public IReactiveNode
         flag_marked,
         flag_changed,
         flag_deferred,
+        flag_repeated,
         flag_initial,
         flag_root
     };
@@ -159,10 +164,10 @@ class EngineBase : public IReactiveEngine<Node,TTurn>
     TopoQueueT      scheduledNodes_;
     vector<Node*>   subtreeRoots_;
 
-    empty_task*     rootTask_   { new(task::allocate_root()) empty_task };
+    empty_task&     rootTask_   = *new(task::allocate_root()) empty_task;
     task_list       spawnList_;
 
-    bool            isInPhase2_ { false };
+    bool            isInPhase2_ = false;
 }; 
 
 class BasicEngine : public EngineBase<Turn> {};
diff --git a/src/engine/PulsecountEngine.cpp b/src/engine/PulsecountEngine.cpp
@@ -112,9 +112,16 @@ class UpdaterTask: public task
             if (node.Mark() == ENodeMark::should_update)
                 node.Tick(&turn_);
 
-            if (node.State == ENodeState::deferred)
+            // Defer if node was dynamically attached to a predecessor that
+            // has not pulsed yet
+            if (node.State == ENodeState::dyn_defer)
                 continue;
 
+            // Repeat the update if a node was dynamically attached to a
+            // predecessor that has already pulsed
+            while (node.State == ENodeState::dyn_repeat)
+                node.Tick(&turn_);
+
             // Mark successors for update?
             bool update = node.State == ENodeState::changed;
             node.State = ENodeState::unchanged;
@@ -253,30 +260,23 @@ void EngineBase<TTurn>::OnNodeIdlePulse(Node& node, TTurn& turn)
 
 template <typename TTurn>
 void EngineBase<TTurn>::OnDynamicNodeAttach(Node& node, Node& parent, TTurn& turn)
-{
-    bool shouldTick = false;
-
-    {// parent.ShiftMutex (write)
-        NodeShiftMutexT::scoped_lock lock(parent.ShiftMutex, true);
+{// parent.ShiftMutex (write)
+    NodeShiftMutexT::scoped_lock lock(parent.ShiftMutex, true);
         
-        parent.Successors.Add(node);
-
-        // Has already nudged its neighbors?
-        if (parent.Mark() == ENodeMark::unmarked)
-        {
-            shouldTick = true;
-        }
-        else
-        {
-            node.State = ENodeState::deferred;
-            node.SetCounter(1);
-            node.SetMark(ENodeMark::should_update);
-        }
-    }// ~parent.ShiftMutex (write)
+    parent.Successors.Add(node);
 
-    if (shouldTick)
-        node.Tick(&turn);
-}
+    // Has already nudged its neighbors?
+    if (parent.Mark() == ENodeMark::unmarked)
+    {
+        node.State = ENodeState::dyn_repeat;
+    }
+    else
+    {
+        node.State = ENodeState::dyn_defer;
+        node.IncCounter();
+        node.SetMark(ENodeMark::should_update);
+    }
+}// ~parent.ShiftMutex (write)
 
 template <typename TTurn>
 void EngineBase<TTurn>::OnDynamicNodeDetach(Node& node, Node& parent, TTurn& turn)
diff --git a/src/engine/SubtreeEngine.cpp b/src/engine/SubtreeEngine.cpp
@@ -85,13 +85,24 @@ class UpdaterTask: public task
             node.ClearInitialFlag();
             node.SetShouldUpdate(false);
 
-            // Skip deferred node
+            // Defer if node was dynamically attached to a predecessor that
+            // has not pulsed yet
             if (node.IsDeferred())
             {
                 node.ClearDeferredFlag();
                 continue;
             }
 
+            // Repeat the update if a node was dynamically attached to a
+            // predecessor that has already pulsed
+            while (node.IsRepeated())
+            {
+                node.ClearRepeatedFlag();
+                node.Tick(&turn_);
+            }
+
+            node.SetReadyCount(0);
+
             // Mark successors for update?
             bool update = node.IsChanged();
 
@@ -107,8 +118,6 @@ class UpdaterTask: public task
                     if (succ->IncReadyCount())
                         continue;
 
-                    succ->SetReadyCount(0);
-
                     // Heavyweight - spawn new task
                     if (succ->IsHeavyweight())
                     {
@@ -124,11 +133,11 @@ class UpdaterTask: public task
 
                         if (nodes_.IsFull())
                         {
-                            splitCount++;
+                            ++splitCount;
 
                             //Delegate half the work to new task
                             auto& t = *new(task::allocate_additional_child_of(*parent()))
-                                UpdaterTask(*this, SplitTag{});
+                                UpdaterTask(*this, SplitTag());
 
                             spawn(t);
                         }
@@ -171,24 +180,24 @@ void EngineBase<TTurn>::Propagate(TTurn& turn)
     // Phase 2
     isInPhase2_ = true;
 
-    rootTask_->set_ref_count(1 + subtreeRoots_.size());
+    rootTask_.set_ref_count(1 + subtreeRoots_.size());
 
     for (auto* node : subtreeRoots_)
     {
         // Ignore if root flag has been cleared because node was part of another subtree
         if (! node->IsRoot())
         {
-            rootTask_->decrement_ref_count();
+            rootTask_.decrement_ref_count();
             continue;
         }
 
-        spawnList_.push_back(*new(rootTask_->allocate_child())
+        spawnList_.push_back(*new(rootTask_.allocate_child())
             UpdaterTask<TTurn>(turn, node));
 
         node->ClearRootFlag();
     }
 
-    rootTask_->spawn_and_wait_for_all(spawnList_);
+    rootTask_.spawn_and_wait_for_all(spawnList_);
 
     subtreeRoots_.clear();
     spawnList_.clear();
@@ -242,36 +251,28 @@ void EngineBase<TTurn>::OnDynamicNodeDetach(Node& node, Node& parent, TTurn& tur
 
 template <typename TTurn>
 void EngineBase<TTurn>::applyAsyncDynamicAttach(Node& node, Node& parent, TTurn& turn)
-{
-    bool shouldTick = false;
-
-    {// parent.ShiftMutex (write)
-        NodeShiftMutexT::scoped_lock    lock(parent.ShiftMutex, true);
+{// parent.ShiftMutex (write)
+    NodeShiftMutexT::scoped_lock    lock(parent.ShiftMutex, true);
         
-        parent.Successors.Add(node);
-
-        // Level recalulation applied when added to topoqueue next time.
-        // During the async phase 2 it's not needed.
-        if (node.NewLevel <= parent.Level)
-            node.NewLevel = parent.Level + 1;
-
-        // Has already nudged its neighbors?
-        if (! parent.IsMarked())
-        {
-            shouldTick = true;
-        }
-        else
-        {
-            node.SetDeferredFlag();
-            node.SetShouldUpdate(true);
+    parent.Successors.Add(node);
 
-            node.WaitCount = 1;
-        }
-    }// ~parent.ShiftMutex (write)
+    // Level recalulation applied when added to topoqueue next time.
+    // During the async phase 2 it's not needed.
+    if (node.NewLevel <= parent.Level)
+        node.NewLevel = parent.Level + 1;
 
-    if (shouldTick)
-        node.Tick(&turn);
-}
+    // Has already nudged its neighbors?
+    if (! parent.IsMarked())
+    {
+        node.SetRepeatedFlag();
+    }
+    else
+    {
+        node.SetDeferredFlag();
+        node.SetShouldUpdate(true);
+        node.DecReadyCount();
+    }
+}// ~parent.ShiftMutex (write)
 
 template <typename TTurn>
 void EngineBase<TTurn>::applyAsyncDynamicDetach(Node& node, Node& parent, TTurn& turn)

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,8 @@ enum class ENodeState`
`57`	`57`	`{`
`58`	`58`	`unchanged,`
`59`	`59`	`changed,`
`60`		`- deferred`
	`60`	`+ dyn_defer,`
	`61`	`+ dyn_repeat`
`61`	`62`	`};`
`62`	`63`
`63`	`64`	`class Node : public IReactiveNode`