diff --git a/phlex/core/declared_fold.hpp b/phlex/core/declared_fold.hpp index c9656d15b..30900ba5c 100644 --- a/phlex/core/declared_fold.hpp +++ b/phlex/core/declared_fold.hpp @@ -114,7 +114,8 @@ namespace phlex::detail { counter_for(index_hash_for_counter).increment(index->layer_hash()); emit_and_evict_if_done(fold_index); - }} + }}, + graph_{g} { if constexpr (num_inputs > 1ull) { make_edge(join_, fold_); @@ -206,6 +207,8 @@ namespace phlex::detail { results_; std::atomic calls_; std::atomic product_count_; + tbb::flow::graph& graph() const override { return graph_; } + std::reference_wrapper graph_; }; } diff --git a/phlex/core/declared_observer.hpp b/phlex/core/declared_observer.hpp index 213df016d..08d688bcc 100644 --- a/phlex/core/declared_observer.hpp +++ b/phlex/core/declared_observer.hpp @@ -70,7 +70,8 @@ namespace phlex::detail { call(ft, messages, std::make_index_sequence{}); ++calls_; return {}; - }} + }}, + graph_{g} { if constexpr (num_inputs > 1ull) { make_edge(join_, observer_); @@ -107,6 +108,8 @@ namespace phlex::detail { join_or_none_t join_; tbb::flow::function_node> observer_; std::atomic calls_; + tbb::flow::graph& graph() const override { return graph_; } + std::reference_wrapper graph_; }; } diff --git a/phlex/core/declared_predicate.hpp b/phlex/core/declared_predicate.hpp index 892d991dd..a101a1748 100644 --- a/phlex/core/declared_predicate.hpp +++ b/phlex/core/declared_predicate.hpp @@ -77,7 +77,8 @@ namespace phlex::detail { bool const rc = call(ft, messages, std::make_index_sequence{}); ++calls_; return {message_id, rc}; - }} + }}, + graph_{g} { if constexpr (num_inputs > 1ull) { make_edge(join_, predicate_); @@ -115,6 +116,8 @@ namespace phlex::detail { join_or_none_t join_; tbb::flow::function_node, predicate_result> predicate_; std::atomic calls_; + tbb::flow::graph& graph() const override { return graph_; } + std::reference_wrapper graph_; }; } diff --git a/phlex/core/declared_transform.hpp b/phlex/core/declared_transform.hpp index 407503d3e..694f51aca 100644 --- a/phlex/core/declared_transform.hpp +++ b/phlex/core/declared_transform.hpp @@ -96,7 +96,8 @@ namespace phlex::detail { store->index(), name(), std::move(new_products)); std::get<0>(output).try_put({.store = std::move(new_store), .id = message_id}); - }} + }}, + graph_{g} { if constexpr (num_inputs > 1ull) { make_edge(join_, transform_); @@ -149,6 +150,9 @@ namespace phlex::detail { tbb::flow::multifunction_node, message_tuple<1u>> transform_; std::atomic calls_; tbb::concurrent_unordered_map> product_count_; + + tbb::flow::graph& graph() const override { return graph_; } + std::reference_wrapper graph_; }; } diff --git a/phlex/core/declared_unfold.hpp b/phlex/core/declared_unfold.hpp index 7d8063a9d..9ace0b96b 100644 --- a/phlex/core/declared_unfold.hpp +++ b/phlex/core/declared_unfold.hpp @@ -117,7 +117,8 @@ namespace phlex::detail { std::get<2>(outputs).try_put({.index = store->index(), .layer_hash = gen.child_layer_hash(), .count = gen.child_count()}); - }} + }}, + graph_{g} { if constexpr (num_inputs > 1ull) { make_edge(join_, unfold_); @@ -200,6 +201,8 @@ namespace phlex::detail { std::atomic msg_counter_{}; // Is this sufficient? Probably not. std::atomic calls_{}; std::atomic product_count_{}; + tbb::flow::graph& graph() const override { return graph_; } + std::reference_wrapper graph_; }; } diff --git a/phlex/core/make_computational_edges.cpp b/phlex/core/make_computational_edges.cpp index 99a68caf9..5af1f6999 100644 --- a/phlex/core/make_computational_edges.cpp +++ b/phlex/core/make_computational_edges.cpp @@ -147,14 +147,16 @@ namespace phlex::detail { for (auto const& query : node->input()) { auto* receiver_port = collector ? collector : &node->port(query); - auto producer = producers.find_producer(query, node->name()); - if (not producer) { + auto producer_ports = producers.find_producers(query, node->name()); + if (producer_ports.empty()) { // Is there a way to detect mis-specified product dependencies? result[node_name].push_back({query, receiver_port}); continue; } - make_edge(*producer->output_port, *receiver_port); + for (auto* producer : producer_ports) { + make_edge(*producer->output_port, *receiver_port); + } } } return result; diff --git a/phlex/core/producer_catalog.cpp b/phlex/core/producer_catalog.cpp index b1d19a90d..0ad0b0140 100644 --- a/phlex/core/producer_catalog.cpp +++ b/phlex/core/producer_catalog.cpp @@ -1,5 +1,6 @@ #include "phlex/core/producer_catalog.hpp" #include "phlex/utilities/bulleted_list.hpp" +#include "phlex/utilities/hashing.hpp" #include "fmt/format.h" #include "fmt/ranges.h" @@ -7,13 +8,20 @@ #include namespace phlex::detail { - producer_catalog::named_output_port const* producer_catalog::find_producer( + std::vector producer_catalog::find_producers( product_selector const& query, phlex::experimental::algorithm_name const& consumer_name) const { + // Will need an update when we have a way to set the current stage name + if (query.stage.has_value() && query.stage.value() != "CURRENT"_idq) { + spdlog::debug( + "{} requires a stage other than the current one. Assuming it comes from a provider.", + query.to_string()); + return {}; + } if (producers_.empty()) { spdlog::debug("No producers found. Skipping and assuming {} comes from a provider.", query.to_string()); - return nullptr; + return {}; } // Now the only way b == e is if we have a suffix and nothing creates matching products auto [b, e] = query.suffix.has_value() ? producers_.equal_range(*query.suffix) @@ -22,9 +30,18 @@ namespace phlex::detail { spdlog::debug( "Failed to find an algorithm that creates {} products. Assuming it comes from a provider", query.suffix.value_or("*"_id)); - return nullptr; + return {}; } - std::map candidates; + std::vector candidates; + + // Don't really want to copy these fields so we'll use hashes and store a pointer to one copy + std::map suffixes; + if (query.suffix.has_value()) { + suffixes.emplace(query.suffix->hash(), &*query.suffix); + } + std::map creators; + std::map types; + for (auto const& [key, producer] : std::ranges::subrange{b, e}) { // Prevent self-edges if (producer.node == consumer_name) { @@ -59,7 +76,14 @@ namespace phlex::detail { query.type.exact_name(), producer.type.exact_name()); } - candidates.emplace(producer.node.to_string(), &producer); + candidates.push_back(&producer); + if (!query.suffix.has_value()) { + suffixes.emplace(key.hash(), &key); + } + creators.emplace( + phlex::detail::hash(producer.node.plugin().hash(), producer.node.algorithm().hash()), + &producer.node); + types.emplace(hash_value(producer.type), &producer.type); } } else { spdlog::debug("Creator name mismatch between ({}) and {}", @@ -72,17 +96,41 @@ namespace phlex::detail { spdlog::debug( "Cannot identify product matching the query {}. Assuming it comes from a provider.", query.to_string()); - return nullptr; } if (candidates.size() > 1ull) { - std::string msg = fmt::format("More than one candidate matches the query {}: \n{}\n", + static auto const port_to_node = + [](named_output_port const* p) -> experimental::algorithm_name const& { return p->node; }; + static auto const deref_view = + std::views::transform([](T const* p) -> T const& { return *p; }); + std::string msg = fmt::format("More than one candidate matches the query {}: \n{}", query.to_string(), - bulleted_list(std::views::keys(candidates), + bulleted_list(std::views::transform(candidates, port_to_node), /*indent=*/1)); - throw std::runtime_error(msg); + if (suffixes.size() == 1 && creators.size() == 1 && types.size() == 1) { + spdlog::info(msg); + spdlog::info("This is permitted -- layers may differ"); + return candidates; + } + spdlog::error(msg); + + if (suffixes.size() > 1) { + spdlog::error("Not permitted -- distinguishable by suffix {}", + suffixes | std::views::values | deref_view); + } + if (creators.size() > 1) { + spdlog::error("Not permitted -- distinguishable by creator {}", + creators | std::views::values | + std::views::transform( + [](experimental::algorithm_name const* p) { return p->to_string(); })); + } + if (types.size() > 1) { + spdlog::error("Not permitted -- distinguishable by type {}", + types | std::views::values | deref_view); + } + throw std::runtime_error("Multiple products in candidate set -- see errors"); } - return candidates.begin()->second; + return candidates; } } diff --git a/phlex/core/producer_catalog.hpp b/phlex/core/producer_catalog.hpp index d425ff403..8ea4bc863 100644 --- a/phlex/core/producer_catalog.hpp +++ b/phlex/core/producer_catalog.hpp @@ -28,7 +28,7 @@ namespace phlex::detail { type_id type; }; - named_output_port const* find_producer( + std::vector find_producers( product_selector const& query, phlex::experimental::algorithm_name const& consumer_name) const; auto values() const { return producers_ | std::views::values; } diff --git a/phlex/core/products_consumer.cpp b/phlex/core/products_consumer.cpp index af268bb42..0f8074bd3 100644 --- a/phlex/core/products_consumer.cpp +++ b/phlex/core/products_consumer.cpp @@ -1,4 +1,5 @@ #include "phlex/core/products_consumer.hpp" +#include namespace { std::vector layers_from(phlex::product_selectors const& queries) @@ -29,7 +30,20 @@ namespace phlex::detail { tbb::flow::receiver& products_consumer::port(product_selector const& input_product) { - return port_for(input_product); + // Everything has a layer for now, so everything needs this + auto& next = port_for(input_product); + + 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; } product_selectors const& products_consumer::input() const noexcept { return input_products_; } diff --git a/phlex/core/products_consumer.hpp b/phlex/core/products_consumer.hpp index 1c77ac382..98030f735 100644 --- a/phlex/core/products_consumer.hpp +++ b/phlex/core/products_consumer.hpp @@ -18,6 +18,8 @@ namespace phlex::detail { 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, @@ -44,9 +46,10 @@ namespace phlex::detail { private: virtual tbb::flow::receiver& port_for(product_selector const& input_product) = 0; - + virtual tbb::flow::graph& graph() const = 0; product_selectors input_products_; std::vector layers_; + std::vector> layer_checkers_; }; } diff --git a/test/product_selecting.cpp b/test/product_selecting.cpp index 47940b831..54ea48053 100644 --- a/test/product_selecting.cpp +++ b/test/product_selecting.cpp @@ -99,4 +99,23 @@ TEST_CASE("Querying products in different ways", "[graph]") g.execute(); CHECK(g.execution_count("creator_and_layer_after_transform") == num_events); } + + SECTION("Products from this job, using layer only") + { + // create a product to be found + g.fold( + "duplicate_temperature", + [](std::atomic& summary, double temp) { summary += temp; }, + concurrency::unlimited, + "job") + .input_family(product_selector{.creator = "input", .layer = "event"}) + .output_product_suffixes("temperature"); + + // Find it + g.transform("layer_only", [](double const& d) { return d; }) + .input_family(product_selector{.layer = "job"}) + .output_product_suffixes("job_temp"); + g.execute(); + CHECK(g.execution_count("layer_only") == 1); // 1 job + } }