Add type checking to graph construction

phlex/core/CMakeLists.txt

@@ -24,7 +24,7 @@ add_library(
   products_consumer.cpp
   registrar.cpp
   registration_api.cpp
-  specified_label.cpp
+  product_query.cpp
   store_counters.cpp
   )
 target_include_directories(phlex_core PRIVATE ${PROJECT_SOURCE_DIR})

phlex/core/declared_fold.cpp

@@ -3,7 +3,7 @@
 namespace phlex::experimental {
   declared_fold::declared_fold(algorithm_name name,
                                std::vector<std::string> predicates,
-                               specified_labels input_products) :
+                               product_queries input_products) :
     products_consumer{std::move(name), std::move(predicates), std::move(input_products)}
   {
   }

phlex/core/declared_fold.hpp

@@ -7,14 +7,14 @@
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/core/product_query.hpp"
 #include "phlex/core/products_consumer.hpp"
-#include "phlex/core/specified_label.hpp"
 #include "phlex/core/store_counters.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/handle.hpp"
 #include "phlex/model/level_id.hpp"
+#include "phlex/model/product_specification.hpp"
 #include "phlex/model/product_store.hpp"
-#include "phlex/model/qualified_name.hpp"
 #include "phlex/utilities/simple_ptr_map.hpp"
 
 #include "oneapi/tbb/concurrent_unordered_map.h"
@@ -37,12 +37,12 @@ namespace phlex::experimental {
   public:
     declared_fold(algorithm_name name,
                   std::vector<std::string> predicates,
-                  specified_labels input_products);
+                  product_queries input_products);
     virtual ~declared_fold();
 
     virtual tbb::flow::sender<message>& sender() = 0;
     virtual tbb::flow::sender<message>& to_output() = 0;
-    virtual qualified_names const& output() const = 0;
+    virtual product_specifications const& output() const = 0;
     virtual std::size_t product_count() const = 0;
   };
 
@@ -68,12 +68,12 @@ namespace phlex::experimental {
               tbb::flow::graph& g,
               AlgorithmBits alg,
               InitTuple initializer,
-              specified_labels product_labels,
+              product_queries product_labels,
               std::vector<std::string> output,
               std::string partition) :
       declared_fold{std::move(name), std::move(predicates), std::move(product_labels)},
       initializer_{std::move(initializer)},
-      output_{to_qualified_names(full_name(), std::move(output))},
+      output_{to_product_specifications(full_name(), std::move(output), make_type_ids<R>())},
       partition_{std::move(partition)},
       join_{make_join_or_none(g, std::make_index_sequence<N>{})},
       fold_{g,
@@ -121,7 +121,7 @@ namespace phlex::experimental {
     }
 
   private:
-    tbb::flow::receiver<message>& port_for(specified_label const& product_label) override
+    tbb::flow::receiver<message>& port_for(product_query const& product_label) override
     {
       return receiver_for<N>(join_, input(), product_label);
     }
@@ -130,7 +130,7 @@ namespace phlex::experimental {
 
     tbb::flow::sender<message>& sender() override { return output_port<0ull>(fold_); }
     tbb::flow::sender<message>& to_output() override { return sender(); }
-    qualified_names const& output() const override { return output_; }
+    product_specifications const& output() const override { return output_; }
 
     template <std::size_t... Is>
     void call(function_t const& ft, messages_t<N> const& messages, std::index_sequence<Is...>)
@@ -175,7 +175,7 @@ namespace phlex::experimental {
 
     InitTuple initializer_;
     input_retriever_types<input_parameter_types> input_{input_arguments<input_parameter_types>()};
-    qualified_names output_;
+    product_specifications output_;
     std::string partition_;
     join_or_none_t<N> join_;
     tbb::flow::multifunction_node<messages_t<N>, messages_t<1>> fold_;

phlex/core/declared_observer.cpp

@@ -6,7 +6,7 @@
 namespace phlex::experimental {
   declared_observer::declared_observer(algorithm_name name,
                                        std::vector<std::string> predicates,
-                                       specified_labels input_products) :
+                                       product_queries input_products) :
     products_consumer{std::move(name), std::move(predicates), std::move(input_products)}
   {
   }

phlex/core/declared_observer.hpp

@@ -5,15 +5,15 @@
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/core/product_query.hpp"
 #include "phlex/core/products_consumer.hpp"
-#include "phlex/core/specified_label.hpp"
 #include "phlex/core/store_counters.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/handle.hpp"
 #include "phlex/model/level_id.hpp"
+#include "phlex/model/product_specification.hpp"
 #include "phlex/model/product_store.hpp"
-#include "phlex/model/qualified_name.hpp"
 #include "phlex/utilities/simple_ptr_map.hpp"
 
 #include "oneapi/tbb/concurrent_hash_map.h"
@@ -34,7 +34,7 @@ namespace phlex::experimental {
   public:
     declared_observer(algorithm_name name,
                       std::vector<std::string> predicates,
-                      specified_labels input_products);
+                      product_queries input_products);
     virtual ~declared_observer();
 
   protected:
@@ -64,7 +64,7 @@ namespace phlex::experimental {
                   std::vector<std::string> predicates,
                   tbb::flow::graph& g,
                   AlgorithmBits alg,
-                  specified_labels input_products) :
+                  product_queries input_products) :
       declared_observer{std::move(name), std::move(predicates), std::move(input_products)},
       join_{make_join_or_none(g, std::make_index_sequence<N>{})},
       observer_{g,
@@ -93,7 +93,7 @@ namespace phlex::experimental {
     ~observer_node() { report_cached_hashes(cached_hashes_); }
 
   private:
-    tbb::flow::receiver<message>& port_for(specified_label const& product_label) override
+    tbb::flow::receiver<message>& port_for(product_query const& product_label) override
     {
       return receiver_for<N>(join_, input(), product_label);
     }

phlex/core/declared_predicate.cpp

@@ -6,7 +6,7 @@
 namespace phlex::experimental {
   declared_predicate::declared_predicate(algorithm_name name,
                                          std::vector<std::string> predicates,
-                                         specified_labels input_products) :
+                                         product_queries input_products) :
     products_consumer{std::move(name), std::move(predicates), std::move(input_products)}
   {
   }

phlex/core/declared_predicate.hpp

@@ -6,8 +6,8 @@
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/core/product_query.hpp"
 #include "phlex/core/products_consumer.hpp"
-#include "phlex/core/specified_label.hpp"
 #include "phlex/core/store_counters.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
@@ -37,7 +37,7 @@ namespace phlex::experimental {
   public:
     declared_predicate(algorithm_name name,
                        std::vector<std::string> predicates,
-                       specified_labels input_products);
+                       product_queries input_products);
     virtual ~declared_predicate();
 
     virtual tbb::flow::sender<predicate_result>& sender() = 0;
@@ -70,7 +70,7 @@ namespace phlex::experimental {
                    std::vector<std::string> predicates,
                    tbb::flow::graph& g,
                    AlgorithmBits alg,
-                   specified_labels input_products) :
+                   product_queries input_products) :
       declared_predicate{std::move(name), std::move(predicates), std::move(input_products)},
       join_{make_join_or_none(g, std::make_index_sequence<N>{})},
       predicate_{
@@ -102,7 +102,7 @@ namespace phlex::experimental {
     ~predicate_node() { report_cached_results(results_); }
 
   private:
-    tbb::flow::receiver<message>& port_for(specified_label const& product_label) override
+    tbb::flow::receiver<message>& port_for(product_query const& product_label) override
     {
       return receiver_for<N>(join_, input(), product_label);
     }

phlex/core/declared_transform.cpp

@@ -6,7 +6,7 @@
 namespace phlex::experimental {
   declared_transform::declared_transform(algorithm_name name,
                                          std::vector<std::string> predicates,
-                                         specified_labels input_products) :
+                                         product_queries input_products) :
     products_consumer{std::move(name), std::move(predicates), std::move(input_products)}
   {
   }
@@ -19,6 +19,10 @@
       spdlog::warn("Transform {} has {} cached stores.", full_name(), stores.size());
     }
     for (auto const& [hash, store] : stores) {
+      if (not store) {
+        spdlog::warn("Store with hash {} is null!", hash);
+        continue;
+      }
       spdlog::debug(" => ID: {} (hash: {})", store->id()->to_string(), hash);
     }
   }

phlex/core/declared_transform.hpp

@@ -8,15 +8,15 @@
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/core/product_query.hpp"
 #include "phlex/core/products_consumer.hpp"
-#include "phlex/core/specified_label.hpp"
 #include "phlex/core/store_counters.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/handle.hpp"
 #include "phlex/model/level_id.hpp"
+#include "phlex/model/product_specification.hpp"
 #include "phlex/model/product_store.hpp"
-#include "phlex/model/qualified_name.hpp"
 #include "phlex/utilities/simple_ptr_map.hpp"
 
 #include "oneapi/tbb/concurrent_hash_map.h"
@@ -41,12 +41,12 @@ namespace phlex::experimental {
   public:
     declared_transform(algorithm_name name,
                        std::vector<std::string> predicates,
-                       specified_labels input_products);
+                       product_queries input_products);
     virtual ~declared_transform();
 
     virtual tbb::flow::sender<message>& sender() = 0;
     virtual tbb::flow::sender<message>& to_output() = 0;
-    virtual qualified_names const& output() const = 0;
+    virtual product_specifications const& output() const = 0;
     virtual std::size_t product_count() const = 0;
 
   protected:
@@ -79,10 +79,11 @@ namespace phlex::experimental {
                    std::vector<std::string> predicates,
                    tbb::flow::graph& g,
                    AlgorithmBits alg,
-                   specified_labels input_products,
+                   product_queries input_products,
                    std::vector<std::string> output) :
       declared_transform{std::move(name), std::move(predicates), std::move(input_products)},
-      output_{to_qualified_names(full_name(), std::move(output))},
+      output_{to_product_specifications(
+        full_name(), std::move(output), make_output_type_ids<function_t>())},
       join_{make_join_or_none(g, std::make_index_sequence<N>{})},
       transform_{g,
                  concurrency,
@@ -126,7 +127,7 @@ namespace phlex::experimental {
     ~transform_node() { report_cached_stores(stores_); }
 
   private:
-    tbb::flow::receiver<message>& port_for(specified_label const& product_label) override
+    tbb::flow::receiver<message>& port_for(product_query const& product_label) override
     {
       return receiver_for<N>(join_, input(), product_label);
     }
@@ -135,7 +136,7 @@ namespace phlex::experimental {
 
     tbb::flow::sender<message>& sender() override { return output_port<0>(transform_); }
     tbb::flow::sender<message>& to_output() override { return output_port<1>(transform_); }
-    qualified_names const& output() const override { return output_; }
+    product_specifications const& output() const override { return output_; }
 
     template <std::size_t... Is>
     auto call(function_t const& ft, messages_t<N> const& messages, std::index_sequence<Is...>)
@@ -154,7 +155,7 @@ namespace phlex::experimental {
     }
 
     input_retriever_types<input_parameter_types> input_{input_arguments<input_parameter_types>()};
-    qualified_names output_;
+    product_specifications output_;
     join_or_none_t<N> join_;
     tbb::flow::multifunction_node<messages_t<N>, messages_t<2u>> transform_;
     stores_t stores_;

phlex/core/declared_unfold.cpp

@@ -35,7 +35,7 @@
 
   declared_unfold::declared_unfold(algorithm_name name,
                                    std::vector<std::string> predicates,
-                                   specified_labels input_products) :
+                                   product_queries input_products) :
     products_consumer{std::move(name), std::move(predicates), std::move(input_products)}
   {
   }
@@ -48,6 +48,10 @@
       spdlog::warn("Unfold {} has {} cached stores.", full_name(), stores.size());
     }
     for (auto const& [hash, store] : stores) {
+      if (not store) {
+        spdlog::warn("Store with hash {} is null!", hash);
+        continue;
+      }
       spdlog::debug(" => ID: {} (hash: {})", store->id()->to_string(), hash);
     }
   }

phlex/core/declared_unfold.hpp

@@ -11,8 +11,8 @@
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/handle.hpp"
 #include "phlex/model/level_id.hpp"
+#include "phlex/model/product_specification.hpp"
 #include "phlex/model/product_store.hpp"
-#include "phlex/model/qualified_name.hpp"
 #include "phlex/utilities/simple_ptr_map.hpp"
 
 #include "oneapi/tbb/concurrent_hash_map.h"
@@ -57,12 +57,12 @@ namespace phlex::experimental {
   public:
     declared_unfold(algorithm_name name,
                     std::vector<std::string> predicates,
-                    specified_labels input_products);
+                    product_queries input_products);
     virtual ~declared_unfold();
 
     virtual tbb::flow::sender<message>& sender() = 0;
     virtual tbb::flow::sender<message>& to_output() = 0;
-    virtual qualified_names const& output() const = 0;
+    virtual product_specifications const& output() const = 0;
     virtual std::size_t product_count() const = 0;
 
   protected:
@@ -91,11 +91,13 @@ namespace phlex::experimental {
                 tbb::flow::graph& g,
                 Predicate&& predicate,
                 Unfold&& unfold,
-                specified_labels product_labels,
+                product_queries product_labels,
                 std::vector<std::string> output_products,
                 std::string new_level_name) :
       declared_unfold{std::move(name), std::move(predicates), std::move(product_labels)},
-      output_{to_qualified_names(full_name(), std::move(output_products))},
+      output_{to_product_specifications(full_name(),
+                                        std::move(output_products),
+                                        make_type_ids<skip_first_type<return_type<Unfold>>>())},
       new_level_name_{std::move(new_level_name)},
       join_{make_join_or_none(g, std::make_index_sequence<N>{})},
       unfold_{
@@ -129,15 +131,15 @@ namespace phlex::experimental {
     ~unfold_node() { report_cached_stores(stores_); }
 
   private:
-    tbb::flow::receiver<message>& port_for(specified_label const& product_label) override
+    tbb::flow::receiver<message>& port_for(product_query const& product_label) override
     {
       return receiver_for<N>(join_, input(), product_label);
     }
     std::vector<tbb::flow::receiver<message>*> ports() override { return input_ports<N>(join_); }
 
     tbb::flow::sender<message>& sender() override { return output_port<0>(unfold_); }
     tbb::flow::sender<message>& to_output() override { return sender(); }
-    qualified_names const& output() const override { return output_; }
+    product_specifications const& output() const override { return output_; }
 
     template <std::size_t... Is>
     void call(Predicate const& predicate,
@@ -175,7 +177,7 @@ namespace phlex::experimental {
     std::size_t product_count() const final { return product_count_.load(); }
 
     input_retriever_types<InputArgs> input_{input_arguments<InputArgs>()};
-    qualified_names output_;
+    product_specifications output_;
     std::string new_level_name_;
     join_or_none_t<N> join_;
     tbb::flow::multifunction_node<messages_t<N>, messages_t<1u>> unfold_;

phlex/core/detail/filter_impl.cpp

@@ -4,9 +4,11 @@
 #include <string>
 
 namespace {
-  phlex::experimental::specified_label const output_dummy{phlex::experimental::qualified_name{
-    phlex::experimental::algorithm_name{"for_output_only", ""}, "for_output_only"}};
-  std::vector<phlex::experimental::specified_label> const for_output_only{output_dummy};
+  phlex::experimental::product_query const output_dummy{phlex::experimental::product_specification{
+    phlex::experimental::algorithm_name{"for_output_only", ""},
+    "for_output_only",
+    phlex::experimental::type_id{}}};
+  std::vector<phlex::experimental::product_query> const for_output_only{output_dummy};
 }
 
 namespace phlex::experimental {
@@ -50,7 +52,7 @@ namespace phlex::experimental {
 
   void decision_map::erase(accessor& a) { results_.erase(a); }
 
-  data_map::data_map(specified_labels const& product_names) :
+  data_map::data_map(product_queries const& product_names) :
     product_names_{&product_names}, nargs_{product_names.size()}
   {
     assert(nargs_ > 0);