Add unit test for observer

clin99 · clin99 · commit c3ee85a1f076 · 2019-04-18T13:25:10.000-05:00
diff --git a/taskflow/executor/observer.hpp b/taskflow/executor/observer.hpp
@@ -85,6 +85,13 @@ class ExecutorObserver : public ExecutorObserverInterface {
     */
     inline void clear();
 
+
+    /**
+    @brief get the number of total tasks in the observer
+    @return number of total tasks
+    */
+    inline int num_tasks() const;
+
   private:
     
     inline void set_up(unsigned num_workers) override final;
@@ -165,6 +172,15 @@ inline std::string ExecutorObserver::dump() const {
   return oss.str();
 }
 
+// Function: num_tasks
+inline int ExecutorObserver::num_tasks() const {
+  return std::accumulate(_begs.begin(), _begs.end(), 0, 
+    [](int sum, const std::vector<std::chrono::time_point<std::chrono::steady_clock>>& vec){ 
+      return sum + vec.size(); 
+    }
+  );
+}
+
 
 }  // end of namespace tf -------------------------------------------
 
diff --git a/taskflow/executor/proactive.hpp b/taskflow/executor/proactive.hpp
@@ -225,8 +225,15 @@ void ProactiveExecutor<Closure>::_spawn(unsigned N) {
           // shutdown cannot have task
           if(w.task) {
             lock.unlock();
+            if(_observer) {
+              _observer->on_entry(me);
+            }
             (*w.task)();
             w.task = std::nullopt;
+
+            if(_observer) {
+              _observer->on_exit(me);
+            }
             lock.lock();
           }
         }
diff --git a/unittest/taskflow.cpp b/unittest/taskflow.cpp
@@ -1190,12 +1190,100 @@ TEST_CASE("Composition-3" * doctest::timeout(300)) {
 }
 
 
+// --------------------------------------------------------
+// Testcase: Observer 
+// -------------------------------------------------------- 
+template <typename T> 
+void ObserverTest() {
+  for(unsigned w=0; w<=8; ++w) {
+    T taskflow {w};
+    auto observer = taskflow.share_executor()->template make_observer<tf::ExecutorObserver>();    
+
+    tf::Framework frameworkA;
+    std::vector<tf::Task> tasks;
+    // Static tasking 
+    for(auto i=0; i < 1024; i ++) {
+      tasks.emplace_back(frameworkA.emplace([](){}));
+    }
 
+    // Randomly specify dependency
+    for(auto i=0; i < 1024; i ++) {
+      for(auto j=i+1; j < 1024; j++) {
+        if(rand()%2 == 0) {
+          tasks[i].precede(tasks[j]);
+        }
+      }
+    }
 
+    taskflow.run_n(frameworkA, 16).get();
+
+    if(w == 0) {
+      REQUIRE(observer->num_tasks() == 0);
+    }
+    else {
+      REQUIRE(observer->num_tasks() == 1024*16);
+    }
+
+    observer->clear();
+    REQUIRE(observer->num_tasks() == 0);
+    tasks.clear();
+
+    // Dynamic tasking  
+    tf::Framework frameworkB;
+    std::atomic<int> num_tasks {0};
+    // Static tasking 
+    for(auto i=0; i < 1024; i ++) {
+      tasks.emplace_back(frameworkB.emplace([&](auto &subflow){
+        num_tasks ++;
+        auto num_spawn = rand() % 10 + 1;
+        // Randomly spawn tasks
+        for(auto i=0; i<num_spawn; i++) {
+          subflow.emplace([&](){ num_tasks ++; });
+        }    
+        if(rand() % 2) {
+          subflow.detach();
+        }
+        else {
+          // In join mode, this task will be visited twice
+          num_tasks ++;
+        }
+      }));
+    }
 
+    // Randomly specify dependency
+    for(auto i=0; i < 1024; i ++) {
+      for(auto j=i+1; j < 1024; j++) {
+        if(rand()%2 == 0) {
+          tasks[i].precede(tasks[j]);
+        }
+      }
+    }
 
+    taskflow.run_n(frameworkB, 16).get();
 
+    if(w == 0) {
+      REQUIRE(observer->num_tasks() == 0);
+    }
+    else {
+      REQUIRE(observer->num_tasks() == num_tasks);
+    }
+  }
+}
 
+TEST_CASE("Observer" * doctest::timeout(300)) {
+  SUBCASE("Simple Executor") {
+    ObserverTest<tf::BasicTaskflow<tf::SimpleExecutor>>();
+  }
+  SUBCASE("Proactive Executor") {
+    ObserverTest<tf::BasicTaskflow<tf::ProactiveExecutor>>();
+  }
+  SUBCASE("Speculative Executor") {
+    ObserverTest<tf::BasicTaskflow<tf::SpeculativeExecutor>>();
+  }
+  SUBCASE("WorkStealing Executor") {
+    ObserverTest<tf::BasicTaskflow<tf::WorkStealingExecutor>>();
+  }
+}
 
 
 
