Allow drivers to access configured sources (#658)

* Support updating the hierarchy
* Adjust framework_graph construction to allow deferred setting of the driver
* Allow drivers to access sources
* Rename g.source<T> to g.add_source<T>
* Adjustments to driver(...) overloads

Author: knoepfel

form/form_source.cpp

@@ -191,7 +191,7 @@ PHLEX_REGISTER_SOURCE(s, config)
   }
 
   // Register the source object with Phlex
-  s.source<FormInputSource>(
+  s.add_source<FormInputSource>(
     module_label, input_cfg, tech_cfg, actual_creator, advertised_creator, products);
 
   std::cout << "FORM input source registered successfully\n";

phlex/app/load_module.cpp

@@ -113,16 +113,17 @@ namespace phlex::experimental {
     creator(g.source_proxy(config), config);
   }
 
-  driver_bundle load_driver(boost::json::object const& raw_config)
+  void load_driver(framework_graph& g, boost::json::object const& raw_config)
   {
     configuration const config{raw_config};
     auto const& spec = config.get<std::string>("cpp");
+    auto const required_sources = config.get<std::vector<std::string>>("uses_sources", {});
     // False positive: clang-analyzer cannot trace ownership through Boost's is_any_of<char>
     // internal reference counting in classification.hpp.
     // NOLINTNEXTLINE(clang-analyzer-cplusplus.NewDeleteLeaks,clang-analyzer-cplusplus.NewDelete)
     create_driver = plugin_loader<detail::driver_shim_t>(spec, "create_driver");
     driver_bundle result;
-    create_driver(driver_proxy{}, config, &result);
-    return result;
+    create_driver(g.driver_proxy(required_sources), config, &result);
+    g.add_driver(result);
   }
 }

phlex/app/load_module.hpp

@@ -22,7 +22,7 @@ namespace phlex::experimental {
   RUN_PHLEX_EXPORT void load_source(framework_graph& g,
                                     std::string const& label,
                                     boost::json::object config);
-  RUN_PHLEX_EXPORT driver_bundle load_driver(boost::json::object const& config);
+  RUN_PHLEX_EXPORT void load_driver(framework_graph& g, boost::json::object const& config);
 }
 
 #endif // PHLEX_APP_LOAD_MODULE_HPP

phlex/app/run.cpp

@@ -17,8 +17,7 @@ namespace {
 namespace phlex::experimental {
   void run(boost::json::object const& configurations, int const max_parallelism)
   {
-    auto const driver_config = object_decorate_exception(configurations, "driver");
-    framework_graph g{load_driver(driver_config), max_parallelism};
+    auto g = framework_graph::without_driver(max_parallelism);
 
     // It is allowed for users to not specify any modules
     boost::json::object module_configs;
@@ -38,6 +37,10 @@ namespace phlex::experimental {
     for (auto const& [key, value] : source_configs) {
       load_source(g, key, value.as_object());
     }
+
+    auto const driver_config = object_decorate_exception(configurations, "driver");
+    load_driver(g, driver_config);
+
     g.execute();
   }
 }

phlex/core/framework_graph.cpp

@@ -15,24 +15,23 @@
 #include <iostream>
 
 namespace phlex::experimental {
-  framework_graph::framework_graph(int const max_parallelism) :
-    framework_graph{[](framework_driver& driver) { driver.yield(data_cell_index::job()); },
-                    max_parallelism}
+  framework_graph framework_graph::with_default_driver(int const max_parallelism)
   {
+    return framework_graph{driver_mode::default_driver, max_parallelism};
   }
 
-  framework_graph::framework_graph(detail::next_index_t next_index, int const max_parallelism) :
-    framework_graph{driver_bundle{std::move(next_index), {}}, max_parallelism}
+  framework_graph framework_graph::without_driver(int const max_parallelism)
   {
+    return framework_graph{driver_mode::deferred_driver, max_parallelism};
   }
 
-  framework_graph::framework_graph(driver_bundle bundle, int const max_parallelism) :
+  // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
+  framework_graph::framework_graph(driver_mode const mode, int const max_parallelism) :
     parallelism_limit_{static_cast<std::size_t>(max_parallelism)},
-    fixed_hierarchy_{std::move(bundle.hierarchy)},
-    driver_{std::move(bundle.driver)},
     src_{graph_,
          [this](tbb::flow_control& fc) mutable -> ready_flushes_then_emit {
-           if (auto item = driver_()) {
+           assert(driver_);
+           if (auto item = (*driver_)()) {
              return {.ready_flushes = cell_tracker_.report_and_evict_ready_flushes(*item),
                      .index_to_emit = *item};
            }
@@ -51,12 +50,33 @@ namespace phlex::experimental {
                     [this](data_cell_index_ptr const& index) -> tbb::flow::continue_msg {
                       hierarchy_.increment_count(index);
                       return {};
-                    }}
+                    }},
+    driver_mode_{mode}
   {
+    if (driver_mode_ == driver_mode::default_driver) {
+      driver_.emplace([](framework_driver& driver) { driver.yield(data_cell_index::job()); });
+    }
+
     spdlog::cfg::load_env_levels();
     spdlog::info("Number of worker threads: {}", max_allowed_parallelism::active_value());
   }
 
+  void framework_graph::add_driver(driver_bundle bundle)
+  {
+    if (driver_mode_ != driver_mode::deferred_driver) {
+      throw std::runtime_error(
+        "Cannot configure framework_graph with a driver when not in deferred mode.");
+    }
+    if (driver_) {
+      throw std::runtime_error("Driver has already been configured for framework_graph.");
+    }
+    if (!bundle.driver) {
+      throw std::runtime_error("Cannot configure framework_graph with an empty driver.");
+    }
+    fixed_hierarchy_ = std::move(bundle.hierarchy);
+    driver_.emplace(std::move(bundle.driver));
+  }
+
   framework_graph::~framework_graph()
   {
     if (shutdown_on_error_) {
@@ -79,19 +99,26 @@ namespace phlex::experimental {
   }
 
   void framework_graph::execute()
-  try {
-    finalize();
-    run();
-  } catch (std::exception const& e) {
-    driver_.stop();
-    spdlog::error(e.what());
-    shutdown_on_error_ = true;
-    throw;
-  } catch (...) {
-    driver_.stop();
-    spdlog::error("Unknown exception during graph execution");
-    shutdown_on_error_ = true;
-    throw;
+  {
+    if (!driver_) {
+      throw std::runtime_error("No driver configured for framework_graph.");
+    }
+
+    try {
+      finalize();
+      run();
+    } catch (std::exception const& e) {
+      driver_->stop();
+
+      spdlog::error(e.what());
+      shutdown_on_error_ = true;
+      throw;
+    } catch (...) {
+      driver_->stop();
+      spdlog::error("Unknown exception during graph execution");
+      shutdown_on_error_ = true;
+      throw;
+    }
   }
 
   void framework_graph::run()

phlex/core/framework_graph.hpp

@@ -26,6 +26,7 @@
 #include <functional>
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <tuple>
 #include <utility>
@@ -38,17 +39,28 @@ namespace phlex {
 namespace phlex::experimental {
   class PHLEX_CORE_EXPORT framework_graph {
   public:
-    explicit framework_graph(int max_parallelism = oneapi::tbb::info::default_concurrency());
-    explicit framework_graph(detail::next_index_t next_index,
-                             int max_parallelism = oneapi::tbb::info::default_concurrency());
-    explicit framework_graph(driver_bundle bundle,
-                             int max_parallelism = oneapi::tbb::info::default_concurrency());
+    [[nodiscard]] static framework_graph with_default_driver(
+      int max_parallelism = oneapi::tbb::info::default_concurrency());
+    [[nodiscard]] static framework_graph without_driver(
+      int max_parallelism = oneapi::tbb::info::default_concurrency());
+
     ~framework_graph();
     framework_graph(framework_graph const&) = delete;
     framework_graph& operator=(framework_graph const&) = delete;
     framework_graph(framework_graph&&) = delete;
     framework_graph& operator=(framework_graph&&) = delete;
 
+    void add_driver(driver_bundle bundle);
+
+    template <typename Generator>
+      requires requires(std::shared_ptr<Generator> generator, std::vector<source const*> sources) {
+        { experimental::driver_proxy{sources}.driver(generator) } -> std::same_as<driver_bundle>;
+      }
+    void add_driver(std::shared_ptr<Generator> generator)
+    {
+      add_driver(driver_proxy().driver(std::move(generator)));
+    }
+
     void execute();
 
     std::size_t seen_cell_count(std::string const& layer_name, bool missing_ok = false) const;
@@ -64,6 +76,11 @@ namespace phlex::experimental {
       return {config, graph_, nodes_, registration_errors_};
     }
 
+    experimental::driver_proxy driver_proxy(std::vector<std::string> strings = {})
+    {
+      return experimental::driver_proxy(nodes_.sources_for(strings));
+    }
+
     // Framework function registrations
 
     // N.B. declare_output() is not directly accessible through framework_graph.  Is this
@@ -115,9 +132,9 @@ namespace phlex::experimental {
     }
 
     template <std::derived_from<source> Source, typename... Args>
-    void source(std::string name, Args&&... args)
+    void add_source(std::string name, Args&&... args)
     {
-      return make_glue().template source<Source>(std::move(name), std::forward<Args>(args)...);
+      return make_glue().template add_source<Source>(std::move(name), std::forward<Args>(args)...);
     }
 
     template <typename T, typename... Args>
@@ -170,6 +187,9 @@ namespace phlex::experimental {
     void finalize_router(index_router::provider_input_ports_t provider_input_ports,
                          std::map<std::string, named_index_ports> multilayer_join_index_ports);
 
+    enum class driver_mode { default_driver, deferred_driver };
+    explicit framework_graph(driver_mode mode, int max_parallelism);
+
     resource_usage graph_resource_usage_{};
     max_allowed_parallelism parallelism_limit_;
     fixed_hierarchy fixed_hierarchy_;
@@ -178,7 +198,7 @@ namespace phlex::experimental {
     std::map<std::string, filter> filters_{};
     // The graph_ object uses the filters_, nodes_, and hierarchy_ objects implicitly.
     tbb::flow::graph graph_{};
-    framework_driver driver_;
+    std::optional<framework_driver> driver_{};
     std::vector<std::string> registration_errors_{};
     data_cell_tracker cell_tracker_{};
     tbb::flow::input_node<ready_flushes_then_emit> src_;
@@ -187,6 +207,7 @@ namespace phlex::experimental {
       index_receiver_;
     tbb::flow::function_node<data_cell_index_ptr, tbb::flow::continue_msg, tbb::flow::lightweight>
       hierarchy_node_;
+    driver_mode driver_mode_{driver_mode::default_driver};
     bool shutdown_on_error_{false};
   };
 }

phlex/core/glue.hpp

@@ -152,7 +152,7 @@ namespace phlex::experimental {
     }
 
     template <std::derived_from<source> Source, typename... Args>
-    void source(std::string name, Args&&... args)
+    void add_source(std::string name, Args&&... args)
     {
       auto [_, inserted] =
         nodes_.sources.try_emplace(name, std::make_unique<Source>(std::forward<Args>(args)...));

phlex/core/graph_proxy.hpp

@@ -109,13 +109,13 @@ namespace phlex::experimental {
 
     /// @brief Registers a source (used by the framework to create provider nodes)
     template <std::derived_from<source> Source, typename... Args>
-    void source(std::string name, Args&&... args)
+    void add_source(std::string name, Args&&... args)
       requires(not is_bound_object<T>)
     {
       // The bound object is created when invoking source<Source>(...), so we explicitly indicate that
       // no bound object should be used in the create_glue(...) call.
-      return create_glue(false).template source<Source>(std::move(name),
-                                                        std::forward<Args>(args)...);
+      return create_glue(false).template add_source<Source>(std::move(name),
+                                                            std::forward<Args>(args)...);
     }
 
     /// @brief Registers an output node.

phlex/core/node_catalog.cpp

@@ -1,6 +1,9 @@
 #include "phlex/core/node_catalog.hpp"
 
+#include "fmt/format.h"
+
 #include <string>
+#include <vector>
 
 using namespace std::string_literals;
 
@@ -53,4 +56,18 @@ namespace phlex::experimental {
   {
     return producer_catalog{transforms, folds, unfolds};
   }
+
+  source_vector node_catalog::sources_for(std::vector<std::string> const& keys) const
+  {
+    source_vector result;
+    result.reserve(keys.size());
+    for (auto const& key : keys) {
+      if (auto src = sources.get(key)) {
+        result.push_back(src);
+      } else {
+        throw std::runtime_error(fmt::format("Unknown source with name: {}", key));
+      }
+    }
+    return result;
+  }
 }

phlex/core/node_catalog.hpp

@@ -28,6 +28,8 @@ namespace phlex::experimental {
       return registrar{ptr_map_for<Ptr>(), errors};
     }
 
+    source_vector sources_for(std::vector<std::string> const& keys) const;
+
     std::size_t execution_count(std::string const& node_name) const;
     std::vector<products_consumer*> consumers() const;
     producer_catalog producers() const;