Add comments

Author: knoepfel

phlex/core/multilayer_join_node.hpp

@@ -16,6 +16,33 @@ namespace phlex::experimental {
   template <typename Input>
   using multilayer_join_node_base_t = tbb::flow::composite_node<Input, std::tuple<Input>>;
 
+  // A multilayer_join_node is a TBB composite flow-graph node that collects one message
+  // from each of its N input streams and forwards the complete tuple downstream once all
+  // N messages for the same data unit have arrived.
+  //
+  // Each input stream is associated with a named hierarchy layer.  When two or more
+  // inputs belong to *different* layers a repeater_node is inserted in front of each
+  // stream so that a product originating at a parent layer is repeated for every
+  // child-layer data unit, allowing the underlying join_node to see a matching message on
+  // every port.  When all inputs share the same layer, repeaters are unnecessary and the
+  // join_node is used directly.
+  //
+  // Schematic with N inputs spanning multiple distinct layers:
+  //
+  //      data[0] ───┐
+  //     flush[0] ───┤ repeater[0] ──┐
+  //     index[0] ───┘               │
+  //                ⋮                ├──► join_node ──► message tuple
+  //    data[N-1] ───┐               │
+  //   flush[N-1] ───┤ repeater[N-1] ┘
+  //   index[N-1] ───┘
+  //
+  // When all inputs share the same layer the repeaters are omitted:
+  //
+  //     data[0] ──┐
+  //               ├──► join_node ──► message tuple
+  //   data[N-1] ──┘
+
   template <std::size_t n_inputs>
     requires(n_inputs > 1)
   class multilayer_join_node : public multilayer_join_node_base_t<message_tuple<n_inputs>> {
@@ -28,6 +55,10 @@ namespace phlex::experimental {
     template <std::size_t... Is>
     static auto make_join(tbb::flow::graph& g, std::index_sequence<Is...>)
     {
+      // tag_matching causes TBB to group messages by the value returned by
+      // message_matcher, which extracts the message ID.  Messages with the same ID across
+      // all ports are forwarded together as one tuple, ensuring that each output contains
+      // exactly the messages that belong to the same data unit.
       return tbb::flow::join_node<args_t, tbb::flow::tag_matching>{
         g, type_t<message_matcher, Is>{}...};
     }
@@ -43,10 +74,12 @@ namespace phlex::experimental {
     {
       assert(n_inputs == layers_.size());
 
-      // Removes duplicates
+      // Collapse to the set of distinct layer names.  More than one distinct layer means
+      // at least one input crosses a layer boundary and therefore every input stream
+      // needs a repeater_node.
       std::set collapsed_layers{layers_.begin(), layers_.end()};
 
-      // Add repeaters only if there are non-duplicate layer specifications
+      // Add repeaters only if the inputs span more than one distinct layer.
       if (collapsed_layers.size() > 1) {
         repeaters_.reserve(n_inputs);
         for (auto const& layer : layers_) {
@@ -69,12 +102,14 @@ namespace phlex::experimental {
       set_ports(std::make_index_sequence<n_inputs>{});
     }
 
+    // Returns one named_index_port per repeater so that the index router can deliver
+    // flush and index messages to each repeater.  Returns an empty list when no repeaters
+    // were constructed (all inputs share the same layer).
     std::vector<named_index_port> index_ports()
     {
-      // Returns an empty list if no repeaters are required
       std::vector<named_index_port> result;
       result.reserve(repeaters_.size());
-      for (std::size_t i = 0; i != n_inputs; ++i) {
+      for (std::size_t i = 0; i != repeaters_.size(); ++i) {
         result.emplace_back(layers_[i], &repeaters_[i]->flush_port(), &repeaters_[i]->index_port());
       }
       return result;
@@ -88,11 +123,14 @@ namespace phlex::experimental {
   };
 
   namespace detail {
-    // Stateless placeholder for cases where no join is needed
+    // Stateless placeholder used instead of multilayer_join_node when a node has only a
+    // single input (no joining is required).
     struct no_join {
       named_index_ports index_ports() const { return {}; }
     };
 
+    // Maps the number of inputs to the appropriate join type: a real multilayer_join_node
+    // for N > 1, or the no_join sentinel for N == 1.
     template <std::size_t N>
     struct pre_node {
       using type = multilayer_join_node<N>;
@@ -104,9 +142,11 @@ namespace phlex::experimental {
     };
   }
 
+  // Resolves to multilayer_join_node<N> for N > 1 and to no_join for N == 1.
   template <std::size_t N>
   using join_or_none_t = typename detail::pre_node<N>::type;
 
+  // Constructs a multilayer_join_node when N > 1, or a no_join when N == 1.
   template <std::size_t N>
   join_or_none_t<N> make_join_or_none(tbb::flow::graph& g,
                                       std::string const& node_name,
@@ -119,6 +159,9 @@ namespace phlex::experimental {
     }
   }
 
+  // Translates a runtime port index into a reference to the corresponding compile-time
+  // input port by recursively incrementing the compile-time parameter I until it matches
+  // the runtime index.
   template <std::size_t I, std::size_t N>
   tbb::flow::receiver<message>& receiver_for(multilayer_join_node<N>& join, std::size_t const index)
   {
@@ -132,6 +175,8 @@ namespace phlex::experimental {
   }
 
   namespace detail {
+    // Returns pointers to all N input ports of the join node.  Only valid for N > 1;
+    // callers with a single input should use the node directly.
     template <std::size_t N>
     std::vector<tbb::flow::receiver<message>*> input_ports(join_or_none_t<N>& join)
     {
@@ -142,6 +187,8 @@ namespace phlex::experimental {
       }(std::make_index_sequence<N>{});
     }
 
+    // Looks up the port index for the given product label, then returns a reference to
+    // the corresponding input port of the join node.  Only valid for N > 1.
     template <std::size_t N>
     tbb::flow::receiver<message>& receiver_for(join_or_none_t<N>& join,
                                                product_queries const& product_labels,
@@ -153,6 +200,8 @@ namespace phlex::experimental {
     }
   }
 
+  // Returns all input-port pointers for a node.  For N == 1 the node itself is the sole
+  // receiver; for N > 1 each port of the join is returned.
   template <std::size_t N, typename Node>
   std::vector<tbb::flow::receiver<message>*> input_ports(join_or_none_t<N>& join, Node& node)
   {
@@ -163,6 +212,9 @@ namespace phlex::experimental {
     }
   }
 
+  // Returns the receiver for the input port that corresponds to the given product label.  For
+  // N == 1 there is only one port so the node itself is returned; for N > 1 the port is looked
+  // up by label within the join.
   template <std::size_t N, typename Node>
   tbb::flow::receiver<message>& receiver_for(join_or_none_t<N>& join,
                                              product_queries const& product_labels,