Support drivers and sources in job-configuration schema

phlex/app/load_module.cpp

@@ -1,6 +1,7 @@
 #include "phlex/app/load_module.hpp"
 #include "phlex/configuration.hpp"
 #include "phlex/core/framework_graph.hpp"
+#include "phlex/driver.hpp"
 #include "phlex/module.hpp"
 #include "phlex/source.hpp"
 
@@ -19,7 +20,8 @@ namespace phlex::experimental {
     // If factory function goes out of scope, then the library is unloaded...and that's
     // bad.
     std::vector<std::function<detail::module_creator_t>> create_module;
-    std::function<detail::source_creator_t> create_source;
+    std::vector<std::function<detail::source_creator_t>> create_source;
+    std::function<detail::driver_creator_t> create_driver;
 
     template <typename creator_t>
     std::function<creator_t> plugin_loader(std::string const& spec, std::string const& symbol_name)
@@ -53,15 +55,29 @@ namespace phlex::experimental {
     raw_config["module_label"] = label;
 
     configuration const config{raw_config};
-    auto module_proxy = g.proxy(config);
-    creator(module_proxy, config);
+    creator(g.module_proxy(config), config);
   }
 
-  detail::next_index_t load_source(boost::json::object const& raw_config)
+  void load_source(framework_graph& g, std::string const& label, boost::json::object raw_config)
+  {
+    auto const& spec = value_to<std::string>(raw_config.at("plugin"));
+    auto& creator =
+      create_source.emplace_back(plugin_loader<detail::source_creator_t>(spec, "create_source"));
+
+    // FIXME: Should probably use the parameter name (e.g.) 'plugin_label' instead of
+    //        'module_label', but that requires adjusting other parts of the system
+    //        (e.g. make_algorithm_name).
+    raw_config["module_label"] = label;
+
+    configuration const config{raw_config};
+    creator(g.source_proxy(config), config);
+  }
+
+  detail::next_index_t load_driver(boost::json::object const& raw_config)
   {
     configuration const config{raw_config};
     auto const& spec = config.get<std::string>("plugin");
-    create_source = plugin_loader<detail::source_creator_t>(spec, "create_source");
-    return create_source(config);
+    create_driver = plugin_loader<detail::driver_creator_t>(spec, "create_driver");
+    return create_driver(config);
   }
 }

phlex/app/load_module.hpp

@@ -2,15 +2,16 @@
 #define PHLEX_APP_LOAD_MODULE_HPP
 
 #include "phlex/core/fwd.hpp"
-#include "phlex/source.hpp"
+#include "phlex/driver.hpp"
 
 #include "boost/json.hpp"
 
 #include <functional>
 
 namespace phlex::experimental {
   void load_module(framework_graph& g, std::string const& label, boost::json::object config);
-  detail::next_index_t load_source(boost::json::object const& config);
+  void load_source(framework_graph& g, std::string const& label, boost::json::object config);
+  detail::next_index_t load_driver(boost::json::object const& config);
 }
 
 #endif // PHLEX_APP_LOAD_MODULE_HPP

phlex/app/run.cpp

@@ -5,14 +5,39 @@
 
 using namespace std::string_literals;
 
+namespace {
+  auto object_decorate_exception(boost::json::object const& obj, std::string const& key)
+  try {
+    return obj.at(key).as_object();
+  } catch (std::exception const& e) {
+    throw std::runtime_error("Error retrieving parameter '" + key + "':\n" + e.what());
+  }
+}
+
 namespace phlex::experimental {
   void run(boost::json::object const& configurations, int const max_parallelism)
   {
-    framework_graph g{load_source(configurations.at("source").as_object()), max_parallelism};
-    auto const module_configs = configurations.at("modules").as_object();
+    auto const driver_config = object_decorate_exception(configurations, "driver");
+    framework_graph g{load_driver(driver_config), max_parallelism};
+
+    // It is allowed for users to not specify any modules
+    boost::json::object module_configs;
+    if (configurations.contains("modules")) {
+      module_configs = object_decorate_exception(configurations, "modules");
+    }
+
     for (auto const& [key, value] : module_configs) {
       load_module(g, key, value.as_object());
     }
+
+    boost::json::object source_configs;
+    if (configurations.contains("sources")) {
+      source_configs = object_decorate_exception(configurations, "sources");
+    }
+
+    for (auto const& [key, value] : source_configs) {
+      load_source(g, key, value.as_object());
+    }
     g.execute();
   }
 }

phlex/core/framework_graph.hpp

@@ -6,13 +6,14 @@
 #include "phlex/core/end_of_message.hpp"
 #include "phlex/core/filter.hpp"
 #include "phlex/core/glue.hpp"
-#include "phlex/core/graph_proxy.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/message_sender.hpp"
 #include "phlex/core/multiplexer.hpp"
 #include "phlex/core/node_catalog.hpp"
+#include "phlex/driver.hpp"
 #include "phlex/model/data_layer_hierarchy.hpp"
 #include "phlex/model/product_store.hpp"
+#include "phlex/module.hpp"
 #include "phlex/source.hpp"
 #include "phlex/utilities/max_allowed_parallelism.hpp"
 #include "phlex/utilities/resource_usage.hpp"
@@ -58,7 +59,12 @@ namespace phlex::experimental {
     std::size_t execution_counts(std::string const& node_name) const;
     std::size_t product_counts(std::string const& node_name) const;
 
-    graph_proxy<void_tag> proxy(configuration const& config)
+    module_graph_proxy<void_tag> module_proxy(configuration const& config)
+    {
+      return {config, graph_, nodes_, registration_errors_};
+    }
+
+    source_graph_proxy<void_tag> source_proxy(configuration const& config)
     {
       return {config, graph_, nodes_, registration_errors_};
     }

phlex/detail/plugin_macros.hpp

@@ -0,0 +1,27 @@
+#ifndef PHLEX_DETAIL_PLUGIN_MACROS_HPP
+#define PHLEX_DETAIL_PLUGIN_MACROS_HPP
+
+#include "boost/preprocessor.hpp"
+
+#define PHLEX_DETAIL_NARGS(...) BOOST_PP_DEC(BOOST_PP_VARIADIC_SIZE(__VA_OPT__(, ) __VA_ARGS__))
+
+#define PHLEX_DETAIL_CREATE_1ARG(token_type, func_name, m)                                         \
+  void func_name(token_type<phlex::experimental::void_tag>& m,                                     \
+                 phlex::experimental::configuration const&)
+
+#define PHLEX_DETAIL_CREATE_2ARGS(token_type, func_name, m, pset)                                  \
+  void func_name(token_type<phlex::experimental::void_tag>& m,                                     \
+                 phlex::experimental::configuration const& config)
+
+#define PHLEX_DETAIL_SELECT_SIGNATURE(token_type, func_name, ...)                                  \
+  BOOST_PP_IF(BOOST_PP_EQUAL(PHLEX_DETAIL_NARGS(__VA_ARGS__), 1),                                  \
+              PHLEX_DETAIL_CREATE_1ARG,                                                            \
+              PHLEX_DETAIL_CREATE_2ARGS)                                                           \
+  (token_type, func_name, __VA_ARGS__)
+
+#define PHLEX_DETAIL_REGISTER_PLUGIN(token_type, func_name, dll_alias, ...)                        \
+  static PHLEX_DETAIL_SELECT_SIGNATURE(token_type, func_name, __VA_ARGS__);                        \
+  BOOST_DLL_ALIAS(func_name, dll_alias)                                                            \
+  PHLEX_DETAIL_SELECT_SIGNATURE(token_type, func_name, __VA_ARGS__)
+
+#endif // PHLEX_DETAIL_PLUGIN_MACROS_HPP

phlex/driver.hpp

@@ -0,0 +1,69 @@
+#ifndef PHLEX_DRIVER_HPP
+#define PHLEX_DRIVER_HPP
+
+#include "boost/dll/alias.hpp"
+
+#include "phlex/configuration.hpp"
+#include "phlex/core/fwd.hpp"
+#include "phlex/model/product_store.hpp"
+
+#include <concepts>
+#include <memory>
+
+namespace phlex::experimental::detail {
+
+  // See note below.
+  template <typename T>
+  auto make(configuration const& config)
+  {
+    if constexpr (requires { T{config}; }) {
+      return std::make_shared<T>(config);
+    } else {
+      return std::make_shared<T>();
+    }
+  }
+
+  template <typename T>
+  concept next_function_with_driver = requires(T t, framework_driver& driver) {
+    { t.next(driver) } -> std::same_as<void>;
+  };
+
+  template <typename T>
+  concept next_function_without_driver = requires(T t) {
+    { t.next() } -> std::same_as<void>;
+  };
+
+  // Workaround for static_assert(false) until P2593R1 is adopted
+  //   https://www.open-std.org/jtc1/sc22/wg21/docs/papers/2023/p2593r1.html
+  // static_assert(false) is supported in GCC 13 and newer
+  template <typename T>
+  constexpr bool always_false{false};
+
+  template <typename T>
+  std::function<void(framework_driver&)> create_next(configuration const& config = {})
+  {
+    // N.B. Because we are initializing an std::function object with a lambda, the lambda
+    //      (and therefore its captured values) must be copy-constructible.  This means
+    //      that make<T>(config) must return a copy-constructible object.  Because we do not
+    //      know if a user's provided driver class is copyable, we create the object on
+    //      the heap, and capture a shared pointer to the object.  This also ensures that
+    //      the driver object is created only once, thus avoiding potential errors in the
+    //      implementations of the driver class' copy/move constructors (e.g. if the
+    //      source is caching an iterator).
+    if constexpr (next_function_with_driver<T>) {
+      return [t = make<T>(config)](framework_driver& driver) { t->next(driver); };
+    } else if constexpr (next_function_without_driver<T>) {
+      return [t = make<T>(config)](framework_driver&) { t->next(); };
+    } else {
+      static_assert(always_false<T>, "Must have a 'next()' function that returns 'void'");
+    }
+  }
+
+  using next_index_t = std::function<void(framework_driver&)>;
+  using driver_creator_t = next_index_t(configuration const&);
+}
+
+#define PHLEX_EXPERIMENTAL_REGISTER_DRIVER(driver)                                                 \
+  BOOST_DLL_ALIAS(phlex::experimental::detail::create_next<driver>, create_driver)
+
+#endif // PHLEX_DRIVER_HPP

phlex/module.hpp

@@ -5,28 +5,34 @@
 #include "phlex/concurrency.hpp"
 #include "phlex/configuration.hpp"
 #include "phlex/core/graph_proxy.hpp"
-
-#include "boost/preprocessor.hpp"
-
-namespace phlex::experimental::detail {
-  using module_creator_t = void(graph_proxy<void_tag>&, configuration const&);
+#include "phlex/detail/plugin_macros.hpp"
+
+namespace phlex::experimental {
+  template <typename T>
+  class module_graph_proxy : graph_proxy<T> {
+    using base = graph_proxy<T>;
+
+  public:
+    using base::graph_proxy;
+
+    // FIXME: make sure functions called from make<T>(...) are restricted to the functions below:
+    //        Users can call make<T>(...).fold(...) but not make<T>(...).provide(...)
+    using base::make;
+
+    using base::fold;
+    using base::observe;
+    using base::predicate;
+    using base::transform;
+    using base::unfold;
+  };
+
+  namespace detail {
+    using module_creator_t = void(module_graph_proxy<void_tag>, configuration const&);
+  }
 }
 
-#define NARGS(...) BOOST_PP_DEC(BOOST_PP_VARIADIC_SIZE(__VA_OPT__(, ) __VA_ARGS__))
-
-#define CREATE_1ARG(m)                                                                             \
-  void create(phlex::experimental::graph_proxy<phlex::experimental::void_tag>& m,                  \
-              phlex::experimental::configuration const&)
-#define CREATE_2ARGS(m, pset)                                                                      \
-  void create(phlex::experimental::graph_proxy<phlex::experimental::void_tag>& m,                  \
-              phlex::experimental::configuration const& config)
-
-#define SELECT_SIGNATURE(...)                                                                      \
-  BOOST_PP_IF(BOOST_PP_EQUAL(NARGS(__VA_ARGS__), 1), CREATE_1ARG, CREATE_2ARGS)(__VA_ARGS__)
-
 #define PHLEX_EXPERIMENTAL_REGISTER_ALGORITHMS(...)                                                \
-  static SELECT_SIGNATURE(__VA_ARGS__);                                                            \
-  BOOST_DLL_ALIAS(create, create_module)                                                           \
-  SELECT_SIGNATURE(__VA_ARGS__)
+  PHLEX_DETAIL_REGISTER_PLUGIN(                                                                    \
+    phlex::experimental::module_graph_proxy, create, create_module, __VA_ARGS__)
 
 #endif // PHLEX_MODULE_HPP

phlex/source.hpp

@@ -2,68 +2,34 @@
 #define PHLEX_SOURCE_HPP
 
 #include "boost/dll/alias.hpp"
-
+#include "phlex/concurrency.hpp"
 #include "phlex/configuration.hpp"
-#include "phlex/core/fwd.hpp"
-#include "phlex/model/product_store.hpp"
-
-#include <concepts>
-#include <memory>
-
-namespace phlex::experimental::detail {
+#include "phlex/core/graph_proxy.hpp"
+#include "phlex/detail/plugin_macros.hpp"
 
-  // See note below.
+namespace phlex::experimental {
   template <typename T>
-  auto make(configuration const& config)
-  {
-    if constexpr (requires { T{config}; }) {
-      return std::make_shared<T>(config);
-    } else {
-      return std::make_shared<T>();
-    }
-  }
+  class source_graph_proxy : graph_proxy<T> {
+    using base = graph_proxy<T>;
 
-  template <typename T>
-  concept next_function_with_driver = requires(T t, framework_driver& driver) {
-    { t.next(driver) } -> std::same_as<void>;
-  };
+  public:
+    using base::graph_proxy;
 
-  template <typename T>
-  concept next_function_without_driver = requires(T t) {
-    { t.next() } -> std::same_as<void>;
-  };
+    // FIXME: make sure functions called from make<T>(...) are restricted to the functions below:
+    //        Users can call make<T>(...).provide(...) but not make<T>(...).fold(...)
+    using base::make;
 
-  // Workaround for static_assert(false) until P2593R1 is adopted
-  //   https://www.open-std.org/jtc1/sc22/wg21/docs/papers/2023/p2593r1.html
-  // static_assert(false) is supported in GCC 13 and newer
-  template <typename T>
-  constexpr bool always_false{false};
+    // Only provide(...) should be accessible
+    using base::provide;
+  };
 
-  template <typename T>
-  std::function<void(framework_driver&)> create_next(configuration const& config = {})
-  {
-    // N.B. Because we are initializing an std::function object with a lambda, the lambda
-    //      (and therefore its captured values) must be copy-constructible.  This means
-    //      that make<T>(config) must return a copy-constructible object.  Because we do not
-    //      know if a user's provided source class is copyable, we create the object on
-    //      the heap, and capture a shared pointer to the object.  This also ensures that
-    //      the source object is created only once, thus avoiding potential errors in the
-    //      implementations of the source class' copy/move constructors (e.g. if the
-    //      source is caching an iterator).
-    if constexpr (next_function_with_driver<T>) {
-      return [t = make<T>(config)](framework_driver& driver) { t->next(driver); };
-    } else if constexpr (next_function_without_driver<T>) {
-      return [t = make<T>(config)](framework_driver&) { t->next(); };
-    } else {
-      static_assert(always_false<T>, "Must have a 'next()' function that returns 'void'");
-    }
+  namespace detail {
+    using source_creator_t = void(source_graph_proxy<void_tag>, configuration const&);
   }
-
-  using next_index_t = std::function<void(framework_driver&)>;
-  using source_creator_t = next_index_t(configuration const&);
 }
 
-#define PHLEX_EXPERIMENTAL_REGISTER_SOURCE(source)                                                 \
-  BOOST_DLL_ALIAS(phlex::experimental::detail::create_next<source>, create_source)
+#define PHLEX_EXPERIMENTAL_REGISTER_PROVIDERS(...)                                                 \
+  PHLEX_DETAIL_REGISTER_PLUGIN(                                                                    \
+    phlex::experimental::source_graph_proxy, create, create_source, __VA_ARGS__)
 
 #endif // PHLEX_SOURCE_HPP

plugins/generate_layers.cpp

@@ -16,7 +16,7 @@
 // Note that 'total' refers to the total number of data cells *per* parent.
 // ==============================================================================================
 
-#include "phlex/source.hpp"
+#include "phlex/driver.hpp"
 #include "plugins/layer_generator.hpp"
 
 #include "phlex/core/framework_graph.hpp"
@@ -52,4 +52,4 @@ namespace {
   };
 }
 
-PHLEX_EXPERIMENTAL_REGISTER_SOURCE(generate_layers)
+PHLEX_EXPERIMENTAL_REGISTER_DRIVER(generate_layers)