fixed subflow exception bug

TSUNG-WEI HUANG · TSUNG-WEI HUANG · commit 5218dfb9cdda · 2024-12-13T19:26:23.000-06:00
diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt
@@ -36,6 +36,7 @@ list(APPEND TF_EXAMPLES
   limited_concurrency
   cancel
   exception
+  subflow_exception
 )
 
 foreach(example IN LISTS TF_EXAMPLES)
diff --git a/examples/subflow_exception.cpp b/examples/subflow_exception.cpp
@@ -0,0 +1,27 @@
+// This program demonstrates the exception in subflow.
+
+#include <taskflow/taskflow.hpp>    
+
+int main() {
+
+  tf::Executor executor;
+  tf::Taskflow taskflow;
+
+  taskflow.emplace([](tf::Subflow& sf) {
+    tf::Task A = sf.emplace([]() { throw std::runtime_error("exception on A"); });
+    tf::Task B = sf.emplace([]() { std::cout << "Task B\n"; });
+    A.precede(B);
+    sf.join();
+  });
+
+  try
+  {
+    executor.run(taskflow).get();
+  }
+  catch (const std::runtime_error& re)
+  {
+    std::cout << re.what() << std::endl;
+  }
+
+  return 0;
+}
diff --git a/taskflow/core/executor.hpp b/taskflow/core/executor.hpp
@@ -1711,15 +1711,16 @@ inline void Executor::_observer_epilogue(Worker& worker, Node* node) {
 // Procedure: _process_exception
 inline void Executor::_process_exception(Worker&, Node* node) {
 
-  constexpr static auto flag = Topology::EXCEPTION | Topology::CANCELLED;
+  constexpr static auto nflag = Node::EXCEPTION | Node::CANCELLED;
+  constexpr static auto tflag = Topology::EXCEPTION | Topology::CANCELLED;
 
   //std::cout << "processing exception from " << node->_name << std::endl;
 
   // the exception occurs under a blocking corun
   if(auto anchor = node->_anchor(); anchor) {
     //std::cout << "\tfind anchor: " << anchor->_name << '\n';
     // multiple tasks may throw, and we only take the first thrown exception
-    if((anchor->_state.fetch_or(Node::EXCEPTION, std::memory_order_relaxed) & Node::EXCEPTION) == 0) {
+    if((anchor->_state.fetch_or(nflag, std::memory_order_relaxed) & Node::EXCEPTION) == 0) {
       //std::cout << node->_name << " stores exception in anchor " << anchor->_name << std::endl;
       anchor->_exception_ptr = std::current_exception();
     }
@@ -1728,7 +1729,7 @@ inline void Executor::_process_exception(Worker&, Node* node) {
   else if(auto tpg = node->_topology; tpg) {
     //std::cout << "\tno anchor - go to topology\n";
     // multiple tasks may throw, and we only take the first thrown exception
-    if((tpg->_state.fetch_or(flag, std::memory_order_relaxed) & Topology::EXCEPTION) == 0) {
+    if((tpg->_state.fetch_or(tflag, std::memory_order_relaxed) & Topology::EXCEPTION) == 0) {
       //std::cout << "store exception in topology " << std::endl;
       tpg->_exception_ptr = std::current_exception();
     }
@@ -2325,7 +2326,6 @@ void Runtime::acquire(S&&... semaphores) {
   _executor._corun_until(_worker, [&](){ 
     return tf::try_acquire(std::forward<S>(semaphores)...); 
   });
-  // TODO: exception?
 }
   
 // Function:: acquire
@@ -2336,7 +2336,6 @@ void Runtime::acquire(I first, I last) {
   _executor._corun_until(_worker, [=](){ 
     return tf::try_acquire(first, last); 
   });
-  // TODO: exception?
 }
 
 // Function: release
diff --git a/taskflow/core/graph.hpp b/taskflow/core/graph.hpp
@@ -651,6 +651,7 @@ class Node {
   constexpr static int EXCEPTION   = 4;
   constexpr static int PREEMPTED   = 8;
   constexpr static int ANCHOR      = 16;
+  constexpr static int CANCELLED   = 32;
 
   using Placeholder = std::monostate;
 
@@ -1052,8 +1053,8 @@ inline bool Node::_is_preempted() const {
 // Function: _is_cancelled
 // we currently only support cancellation of taskflow (no async task)
 inline bool Node::_is_cancelled() const {
-  return _topology &&
-         (_topology->_state.load(std::memory_order_relaxed) & Topology::CANCELLED);
+  return (_topology && (_topology->_state.load(std::memory_order_relaxed) & Topology::CANCELLED)) ||
+         (_parent && (_parent->_state.load(std::memory_order_relaxed) & Node::CANCELLED));
 }
 
 // Procedure: _set_up_join_counter
diff --git a/unittests/test_exceptions.cpp b/unittests/test_exceptions.cpp

Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,7 @@ list(APPEND TF_EXAMPLES`
`36`	`36`	`limited_concurrency`
`37`	`37`	`cancel`
`38`	`38`	`exception`
	`39`	`+ subflow_exception`
`39`	`40`	`)`
`40`	`41`
`41`	`42`	`foreach(example IN LISTS TF_EXAMPLES)`