Shift async_driver gear into park if exception from workflow graph

Author: knoepfel

phlex/core/framework_graph.cpp

@@ -82,9 +82,11 @@ namespace phlex::experimental {
     finalize();
     run();
   } catch (std::exception const& e) {
+    driver_.stop();
     spdlog::error(e.what());
     throw;
   } catch (...) {
+    driver_.stop();
     spdlog::error("Unknown exception during graph execution");
     throw;
   }

phlex/core/fwd.hpp

@@ -2,7 +2,6 @@
 #define PHLEX_CORE_FWD_HPP
 
 #include "phlex/model/fwd.hpp"
-#include "phlex/utilities/async_driver.hpp"
 
 #include <memory>
 
@@ -20,10 +19,6 @@ namespace phlex::experimental {
   using end_of_message_ptr = std::shared_ptr<end_of_message>;
 }
 
-namespace phlex {
-  using framework_driver = experimental::async_driver<data_cell_index_ptr>;
-}
-
 #endif // PHLEX_CORE_FWD_HPP
 
 // Local Variables:

phlex/driver.hpp

@@ -6,10 +6,15 @@
 #include "phlex/configuration.hpp"
 #include "phlex/core/fwd.hpp"
 #include "phlex/model/product_store.hpp"
+#include "phlex/utilities/async_driver.hpp"
 
 #include <concepts>
 #include <memory>
 
+namespace phlex {
+  using framework_driver = experimental::async_driver<data_cell_index_ptr>;
+}
+
 namespace phlex::experimental::detail {
 
   // See note below.

phlex/utilities/async_driver.hpp

@@ -50,12 +50,19 @@ namespace phlex::experimental {
       return std::exchange(current_, std::nullopt);
     }
 
+    void stop()
+    {
+      // API that should only be called by the framework_graph
+      gear_ = states::park;
+      cv_.notify_one();
+    }
+
     void yield(RT rt)
     {
       std::unique_lock lock{mutex_};
       current_ = std::make_optional(std::move(rt));
       cv_.notify_one();
-      cv_.wait(lock);
+      cv_.wait(lock, [&] { return !current_.has_value() or gear_ == states::park; });
     }
 
   private: