PulseCount engine now spawns new task for heavyweight nodes and splits range of changed inputs to multiple tasks if necessary.

schlangster · schlangster · commit 2fe0b5217ebc · 2014-04-15T15:30:55.000+02:00
diff --git a/include/react/propagation/PulseCountEngine.h b/include/react/propagation/PulseCountEngine.h
@@ -29,6 +29,7 @@ using std::vector;
 using tbb::task;
 using tbb::empty_task;
 using tbb::spin_rw_mutex;
+using tbb::task_list;
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 /// Turn
@@ -84,6 +85,7 @@ class Node : public IReactiveNode
     NodeVector<Node>    Successors;
     
     ENodeState          State       = ENodeState::unchanged;
+    uint                Weight      = 0;
 
 private:
     atomic<int>         counter_    = 0;
@@ -112,9 +114,12 @@ class EngineBase : public IReactiveEngine<Node,TTurn>
     void OnDynamicNodeAttach(Node& node, Node& parent, TTurn& turn);
     void OnDynamicNodeDetach(Node& node, Node& parent, TTurn& turn);
 
+    void HintUpdateDuration(Node& node, uint dur);
+
 private:
     NodeVectT       changedInputs_;
     empty_task*     rootTask_ = new(task::allocate_root()) empty_task;
+    task_list       spawnList_;
 };
 
 class BasicEngine : public EngineBase<Turn> {};
@@ -135,3 +140,16 @@ template <> class PulseCountEngine<parallel> : public REACT_IMPL::pulsecount::Ba
 template <> class PulseCountEngine<parallel_queuing> : public REACT_IMPL::pulsecount::QueuingEngine {};
 
 /******************************************/ REACT_END /******************************************/
+
+/***************************************/ REACT_IMPL_BEGIN /**************************************/
+
+template <typename T>
+struct EnableNodeUpdateTimer;
+
+template <>
+struct EnableNodeUpdateTimer<PulseCountEngine<parallel>> : std::true_type {};
+
+template <>
+struct EnableNodeUpdateTimer<PulseCountEngine<parallel_queuing>> : std::true_type {};
+
+/****************************************/ REACT_IMPL_END /***************************************/
diff --git a/src/react/propagation/PulseCountEngine.cpp b/src/react/propagation/PulseCountEngine.cpp
@@ -14,30 +14,39 @@
 /***************************************/ REACT_IMPL_BEGIN /**************************************/
 namespace pulsecount {
 
+///////////////////////////////////////////////////////////////////////////////////////////////////
+/// Constants
+///////////////////////////////////////////////////////////////////////////////////////////////////
+static const uint chunk_size    = 8;
+static const uint dfs_threshold = 3;
+static const uint heavy_weight  = 1000;
+
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 /// MarkerTask
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 class MarkerTask: public task
 {
 public:
-    using StackT = NodeBuffer<Node,8>;
+    using BufferT = NodeBuffer<Node,chunk_size>;
 
     template <typename TInput>
     MarkerTask(TInput srcBegin, TInput srcEnd) :
         nodes_{ srcBegin, srcEnd }
-    {}
+    {
+    }
 
     MarkerTask(MarkerTask& other, SplitTag) :
         nodes_{ other.nodes_, SplitTag{} }
-    {}
+    {
+    }
 
     task* execute()
     {
         int splitCount = 0;
 
         while (! nodes_.IsEmpty())
         {
-            Node& node = splitCount > 3 ? *nodes_.PopBack() : *nodes_.PopFront();
+            Node& node = splitCount > dfs_threshold ? *nodes_.PopBack() : *nodes_.PopFront();
 
             // Increment counter of each successor and add it to smaller stack
             for (auto* succ : node.Successors)
@@ -67,7 +76,7 @@ class MarkerTask: public task
     }
 
 private:
-    StackT  nodes_;
+    BufferT  nodes_;
 };
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -77,14 +86,19 @@ template <typename TTurn>
 class UpdaterTask: public task
 {
 public:
-    using BufferT = NodeBuffer<Node,8>;
+    using BufferT = NodeBuffer<Node,chunk_size>;
 
     template <typename TInput>
     UpdaterTask(TTurn& turn, TInput srcBegin, TInput srcEnd) :
         turn_{ turn },
         nodes_{ srcBegin, srcEnd }
     {}
 
+    UpdaterTask(TTurn& turn, Node* node) :
+        turn_{ turn },
+        nodes_{ node }
+    {}
+
     UpdaterTask(UpdaterTask& other, SplitTag) :
         turn_{ other.turn_ },
         nodes_{ other.nodes_, SplitTag{} }
@@ -96,7 +110,7 @@ class UpdaterTask: public task
 
         while (!nodes_.IsEmpty())
         {
-            Node& node = splitCount > 3 ? *nodes_.PopBack() : *nodes_.PopFront();
+            Node& node = splitCount > dfs_threshold ? *nodes_.PopBack() : *nodes_.PopFront();
 
             if (node.Mark() == ENodeMark::should_update)
                 node.Tick(&turn_);
@@ -120,18 +134,30 @@ class UpdaterTask: public task
                     if (succ->DecCounter())
                         continue;
 
-                    nodes_.PushBack(succ);
-
-                    if (nodes_.IsFull())
+                    // Heavyweight - spawn new task
+                    if (succ->Weight > heavy_weight)
                     {
-                        splitCount++;
-
-                        //Delegate half the work to new task
                         auto& t = *new(task::allocate_additional_child_of(*parent()))
-                            UpdaterTask(*this, SplitTag{});
+                            UpdaterTask(turn_, succ);
 
                         spawn(t);
                     }
+                    // Leightweight - add to buffer, split if full
+                    else
+                    {
+                        nodes_.PushBack(succ);
+
+                        if (nodes_.IsFull())
+                        {
+                            splitCount++;
+
+                            //Delegate half the work to new task
+                            auto& t = *new(task::allocate_additional_child_of(*parent()))
+                                UpdaterTask(*this, SplitTag{});
+
+                            spawn(t);
+                        }
+                    }
                 }
 
                 node.SetMark(ENodeMark::unmarked);
@@ -178,31 +204,41 @@ void EngineBase<TTurn>::OnTurnInputChange(Node& node, TTurn& turn)
     node.State = ENodeState::changed;
 }
 
-//static vector<Node*> markNodes_;
-
-template <typename TTurn>
-void EngineBase<TTurn>::OnTurnPropagate(TTurn& turn)
+template <typename TTask, typename TIt, typename ... TArgs>
+void spawnHelper
+(
+    task* rootTask, task_list& spawnList,
+    const int count, TIt start, TIt end,
+    TArgs& ... args
+)
 {
-    if (changedInputs_.size() <= 8)
-    {
-        auto& markerTask = *new(rootTask_->allocate_child())
-            MarkerTask(changedInputs_.begin(), changedInputs_.end());
+    rootTask->set_ref_count(1 + count);
 
-        rootTask_->set_ref_count(2);
-        rootTask_->spawn_and_wait_for_all(markerTask);
-    }
-    else
+    for (int i=0; i < (count - 1); i++)
     {
-        REACT_ASSERT(false, "not implemented yet\n");
+        spawnList.push_back(*new(rootTask->allocate_child())
+            TTask(args ..., start, start + chunk_size));
+        start += chunk_size;
     }
 
-    auto& updaterTask = *new(rootTask_->allocate_child())
-        UpdaterTask<TTurn>(turn, changedInputs_.begin(), changedInputs_.end());
+    spawnList.push_back(*new(rootTask->allocate_child())
+        TTask(args ..., start, end));
 
-    rootTask_->set_ref_count(2);
-    rootTask_->spawn_and_wait_for_all(updaterTask);
+    rootTask->spawn_and_wait_for_all(spawnList);
 
-    changedInputs_.clear();
+    spawnList.clear();
+}
+
+template <typename TTurn>
+void EngineBase<TTurn>::OnTurnPropagate(TTurn& turn)
+{
+    const int initialTaskCount = (changedInputs_.size() - 1) / chunk_size + 1;
+
+    spawnHelper<MarkerTask>(rootTask_, spawnList_, initialTaskCount,
+        changedInputs_.begin(), changedInputs_.end());
+
+    spawnHelper<UpdaterTask<TTurn>>(rootTask_, spawnList_, initialTaskCount,
+        changedInputs_.begin(), changedInputs_.end(), turn);
 }
 
 template <typename TTurn>
@@ -252,6 +288,12 @@ void EngineBase<TTurn>::OnDynamicNodeDetach(Node& node, Node& parent, TTurn& tur
     parent.Successors.Remove(node);
 }// ~oldParent.ShiftMutex (write)
 
+template <typename TTurn>
+void EngineBase<TTurn>::HintUpdateDuration(Node& node, uint dur)
+{
+    node.Weight = dur;
+}
+
 // Explicit instantiation
 template class EngineBase<Turn>;
 template class EngineBase<DefaultQueueableTurn<Turn>>;
