From 755705ea80b1d07b1924265a14568d3af3c81a8d Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Fri, 9 Oct 2026 10:26:51 -0500 Subject: [PATCH] refactor: remove unnecessary layer in consumer hierarchy Aided by GPT-6.1 Sol --- phlex/core/CMakeLists.txt | 2 - phlex/core/consumer.cpp | 96 +++++++++++++++++++-- phlex/core/consumer.hpp | 41 ++++++++- phlex/core/declared_fold.cpp | 12 +-- phlex/core/declared_fold.hpp | 6 +- phlex/core/declared_observer.cpp | 12 +-- phlex/core/declared_observer.hpp | 4 +- phlex/core/declared_output.cpp | 11 ++- phlex/core/declared_output.hpp | 9 +- phlex/core/declared_predicate.cpp | 12 +-- phlex/core/declared_predicate.hpp | 4 +- phlex/core/declared_transform.cpp | 12 +-- phlex/core/declared_transform.hpp | 6 +- phlex/core/declared_unfold.cpp | 12 +-- phlex/core/declared_unfold.hpp | 6 +- phlex/core/filter.cpp | 4 +- phlex/core/filter.hpp | 2 +- phlex/core/fold_node.hpp | 2 +- phlex/core/fwd.hpp | 1 - phlex/core/make_computational_edges.cpp | 7 +- phlex/core/node_catalog.cpp | 9 +- phlex/core/node_catalog.hpp | 4 +- phlex/core/observer_node.hpp | 2 +- phlex/core/predicate_node.hpp | 2 +- phlex/core/products_consumer.cpp | 110 ------------------------ phlex/core/products_consumer.hpp | 58 ------------- phlex/core/transform_node.hpp | 2 +- phlex/core/unfold_node.hpp | 2 +- test/core_misc_test.cpp | 13 --- 29 files changed, 202 insertions(+), 261 deletions(-) delete mode 100644 phlex/core/products_consumer.cpp delete mode 100644 phlex/core/products_consumer.hpp diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt index 0496d0e4e..37d37dc75 100644 --- a/phlex/core/CMakeLists.txt +++ b/phlex/core/CMakeLists.txt @@ -24,7 +24,6 @@ cet_make_library( node_catalog.cpp producer_catalog.cpp product_selector.cpp - products_consumer.cpp provider_node.cpp registrar.cpp registration_api.cpp @@ -69,7 +68,6 @@ install( observer_node.hpp predicate_node.hpp product_selector.hpp - products_consumer.hpp provider_node.hpp registrar.hpp registration_api.hpp diff --git a/phlex/core/consumer.cpp b/phlex/core/consumer.cpp index 9d1cd3764..0e80d64c1 100644 --- a/phlex/core/consumer.cpp +++ b/phlex/core/consumer.cpp @@ -1,28 +1,108 @@ #include "phlex/core/consumer.hpp" +#include "phlex/core/message.hpp" +#include "phlex/core/product_selector.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/identifier.hpp" +#include "phlex/utilities/bulleted_list.hpp" +#include +#include + +#include +#include +#include #include #include #include +namespace { + std::vector layers_from(phlex::product_selectors const& queries) + { + using namespace phlex::experimental::literals; + std::vector result; + result.reserve(queries.size()); + for (auto const& query : queries) { + if (query.layer) { + result.push_back(query.layer); + } else { + result.push_back("*"_id); + } + } + return result; + } + + void validate_layers(phlex::detail::require_layers layers_required, + phlex::product_selectors const& inputs, + phlex::experimental::algorithm_name const& algo) + { + using namespace phlex::detail; + if (layers_required == require_layers::multi_input_only && inputs.size() <= 1) { + return; + } + std::vector err_selectors{}; + for (auto const& p : inputs) { + if (!p.layer) { + err_selectors.push_back(p.to_string()); + } + } + if (!err_selectors.empty()) { + std::string const error = + fmt::format("Must specify layers in the product selectors for node {}:\n" + " (Only invalid selectors are listed)\n{}", + algo.to_string(), + bulleted_list(err_selectors)); + throw std::runtime_error(error); + } + } +} + namespace phlex::detail { consumer::consumer(phlex::experimental::algorithm_name name, - std::vector predicates) : - name_{std::move(name)}, predicates_{std::move(predicates)} + std::vector predicates, + product_selectors input_products, + tbb::flow::graph& graph, + require_layers layers_required) : + name_{std::move(name)}, + predicates_{std::move(predicates)}, + graph_{graph}, + input_products_{std::move(input_products)}, + layers_{layers_from(input_products_)} { + validate_layers(layers_required, input_products_, this->name()); } + consumer::~consumer() = default; + phlex::experimental::algorithm_name const& consumer::name() const noexcept { return name_; } - phlex::experimental::identifier const& consumer::plugin() const noexcept + + std::vector const& consumer::when() const noexcept { return predicates_; } + + tbb::flow::receiver& consumer::port(product_selector const& input_product) { - return name_.plugin(); + auto& next = port_for(input_product); + + // If input_product doesn't have a layer, it must be for a node that allows layer omission + if (input_product.layer) { + auto& layer_check = layer_checkers_.emplace_back(std::make_unique( + graph_, + tbb::flow::unlimited, + [&layer = static_cast(input_product.layer)]( + message const& msg, auto& output) { + if (msg.store->layer_name() == layer) { + std::get<0>(output).try_put(msg); + } + })); + make_edge(tbb::flow::output_port<0>(*layer_check), next); + return *layer_check; + } + // else + return next; } - phlex::experimental::identifier const& consumer::algorithm() const noexcept + + product_selectors const& consumer::input() const noexcept { return input_products_; } + std::vector const& consumer::layers() const noexcept { - return name_.algorithm(); + return layers_; } - - std::vector const& consumer::when() const noexcept { return predicates_; } } diff --git a/phlex/core/consumer.hpp b/phlex/core/consumer.hpp index aa9e328cc..0694ea7d9 100644 --- a/phlex/core/consumer.hpp +++ b/phlex/core/consumer.hpp @@ -1,25 +1,60 @@ #ifndef PHLEX_CORE_CONSUMER_HPP #define PHLEX_CORE_CONSUMER_HPP +#include "phlex/core/fwd.hpp" +#include "phlex/core/input_arguments.hpp" +#include "phlex/core/message.hpp" +#include "phlex/core/product_selector.hpp" #include "phlex/model/algorithm_name.hpp" +#include "phlex/model/identifier.hpp" #include "phlex/phlex_core_export.hpp" +#include + +#include +#include +#include #include #include namespace phlex::detail { + enum class require_layers : char { multi_input_only, always }; class PHLEX_CORE_EXPORT consumer { + using layer_check_node_t = tbb::flow::multifunction_node>; + public: - consumer(phlex::experimental::algorithm_name name, std::vector predicates); + consumer(phlex::experimental::algorithm_name name, + std::vector predicates, + product_selectors input_products, + tbb::flow::graph& graph, + require_layers layers_required); + virtual ~consumer(); phlex::experimental::algorithm_name const& name() const noexcept; - phlex::experimental::identifier const& plugin() const noexcept; - phlex::experimental::identifier const& algorithm() const noexcept; std::vector const& when() const noexcept; + product_selectors const& input() const noexcept; + std::vector const& layers() const noexcept; + tbb::flow::receiver& port(product_selector const& input_product); + + virtual named_index_ports index_ports() = 0; + virtual std::size_t num_calls() const = 0; + + protected: + template + auto input_arguments() + { + return form_input_arguments(input_products_); + } + private: + virtual tbb::flow::receiver& port_for(product_selector const& input_product) = 0; phlex::experimental::algorithm_name name_; std::vector predicates_; + std::reference_wrapper graph_; + product_selectors input_products_; + std::vector layers_; + std::vector> layer_checkers_; }; } diff --git a/phlex/core/declared_fold.cpp b/phlex/core/declared_fold.cpp index 1e3b09b50..b27ec537f 100644 --- a/phlex/core/declared_fold.cpp +++ b/phlex/core/declared_fold.cpp @@ -1,7 +1,7 @@ #include "phlex/core/declared_fold.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include @@ -16,11 +16,11 @@ namespace phlex::detail { product_selectors input_products, tbb::flow::graph& graph, std::string partition_layer) : - products_consumer{std::move(name), - std::move(predicates), - std::move(input_products), - graph, - require_layers::always}, + consumer{std::move(name), + std::move(predicates), + std::move(input_products), + graph, + require_layers::always}, partition_layer_{std::move(partition_layer)} { } diff --git a/phlex/core/declared_fold.hpp b/phlex/core/declared_fold.hpp index 8638824ac..efe45ceb4 100644 --- a/phlex/core/declared_fold.hpp +++ b/phlex/core/declared_fold.hpp @@ -1,10 +1,10 @@ #ifndef PHLEX_CORE_DECLARED_FOLD_HPP #define PHLEX_CORE_DECLARED_FOLD_HPP +#include "phlex/core/consumer.hpp" #include "phlex/core/message.hpp" #include "phlex/core/producer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/identifier.hpp" #include "phlex/phlex_core_export.hpp" @@ -17,7 +17,7 @@ #include namespace phlex::detail { - class PHLEX_CORE_EXPORT declared_fold : public products_consumer, public producer { + class PHLEX_CORE_EXPORT declared_fold : public consumer, public producer { public: declared_fold(phlex::experimental::algorithm_name name, std::vector predicates, @@ -28,7 +28,7 @@ namespace phlex::detail { phlex::experimental::algorithm_name const& name() const noexcept override { - return products_consumer::name(); + return consumer::name(); } virtual tbb::flow::receiver& partition_port() = 0; diff --git a/phlex/core/declared_observer.cpp b/phlex/core/declared_observer.cpp index e9b22d746..c1de89232 100644 --- a/phlex/core/declared_observer.cpp +++ b/phlex/core/declared_observer.cpp @@ -1,7 +1,7 @@ #include "phlex/core/declared_observer.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include @@ -15,11 +15,11 @@ namespace phlex::detail { std::vector predicates, product_selectors input_products, tbb::flow::graph& graph) : - products_consumer{std::move(name), - std::move(predicates), - std::move(input_products), - graph, - require_layers::multi_input_only} + consumer{std::move(name), + std::move(predicates), + std::move(input_products), + graph, + require_layers::multi_input_only} { } diff --git a/phlex/core/declared_observer.hpp b/phlex/core/declared_observer.hpp index 9449958e2..4049bde75 100644 --- a/phlex/core/declared_observer.hpp +++ b/phlex/core/declared_observer.hpp @@ -1,8 +1,8 @@ #ifndef PHLEX_CORE_DECLARED_OBSERVER_HPP #define PHLEX_CORE_DECLARED_OBSERVER_HPP +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/phlex_core_export.hpp" #include "phlex/utilities/simple_ptr_map.hpp" @@ -14,7 +14,7 @@ #include namespace phlex::detail { - class PHLEX_CORE_EXPORT declared_observer : public products_consumer { + class PHLEX_CORE_EXPORT declared_observer : public consumer { public: declared_observer(phlex::experimental::algorithm_name name, std::vector predicates, diff --git a/phlex/core/declared_output.cpp b/phlex/core/declared_output.cpp index 3e202056f..ec55e8929 100644 --- a/phlex/core/declared_output.cpp +++ b/phlex/core/declared_output.cpp @@ -1,8 +1,8 @@ #include "phlex/core/declared_output.hpp" -#include "phlex/core/consumer.hpp" #include "phlex/core/detail/make_algorithm_name.hpp" #include "phlex/core/message.hpp" +#include "phlex/model/algorithm_name.hpp" #include @@ -17,7 +17,8 @@ namespace phlex::detail { std::vector predicates, tbb::flow::graph& g, internal::output_function_t&& ft) : - consumer{std::move(name), std::move(predicates)}, + name_{std::move(name)}, + predicates_{std::move(predicates)}, node_{g, concurrency, [this, f = std::move(ft)](message const& msg) -> tbb::flow::continue_msg { f(*msg.store); ++calls_; @@ -26,5 +27,11 @@ namespace phlex::detail { { } + phlex::experimental::algorithm_name const& declared_output::name() const noexcept + { + return name_; + } + std::vector const& declared_output::when() const noexcept { return predicates_; } + tbb::flow::receiver& declared_output::port() noexcept { return node_; } } diff --git a/phlex/core/declared_output.hpp b/phlex/core/declared_output.hpp index a1ba68341..6e7dc4e1e 100644 --- a/phlex/core/declared_output.hpp +++ b/phlex/core/declared_output.hpp @@ -1,10 +1,10 @@ #ifndef PHLEX_CORE_DECLARED_OUTPUT_HPP #define PHLEX_CORE_DECLARED_OUTPUT_HPP -#include "phlex/core/consumer.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/message.hpp" #include "phlex/model/algorithm_name.hpp" +#include "phlex/model/identifier.hpp" #include "phlex/model/product_store.hpp" #include "phlex/phlex_core_export.hpp" #include "phlex/utilities/simple_ptr_map.hpp" @@ -21,7 +21,7 @@ namespace phlex::detail { namespace internal { using output_function_t = std::function; } - class PHLEX_CORE_EXPORT declared_output : public consumer { + class PHLEX_CORE_EXPORT declared_output { public: declared_output(phlex::experimental::algorithm_name name, std::size_t concurrency, @@ -29,10 +29,15 @@ namespace phlex::detail { tbb::flow::graph& g, internal::output_function_t&& ft); + phlex::experimental::algorithm_name const& name() const noexcept; + std::vector const& when() const noexcept; + tbb::flow::receiver& port() noexcept; std::size_t num_calls() const { return calls_; } private: + phlex::experimental::algorithm_name name_; + std::vector predicates_; tbb::flow::function_node node_; std::atomic calls_; }; diff --git a/phlex/core/declared_predicate.cpp b/phlex/core/declared_predicate.cpp index e4fb68018..40b722a65 100644 --- a/phlex/core/declared_predicate.cpp +++ b/phlex/core/declared_predicate.cpp @@ -1,7 +1,7 @@ #include "phlex/core/declared_predicate.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include @@ -15,11 +15,11 @@ namespace phlex::detail { std::vector predicates, product_selectors input_products, tbb::flow::graph& graph) : - products_consumer{std::move(name), - std::move(predicates), - std::move(input_products), - graph, - require_layers::multi_input_only} + consumer{std::move(name), + std::move(predicates), + std::move(input_products), + graph, + require_layers::multi_input_only} { } diff --git a/phlex/core/declared_predicate.hpp b/phlex/core/declared_predicate.hpp index 96d8612cf..e5c0416b1 100644 --- a/phlex/core/declared_predicate.hpp +++ b/phlex/core/declared_predicate.hpp @@ -1,8 +1,8 @@ #ifndef PHLEX_CORE_DECLARED_PREDICATE_HPP #define PHLEX_CORE_DECLARED_PREDICATE_HPP +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/phlex_core_export.hpp" #include "phlex/utilities/simple_ptr_map.hpp" @@ -16,7 +16,7 @@ namespace phlex::detail { struct predicate_result; - class PHLEX_CORE_EXPORT declared_predicate : public products_consumer { + class PHLEX_CORE_EXPORT declared_predicate : public consumer { public: declared_predicate(phlex::experimental::algorithm_name name, std::vector predicates, diff --git a/phlex/core/declared_transform.cpp b/phlex/core/declared_transform.cpp index cf1d3c270..350a475e4 100644 --- a/phlex/core/declared_transform.cpp +++ b/phlex/core/declared_transform.cpp @@ -1,7 +1,7 @@ #include "phlex/core/declared_transform.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include @@ -15,11 +15,11 @@ namespace phlex::detail { std::vector predicates, product_selectors input_products, tbb::flow::graph& graph) : - products_consumer{std::move(name), - std::move(predicates), - std::move(input_products), - graph, - require_layers::multi_input_only} + consumer{std::move(name), + std::move(predicates), + std::move(input_products), + graph, + require_layers::multi_input_only} { } diff --git a/phlex/core/declared_transform.hpp b/phlex/core/declared_transform.hpp index 2bfbaa733..2178d0e51 100644 --- a/phlex/core/declared_transform.hpp +++ b/phlex/core/declared_transform.hpp @@ -1,9 +1,9 @@ #ifndef PHLEX_CORE_DECLARED_TRANSFORM_HPP #define PHLEX_CORE_DECLARED_TRANSFORM_HPP +#include "phlex/core/consumer.hpp" #include "phlex/core/producer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/phlex_core_export.hpp" #include "phlex/utilities/simple_ptr_map.hpp" @@ -15,7 +15,7 @@ #include namespace phlex::detail { - class PHLEX_CORE_EXPORT declared_transform : public products_consumer, public producer { + class PHLEX_CORE_EXPORT declared_transform : public consumer, public producer { public: declared_transform(phlex::experimental::algorithm_name name, std::vector predicates, @@ -25,7 +25,7 @@ namespace phlex::detail { phlex::experimental::algorithm_name const& name() const noexcept override { - return products_consumer::name(); + return consumer::name(); } }; diff --git a/phlex/core/declared_unfold.cpp b/phlex/core/declared_unfold.cpp index 0407b0059..f50437a18 100644 --- a/phlex/core/declared_unfold.cpp +++ b/phlex/core/declared_unfold.cpp @@ -1,7 +1,7 @@ #include "phlex/core/declared_unfold.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/fwd.hpp" #include "phlex/model/identifier.hpp" @@ -53,11 +53,11 @@ namespace phlex::detail { product_selectors input_products, tbb::flow::graph& graph, std::string child_layer) : - products_consumer{std::move(name), - std::move(predicates), - std::move(input_products), - graph, - require_layers::always}, + consumer{std::move(name), + std::move(predicates), + std::move(input_products), + graph, + require_layers::always}, child_layer_{std::move(child_layer)} { // Forbid multi-layer inputs until Phlex v0.5. diff --git a/phlex/core/declared_unfold.hpp b/phlex/core/declared_unfold.hpp index 5676d9e1f..b04f8f05c 100644 --- a/phlex/core/declared_unfold.hpp +++ b/phlex/core/declared_unfold.hpp @@ -1,10 +1,10 @@ #ifndef PHLEX_CORE_DECLARED_UNFOLD_HPP #define PHLEX_CORE_DECLARED_UNFOLD_HPP +#include "phlex/core/consumer.hpp" #include "phlex/core/message.hpp" #include "phlex/core/producer.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/fwd.hpp" #include "phlex/model/identifier.hpp" @@ -45,7 +45,7 @@ namespace phlex::detail { std::size_t child_count_ = 0; }; - class PHLEX_CORE_EXPORT declared_unfold : public products_consumer, public producer { + class PHLEX_CORE_EXPORT declared_unfold : public consumer, public producer { public: declared_unfold(phlex::experimental::algorithm_name name, std::vector predicates, @@ -56,7 +56,7 @@ namespace phlex::detail { phlex::experimental::algorithm_name const& name() const noexcept override { - return products_consumer::name(); + return consumer::name(); } virtual tbb::flow::sender& output_index_port() = 0; diff --git a/phlex/core/filter.cpp b/phlex/core/filter.cpp index 1cce62527..f39970fc0 100644 --- a/phlex/core/filter.cpp +++ b/phlex/core/filter.cpp @@ -1,10 +1,10 @@ #include "phlex/core/filter.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_output.hpp" #include "phlex/core/detail/filter_impl.hpp" #include "phlex/core/message.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include @@ -16,7 +16,7 @@ using namespace phlex::detail; using namespace oneapi::tbb; namespace phlex::detail { - filter::filter(flow::graph& g, products_consumer& consumer) : + filter::filter(flow::graph& g, consumer& consumer) : filter_base{g}, decisions_{static_cast(consumer.when().size())}, data_{consumer.input()}, diff --git a/phlex/core/filter.hpp b/phlex/core/filter.hpp index 3c4aa2bd3..1da03eeed 100644 --- a/phlex/core/filter.hpp +++ b/phlex/core/filter.hpp @@ -21,7 +21,7 @@ namespace phlex::detail { using filter_base::input_ports_type; using filter_base::output_ports_type; - explicit filter(oneapi::tbb::flow::graph& g, products_consumer& consumer); + explicit filter(oneapi::tbb::flow::graph& g, consumer& consumer); explicit filter(oneapi::tbb::flow::graph& g, declared_output& output); auto& data_port() { return input_port<0>(*this); } diff --git a/phlex/core/fold_node.hpp b/phlex/core/fold_node.hpp index 0d5847eea..f1498f249 100644 --- a/phlex/core/fold_node.hpp +++ b/phlex/core/fold_node.hpp @@ -3,6 +3,7 @@ #include "phlex/concurrency.hpp" #include "phlex/core/concepts.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_fold.hpp" #include "phlex/core/fold/send.hpp" #include "phlex/core/fold_join_node.hpp" @@ -11,7 +12,6 @@ #include "phlex/core/message.hpp" #include "phlex/core/node_builder.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/resource_api.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" diff --git a/phlex/core/fwd.hpp b/phlex/core/fwd.hpp index 8d6dc281b..d3f69c6b2 100644 --- a/phlex/core/fwd.hpp +++ b/phlex/core/fwd.hpp @@ -10,7 +10,6 @@ namespace phlex::detail { class framework_graph; struct message; class index_router; - class products_consumer; } #endif // PHLEX_CORE_FWD_HPP diff --git a/phlex/core/make_computational_edges.cpp b/phlex/core/make_computational_edges.cpp index 859a2d70a..3f2d35214 100644 --- a/phlex/core/make_computational_edges.cpp +++ b/phlex/core/make_computational_edges.cpp @@ -1,5 +1,6 @@ #include "phlex/core/make_computational_edges.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_output.hpp" #include "phlex/core/filter.hpp" #include "phlex/core/index_router.hpp" @@ -7,7 +8,6 @@ #include "phlex/core/node_catalog.hpp" #include "phlex/core/producer_catalog.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/provider_node.hpp" #include "phlex/core/source.hpp" #include "phlex/model/identifier.hpp" @@ -187,7 +187,7 @@ namespace phlex::detail { index_router::head_ports_t edges_within_computational_graph( producer_catalog const& producers, std::map& filters, - std::span consumers, + std::span consumers, phlex::experimental::identifier const& stage) { index_router::head_ports_t result; @@ -215,8 +215,7 @@ namespace phlex::detail { return result; } - std::map multilayer_ports( - std::span consumers) + std::map multilayer_ports(std::span consumers) { std::map result; for (auto* node : consumers) { diff --git a/phlex/core/node_catalog.cpp b/phlex/core/node_catalog.cpp index c1d9c39a2..7c142a5aa 100644 --- a/phlex/core/node_catalog.cpp +++ b/phlex/core/node_catalog.cpp @@ -1,8 +1,8 @@ #include "phlex/core/node_catalog.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/producer.hpp" #include "phlex/core/producer_catalog.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/source.hpp" #include @@ -16,15 +16,14 @@ using namespace std::string_literals; namespace phlex::detail { - std::vector node_catalog::consumers() const + std::vector node_catalog::consumers() const { auto as_product_consumers = [](auto const& nodes) { return nodes | std::views::values | - std::views::transform( - [](auto const& node) -> products_consumer* { return node.get(); }); + std::views::transform([](auto const& node) -> consumer* { return node.get(); }); }; - std::vector result; + std::vector result; result.append_range(as_product_consumers(predicates)); result.append_range(as_product_consumers(observers)); result.append_range(as_product_consumers(folds)); diff --git a/phlex/core/node_catalog.hpp b/phlex/core/node_catalog.hpp index 219f83f92..4d38b2394 100644 --- a/phlex/core/node_catalog.hpp +++ b/phlex/core/node_catalog.hpp @@ -1,6 +1,7 @@ #ifndef PHLEX_CORE_NODE_CATALOG_HPP #define PHLEX_CORE_NODE_CATALOG_HPP +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_fold.hpp" #include "phlex/core/declared_observer.hpp" #include "phlex/core/declared_output.hpp" @@ -8,7 +9,6 @@ #include "phlex/core/declared_transform.hpp" #include "phlex/core/declared_unfold.hpp" #include "phlex/core/producer_catalog.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/provider_node.hpp" #include "phlex/core/registrar.hpp" #include "phlex/core/source.hpp" @@ -102,7 +102,7 @@ namespace phlex::detail { source_vector sources_for(std::vector const& keys) const; std::size_t execution_count(std::string const& node_name) const; - std::vector consumers() const; + std::vector consumers() const; producer_catalog producers() const; simple_ptr_map predicates; diff --git a/phlex/core/observer_node.hpp b/phlex/core/observer_node.hpp index 768c43cbe..b0e40b0ee 100644 --- a/phlex/core/observer_node.hpp +++ b/phlex/core/observer_node.hpp @@ -2,6 +2,7 @@ #define PHLEX_CORE_OBSERVER_NODE_HPP #include "phlex/core/concepts.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_observer.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" @@ -9,7 +10,6 @@ #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/resource_api.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" diff --git a/phlex/core/predicate_node.hpp b/phlex/core/predicate_node.hpp index ffa8483c4..605c36bff 100644 --- a/phlex/core/predicate_node.hpp +++ b/phlex/core/predicate_node.hpp @@ -2,6 +2,7 @@ #define PHLEX_CORE_PREDICATE_NODE_HPP #include "phlex/core/concepts.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_predicate.hpp" #include "phlex/core/detail/filter_impl.hpp" #include "phlex/core/fwd.hpp" @@ -10,7 +11,6 @@ #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/resource_api.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" diff --git a/phlex/core/products_consumer.cpp b/phlex/core/products_consumer.cpp deleted file mode 100644 index 1224068e7..000000000 --- a/phlex/core/products_consumer.cpp +++ /dev/null @@ -1,110 +0,0 @@ -#include "phlex/core/products_consumer.hpp" - -#include "phlex/core/consumer.hpp" -#include "phlex/core/message.hpp" -#include "phlex/core/product_selector.hpp" -#include "phlex/model/algorithm_name.hpp" -#include "phlex/model/identifier.hpp" -#include "phlex/utilities/bulleted_list.hpp" - -#include -#include - -#include -#include -#include -#include -#include -#include - -namespace { - std::vector layers_from(phlex::product_selectors const& queries) - { - using namespace phlex::experimental::literals; - std::vector result; - result.reserve(queries.size()); - for (auto const& query : queries) { - if (query.layer) { - result.push_back(query.layer); - } else { - result.push_back("*"_id); - } - } - return result; - } - - void validate_layers(phlex::detail::require_layers layers_required, - phlex::product_selectors const& inputs, - phlex::experimental::algorithm_name const& algo) - { - using namespace phlex::detail; - if (layers_required == require_layers::never) { - return; - } - if (layers_required == require_layers::multi_input_only && inputs.size() <= 1) { - return; - } - std::vector err_selectors{}; - for (auto const& p : inputs) { - if (!p.layer) { - err_selectors.push_back(p.to_string()); - } - } - if (!err_selectors.empty()) { - std::string const error = - fmt::format("Must specify layers in the product selectors for node {}:\n" - " (Only invalid selectors are listed)\n{}", - algo.to_string(), - bulleted_list(err_selectors)); - throw std::runtime_error(error); - } - } -} - -namespace phlex::detail { - - products_consumer::products_consumer(phlex::experimental::algorithm_name name, - std::vector predicates, - product_selectors input_products, - tbb::flow::graph& graph, - require_layers layers_required) : - consumer{std::move(name), std::move(predicates)}, - graph_{graph}, - input_products_{std::move(input_products)}, - layers_{layers_from(input_products_)} - { - validate_layers(layers_required, input_products_, this->name()); - } - - products_consumer::~products_consumer() = default; - - std::size_t products_consumer::num_inputs() const { return input().size(); } - - tbb::flow::receiver& products_consumer::port(product_selector const& input_product) - { - auto& next = port_for(input_product); - - // If input_product doesn't have a layer, it must be for a node that allows layer omission - if (input_product.layer) { - auto& layer_check = layer_checkers_.emplace_back(std::make_unique( - graph_, - tbb::flow::unlimited, - [&layer = static_cast(input_product.layer)]( - message const& msg, auto& output) { - if (msg.store->layer_name() == layer) { - std::get<0>(output).try_put(msg); - } - })); - make_edge(tbb::flow::output_port<0>(*layer_check), next); - return *layer_check; - } - // else - return next; - } - - product_selectors const& products_consumer::input() const noexcept { return input_products_; } - std::vector const& products_consumer::layers() const noexcept - { - return layers_; - } -} diff --git a/phlex/core/products_consumer.hpp b/phlex/core/products_consumer.hpp deleted file mode 100644 index 1ac80dd78..000000000 --- a/phlex/core/products_consumer.hpp +++ /dev/null @@ -1,58 +0,0 @@ -#ifndef PHLEX_CORE_PRODUCTS_CONSUMER_HPP -#define PHLEX_CORE_PRODUCTS_CONSUMER_HPP - -#include "phlex/core/consumer.hpp" -#include "phlex/core/fwd.hpp" -#include "phlex/core/input_arguments.hpp" -#include "phlex/core/message.hpp" -#include "phlex/core/product_selector.hpp" -#include "phlex/model/algorithm_name.hpp" -#include "phlex/model/identifier.hpp" -#include "phlex/phlex_core_export.hpp" - -#include - -#include -#include -#include - -namespace phlex::detail { - enum class require_layers : char { never, multi_input_only, always }; - class PHLEX_CORE_EXPORT products_consumer : public consumer { - using layer_check_node_t = tbb::flow::multifunction_node>; - - public: - products_consumer(phlex::experimental::algorithm_name name, - std::vector predicates, - product_selectors input_products, - tbb::flow::graph& graph, - require_layers layers_required); - - virtual ~products_consumer(); - - std::size_t num_inputs() const; - - product_selectors const& input() const noexcept; - std::vector const& layers() const noexcept; - tbb::flow::receiver& port(product_selector const& input_product); - - virtual named_index_ports index_ports() = 0; - virtual std::size_t num_calls() const = 0; - - protected: - template - auto input_arguments() - { - return form_input_arguments(input_products_); - } - - private: - virtual tbb::flow::receiver& port_for(product_selector const& input_product) = 0; - std::reference_wrapper graph_; - product_selectors input_products_; - std::vector layers_; - std::vector> layer_checkers_; - }; -} - -#endif // PHLEX_CORE_PRODUCTS_CONSUMER_HPP diff --git a/phlex/core/transform_node.hpp b/phlex/core/transform_node.hpp index 996a74f53..56245e3b8 100644 --- a/phlex/core/transform_node.hpp +++ b/phlex/core/transform_node.hpp @@ -5,6 +5,7 @@ // of the process a given section of code is addressing. #include "phlex/core/concepts.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_transform.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" @@ -12,7 +13,6 @@ #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" #include "phlex/core/product_selector.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/resource_api.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" diff --git a/phlex/core/unfold_node.hpp b/phlex/core/unfold_node.hpp index 703b97e86..a6f3b2dc9 100644 --- a/phlex/core/unfold_node.hpp +++ b/phlex/core/unfold_node.hpp @@ -2,13 +2,13 @@ #define PHLEX_CORE_UNFOLD_NODE_HPP #include "phlex/core/concepts.hpp" +#include "phlex/core/consumer.hpp" #include "phlex/core/declared_unfold.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" #include "phlex/core/message.hpp" #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" -#include "phlex/core/products_consumer.hpp" #include "phlex/core/resource_api.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" diff --git a/test/core_misc_test.cpp b/test/core_misc_test.cpp index 62291a3da..36c2cb75f 100644 --- a/test/core_misc_test.cpp +++ b/test/core_misc_test.cpp @@ -1,5 +1,4 @@ #include "phlex/configuration.hpp" -#include "phlex/core/consumer.hpp" #include "phlex/core/glue.hpp" #include "phlex/core/registrar.hpp" #include "phlex/model/algorithm_name.hpp" @@ -62,18 +61,6 @@ TEST_CASE("algorithm_name tests", "[model]") } } -TEST_CASE("consumer tests", "[core]") -{ - using namespace phlex::experimental::literals; - auto an = phlex::experimental::algorithm_name::create("p:a"); - phlex::detail::consumer const c(an, {"pred1"}); - - CHECK(c.name().to_string() == "p:a"); - CHECK(c.plugin() == "p"_idq); - CHECK(c.algorithm() == "a"_idq); - CHECK(c.when().size() == 1); -} - TEST_CASE("verify_name tests", "[core]") { using namespace phlex::detail::internal;