Layer checking as products pass over edges

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<std::size_t> calls_;
     std::atomic<std::size_t> product_count_;
+    tbb::flow::graph& graph() const override { return graph_; }
+    std::reference_wrapper<tbb::flow::graph> graph_;
   };
 }
 

phlex/core/declared_observer.hpp

@@ -70,7 +70,8 @@ namespace phlex::detail {
                   call(ft, messages, std::make_index_sequence<num_inputs>{});
                   ++calls_;
                   return {};
-                }}
+                }},
+      graph_{g}
     {
       if constexpr (num_inputs > 1ull) {
         make_edge(join_, observer_);
@@ -107,6 +108,8 @@ namespace phlex::detail {
     join_or_none_t<num_inputs> join_;
     tbb::flow::function_node<messages_t<num_inputs>> observer_;
     std::atomic<std::size_t> calls_;
+    tbb::flow::graph& graph() const override { return graph_; }
+    std::reference_wrapper<tbb::flow::graph> graph_;
   };
 }
 

phlex/core/declared_predicate.hpp

@@ -77,7 +77,8 @@ namespace phlex::detail {
                    bool const rc = call(ft, messages, std::make_index_sequence<num_inputs>{});
                    ++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<num_inputs> join_;
     tbb::flow::function_node<messages_t<num_inputs>, predicate_result> predicate_;
     std::atomic<std::size_t> calls_;
+    tbb::flow::graph& graph() const override { return graph_; }
+    std::reference_wrapper<tbb::flow::graph> graph_;
   };
 
 }

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<messages_t<num_inputs>, message_tuple<1u>> transform_;
     std::atomic<std::size_t> calls_;
     tbb::concurrent_unordered_map<std::size_t, std::atomic<std::size_t>> product_count_;
+
+    tbb::flow::graph& graph() const override { return graph_; }
+    std::reference_wrapper<tbb::flow::graph> graph_;
   };
 
 }

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<std::size_t> msg_counter_{}; // Is this sufficient?  Probably not.
     std::atomic<std::size_t> calls_{};
     std::atomic<std::size_t> product_count_{};
+    tbb::flow::graph& graph() const override { return graph_; }
+    std::reference_wrapper<tbb::flow::graph> graph_;
   };
 }
 

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;

phlex/core/producer_catalog.cpp

@@ -1,19 +1,27 @@
 #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"
 #include "spdlog/spdlog.h"
 #include <ranges>
 
 namespace phlex::detail {
-  producer_catalog::named_output_port const* producer_catalog::find_producer(
+  std::vector<producer_catalog::named_output_port const*> 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<std::string, named_output_port const*> candidates;
+    std::vector<named_output_port const*> candidates;
+
+    // Don't really want to copy these fields so we'll use hashes and store a pointer to one copy
+    std::map<std::uint64_t, experimental::identifier const*> suffixes;
+    if (query.suffix.has_value()) {
+      suffixes.emplace(query.suffix->hash(), &*query.suffix);
+    }
+    std::map<std::uint64_t, experimental::algorithm_name const*> creators;
+    std::map<std::uint64_t, type_id const*> 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([]<typename T>(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;
   }
 }

phlex/core/producer_catalog.hpp

@@ -28,7 +28,7 @@ namespace phlex::detail {
       type_id type;
     };
 
-    named_output_port const* find_producer(
+    std::vector<named_output_port const*> find_producers(
       product_selector const& query,
       phlex::experimental::algorithm_name const& consumer_name) const;
     auto values() const { return producers_ | std::views::values; }

phlex/core/products_consumer.cpp

@@ -1,4 +1,5 @@
 #include "phlex/core/products_consumer.hpp"
+#include <spdlog/spdlog.h>
 
 namespace {
   std::vector<phlex::experimental::identifier> layers_from(phlex::product_selectors const& queries)
@@ -29,7 +30,20 @@ namespace phlex::detail {
 
   tbb::flow::receiver<message>& 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<layer_check_node_t>(
+      graph(),
+      tbb::flow::unlimited,
+      [&layer = static_cast<experimental::identifier const&>(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_; }

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<message, message_tuple<1UZ>>;
+
   public:
     products_consumer(phlex::experimental::algorithm_name name,
                       std::vector<std::string> predicates,
@@ -44,9 +46,10 @@ namespace phlex::detail {
 
   private:
     virtual tbb::flow::receiver<message>& port_for(product_selector const& input_product) = 0;
-
+    virtual tbb::flow::graph& graph() const = 0;
     product_selectors input_products_;
     std::vector<phlex::experimental::identifier> layers_;
+    std::vector<std::unique_ptr<layer_check_node_t>> layer_checkers_;
   };
 }
 

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<double>& 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
+  }
 }