From 4c5a0dd1f0466280b8fa55816d4bfff154c2fbce Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Tue, 13 Oct 2026 09:07:23 -0500 Subject: [PATCH 1/6] fix(core): resolve hierarchy paths and unfold parentage Normalize driver declarations to complete ancestor paths and resolve grouped unfold inputs to their compatible parent chains. Expand generated paths to a fixed point and record flush expectations per parent-path hash, rather than per layer name. Add layer-path inspection coverage. Keep join accounting and enabling multi-layer unfolds for the subsequent invocation-domain change; this establishes the hierarchy needed for #955 and #973. --- phlex/core/framework_graph.cpp | 37 +++---- phlex/core/index_router.cpp | 161 ++++++++++++++++------------- phlex/core/index_router.hpp | 96 +++++++---------- phlex/model/layer_path.cpp | 5 + phlex/model/layer_path.hpp | 6 ++ phlex/utilities/simple_ptr_map.hpp | 6 +- test/layer_path_test.cpp | 15 +++ 7 files changed, 168 insertions(+), 158 deletions(-) diff --git a/phlex/core/framework_graph.cpp b/phlex/core/framework_graph.cpp index 69f9125c0..ebeff0080 100644 --- a/phlex/core/framework_graph.cpp +++ b/phlex/core/framework_graph.cpp @@ -24,10 +24,10 @@ #include #include #include -#include #include #include #include +#include namespace phlex::detail { namespace { @@ -65,30 +65,23 @@ namespace phlex::detail { return result; } - // Collects (input layer, child layer) pairs and per-input-layer unfold counts from all - // registered unfold nodes. One pair is emitted per (unfold node, input layer): a - // multi-input unfold contributes the same child layer under each of its input layers, - // which is over-approximate but harmless — only the most-derived input actually parents - // children at runtime, and the extra synthetic paths are never reached by - // counting_layer_hashes_under for any real fold. The per-input-layer count lets - // flush_gates wait for a flush message from every unfold that consumes a given layer - // before evaluating done(). + // Keep each unfold's inputs together: only their most-derived cell parents the children + // and receives the unfold flush. Finalization resolves that parent on each applicable branch. index_router::unfold_data unfold_layers(declared_unfolds const& unfolds) { index_router::unfold_data result; - for (auto const& n : unfolds | std::views::values) { - phlex::experimental::identifier const child_layer{n->child_layer()}; - std::set input_layers_for_unfold; - for (auto const& input : n->input()) { - // Existence of layer already validated - auto const& input_layer = - static_cast(input.layer); - if (not input_layers_for_unfold.insert(input_layer).second) { - continue; - } - ++result.count_per_input_layer[input_layer]; - result.layer_pairs.push_back({.input = input_layer, .output = child_layer}); - } + result.reserve(unfolds.size()); + + using phlex::experimental::identifier; + for (auto const& [name, n] : unfolds) { + index_router::unfold_layer_spec spec{ + .name = name, + .input_layers = + std::views::transform( + n->input(), [](auto const& input) -> identifier const& { return input.layer; }) | + std::ranges::to(), + .output_layer = identifier{n->child_layer()}}; + result.push_back(std::move(spec)); } return result; } diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index d187dcc7f..f785d71e3 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -10,6 +10,7 @@ #include "phlex/utilities/bulleted_list.hpp" #include +#include #include #include @@ -22,7 +23,6 @@ #include #include #include -#include #include #include @@ -30,6 +30,42 @@ using phlex::experimental::identifier; using phlex::experimental::layer_path; namespace phlex::detail { + namespace { + // Build the complete set of paths implied by the driver's declarations by adding each path's + // intermediate parents. Incomplete paths are rooted at /job; for example, declarations + // {"/job/run/spill", "spill"} produce /job, /job/run, /job/run/spill, and /job/spill. + std::set driver_layer_paths(std::vector const& layer_paths_from_driver) + { + std::set paths{layer_path{"/job"}}; + for (auto const& path : layer_paths_from_driver) { + // Driver declarations may contain only leaves. Include their intermediate parents too. + std::vector prefix; + if (not path.is_complete()) { + prefix.emplace_back("job"); + } + for (auto const& component : path.components()) { + prefix.push_back(component); + paths.emplace(prefix); + } + } + return paths; + } + + bool matches_input_chain(layer_path const& path, std::vector const& input_layers) + { + Expects(not input_layers.empty()); + + // At least one of the input layers must equal the path's last component. + if (std::ranges::none_of(input_layers, + [&path](auto const& input) { return path.ends_with(input); })) { + return false; + } + + // Check that all input layers are present in the path. + return std::ranges::all_of(input_layers, + [&path](auto const& input) { return path.contains(input); }); + } + } //======================================================================================== // multilayer_slot implementation @@ -113,8 +149,8 @@ namespace phlex::detail { } void index_router::finalize(tbb::flow::graph& g, - std::vector layer_paths_from_driver, - unfold_data unfolds, + std::vector const& layer_paths_from_driver, + unfold_data const& unfolds, provider_input_ports_t provider_input_ports, fold_partition_ports_t fold_partition_ports, std::map const& multilayer_join_ports) @@ -123,68 +159,69 @@ namespace phlex::detail { // We must have at least one provider port, or there can be no data to process. assert(!provider_input_ports.empty()); - establish_layer_hierarchy(std::move(layer_paths_from_driver), unfolds.layer_pairs); - unfold_count_per_input_layer_ = std::move(unfolds.count_per_input_layer); + establish_layer_hierarchy(layer_paths_from_driver, unfolds); wire_provider_index_sets(g, std::move(provider_input_ports)); wire_fold_partition_index_sets(g, std::move(fold_partition_ports)); build_multilayer_join_slots(g, multilayer_join_ports); } // -------------------------------------------------------------------------------------------- - // Extend the static layer hierarchy with the dynamic paths that unfolds will produce at runtime. - // For each pair (in, out) and each known path P whose trailing segment is `in`, add a path - // `P + [out]`. Iterate to a fixed point so that chained unfolds (the output of one feeding the - // input of another) are fully expanded, but cap expansion depth at - // `max(initial_depth) + layer_pairs.size()`. Any candidate that exceeds this bound indicates - // a cyclic/misconfigured unfold pair chain; we fail fast with a diagnostic naming the offending - // pair(s). After sorting, populate is_lowest_layer_hashes_ for every path so that - // index_is_lowest_layer() can give a definitive answer without a heuristic fallback. - void index_router::establish_layer_hierarchy(std::vector layer_paths_from_driver, - std::vector const& layer_pairs) + // Resolve each unfold on compatible ancestor chains. Its deepest input is the only parent + // that receives its flush, so paths and expectations must be derived from the same application. + // Append the output layer to each matching parent path and iterate to a fixed point, allowing + // chained unfolds to resolve even when a dependent is registered before its producer. + // An acyclic chain adds at most one layer per unfold node; expansion beyond the initial deepest + // path plus the number of unfolds indicates a cycle and fails with the offending node names. + void index_router::establish_layer_hierarchy( + std::vector const& layer_paths_from_driver, unfold_data const& unfolds) { - sorted_layer_paths_ = std::move(layer_paths_from_driver); - std::size_t initial_deepest_path_depth = 0; - for (auto const& path : sorted_layer_paths_) { - initial_deepest_path_depth = std::max(path.depth(), initial_deepest_path_depth); - } - std::size_t const max_allowed_depth = initial_deepest_path_depth + layer_pairs.size(); + auto paths = driver_layer_paths(layer_paths_from_driver); + sorted_layer_paths_.assign_range(paths); + assert(!sorted_layer_paths_.empty()); + + std::size_t const initial_deepest_path_depth = + std::ranges::max_element(sorted_layer_paths_, {}, &layer_path::depth)->depth(); + std::size_t const max_allowed_depth = initial_deepest_path_depth + unfolds.size(); + std::set> counted_unfold_parents; bool changed = true; while (changed) { changed = false; - std::set offending_pairs; - for (auto const& [input_layer, output_layer] : layer_pairs) { - std::string_view const output_name = static_cast(output_layer); - // Snapshot current size: new paths appended in this loop become candidates only in the next - // fixed-point iteration. This keeps the inner loop deterministic and avoids unbounded - // growth from self-referential pairs (which shouldn't occur but would otherwise loop - // forever). + std::set offending_unfolds; + for (auto const& unfold : unfolds) { + // Snapshot the current size so paths appended for this unfold are not revisited until + // the next fixed-point iteration. In particular, self-referential unfolds stay bounded. std::size_t const snapshot = sorted_layer_paths_.size(); for (std::size_t i = 0; i < snapshot; ++i) { auto const& parent_path = sorted_layer_paths_[i]; - if (not parent_path.ends_with(input_layer)) { + if (not matches_input_chain(parent_path, unfold.input_layers)) { continue; } - layer_path candidate{parent_path.to_string() + "/" + std::string(output_name)}; + layer_path candidate{fmt::format("{}/{}", parent_path, unfold.output_layer)}; if (candidate.depth() > max_allowed_depth) { - offending_pairs.insert(fmt::format("{} -> {}", input_layer, output_layer)); + offending_unfolds.insert(unfold.name); continue; } - if (not std::ranges::contains(sorted_layer_paths_, candidate)) { + // Count each unfold once per parent across iterations, even when multiple unfolds + // produce the same child path: each still sends its own flush message. + if (counted_unfold_parents.emplace(unfold.name, parent_path.hash()).second) { + ++number_unfolds_per_parent_path_[parent_path.hash()]; + } + if (paths.insert(candidate).second) { sorted_layer_paths_.push_back(std::move(candidate)); changed = true; } } } - if (not offending_pairs.empty()) { + if (not offending_unfolds.empty()) { throw std::runtime_error(fmt::format( "Unfold layer hierarchy expansion exceeded max depth {} (initial deepest {} + {} unfold " - "pair(s)). Offending unfold pair(s):\n{}", + "node(s)). Offending unfold(s):\n{}", max_allowed_depth, initial_deepest_path_depth, - layer_pairs.size(), - bulleted_list(offending_pairs))); + unfolds.size(), + bulleted_list(offending_unfolds))); } } @@ -197,9 +234,8 @@ namespace phlex::detail { i + 1 == sorted_layer_paths_.size() or not sorted_layer_paths_[i].is_strict_prefix_of(sorted_layer_paths_[i + 1]); // Record every known layer, both lowest and non-lowest. Pre-populating the lowest entries - // lets index_is_lowest_layer() return a definitive answer without falling back to the - // unfold-name-based heuristic, which is important now that the augmentation above already - // accounts for unfold-produced layers. + // lets index_is_lowest_layer() return a definitive answer without its unknown-path fallback, + // including for unfold outputs that themselves parent another unfold's children. is_lowest_layer_hashes_.emplace(layer_hash, is_lowest_layer); } } @@ -224,24 +260,13 @@ namespace phlex::detail { } // -------------------------------------------------------------------------------------------- - // For each multi-layer join node: compute its deepest slot layer, normalize any slot whose - // counting_layer is unset to the node's deepest layer, construct the multilayer_slot objects, - // and store everything in multilayer_join_slots_. + // Resolve unset counting layers to the node's deepest slot layer. void index_router::build_multilayer_join_slots( tbb::flow::graph& g, std::map const& multilayer_join_ports) { for (auto const& [node_name, join_ports] : multilayer_join_ports) { spdlog::trace("Making multilayer slots for {}", node_name); - // Compute the node's deepest slot layer. The slot whose routing layer equals the deepest is - // the one that drives the join's tag stream: every other slot in the node will be triggered - // (and have its counter incremented) once per cell at that deepest layer. A slot's flush - // count must therefore balance against the deepest layer's child count under the routed - // partition path, not against its own routing layer. We normalize here so that join-node - // authors don't have to reason about depth: any slot that leaves `counting_layer` - // unset gets the node's deepest layer instead. Slots that explicitly choose a different - // counting layer (e.g. fold_join_node's partition slot, which selects the fold's data - // input layer) are left untouched. std::vector slot_layers; slot_layers.reserve(join_ports.size()); for (auto const& port : join_ports) { @@ -311,11 +336,10 @@ namespace phlex::detail { return it->second; } - // Unknown layer hash: establish_layers() augments sorted_layer_paths_ with every (driver path, - // unfold-produced descendant) so a hash absent from is_lowest_layer_hashes_ corresponds to a - // layer the router was never told about. Treating it as lowest is the safe default — skips - // the rollup/expected-count bookkeeping that requires path knowledge — and matches the prior - // behavior for unfold output layers. + // Unknown layer hash: establish_layer_hierarchy() includes driver paths and unfold-produced + // descendants, so a hash absent from is_lowest_layer_hashes_ corresponds to a layer the router + // was never told about. Treating it as lowest skips the rollup/expected-count bookkeeping that + // requires path knowledge and matches the prior behavior for unfold output layers. return is_lowest_layer_hashes_.emplace(index->layer_hash(), true).first->second; } @@ -401,9 +425,7 @@ namespace phlex::detail { // Message entries: All slots from a node are appended if at least one slot exactly matches the // current layer and every slot either exactly matches or is a parent of the routed index. // - // End-token entries: For each exactly-matching fold partition slot, append an entry for every - // descendant of layer_path whose trailing layer equals the counting layer. No entry is needed - // when the counting layer equals the routing layer because that slot is pass-through. + // End-token entries select descendants ending in the slot's counting layer. auto index_router::resolve_join_slots(data_cell_index_ptr const& index, layer_path const& layer_path, internal::join_node_slots const& node_slots) const @@ -425,10 +447,10 @@ namespace phlex::detail { if (slot->matches_exactly(layer_path)) { has_exact_match = true; if (flush.counting_layer != slot->layer()) { - // Enumerate descendants under the routed partition path at the counting layer. - auto const hashes = counting_layer_hashes_under(layer_path, flush.counting_layer); - for (auto const& h : hashes) { - end_token_entries.push_back({.counting_layer_hash = h, .flush_port = flush.flush_port}); + auto hashes = counting_layer_hashes_under(layer_path, flush.counting_layer); + for (auto const hash : hashes) { + end_token_entries.push_back( + {.counting_layer_hash = hash, .flush_port = flush.flush_port}); } } message_slots.push_back(slot); @@ -456,7 +478,6 @@ namespace phlex::detail { if (not partition_layer_path.is_strict_prefix_of(candidate)) { continue; } - // ...and its trailing segment must equal the counting layer name if (not candidate.ends_with(counting_layer_name)) { continue; } @@ -467,17 +488,11 @@ namespace phlex::detail { identifier index_router::deepest_layer_name(std::vector const& layer_names) const { - // Compute the maximum depth (= max path length) of any registered path whose trailing segment - // matches each layer name. The layer name with the greatest such depth is the most-derived - // one. Ties resolve lexicographically on the name itself for determinism — in practice ties - // only occur when all slots are at the same depth, in which case any tie winner is equivalent - // and the tie resolution doesn't matter. auto depth_of = [this](identifier const& name) -> std::size_t { std::size_t best = 0; for (auto const& path : sorted_layer_paths_) { if (path.ends_with(name)) { - std::size_t const depth = path.depth(); - best = std::max(depth, best); + best = std::max(path.depth(), best); } } return best; @@ -550,8 +565,8 @@ namespace phlex::detail { // of them before it can evaluate done(). Without this, the first unfold to finish could // cause the gate to fire before the others have reported their counts. std::size_t const expected_flush_count = [&]() -> std::size_t { - auto it = unfold_count_per_input_layer_.find(index->layer_name()); - return it != unfold_count_per_input_layer_.end() ? it->second : 0; + auto it = number_unfolds_per_parent_path_.find(index->layer_hash()); + return it != number_unfolds_per_parent_path_.end() ? it->second : 0; }(); a->second = std::make_shared(index, expected_flush_count); } diff --git a/phlex/core/index_router.hpp b/phlex/core/index_router.hpp index 8529cf517..be8bd275d 100644 --- a/phlex/core/index_router.hpp +++ b/phlex/core/index_router.hpp @@ -17,6 +17,7 @@ #include #include #include +#include namespace phlex::detail { namespace internal { @@ -26,19 +27,16 @@ namespace phlex::detail { // ========================================================================================== // A multilayer_slot (one per registered named_index_port) captures the routing decision // for one input slot of a multi-layer join node. See the implementation file for the full - // description. The slot is concerned purely with message routing — flush-side metadata is + // description. The slot is concerned purely with message routing; flush-side metadata is // carried separately in a paired flush_spec (see below) so that the slot need not know // about counting layers or flush ports. class multilayer_slot; using multilayer_slots = std::vector>; using multilayer_slots_ptr = std::shared_ptr; - // The flush-side metadata for one registered named_index_port, paired positionally - // with the corresponding multilayer_slot in join_node_slots (below). When the - // paired slot exactly matches a routed partition index, the router materializes one - // end_token_entry per descendant of the routed path whose trailing layer name equals - // `counting_layer`; the entry's count is later drawn from the gate's - // committed_counts_ at the resolved hash and forwarded to `flush_port`. + // Flush metadata paired positionally with the routing slot in join_node_slots. + // Descendants ending in the counting layer supply the committed counts forwarded + // individually as indexed_end_tokens to flush_port. struct flush_spec { phlex::experimental::identifier counting_layer; tbb::flow::receiver* flush_port; @@ -52,10 +50,7 @@ namespace phlex::detail { std::vector flush_specs; }; - // An end_token_entry binds a downstream flush_port to a specific path-aware - // committed-counts entry that the partition flush must subtract on the receiving side. - // The router materializes one entry per `(slot, counting-layer descendant path)` - // pair when resolving end tokens for a routed partition index. + // Each entry identifies one descendant count in the partition gate's committed_counts_. struct end_token_entry { std::size_t counting_layer_hash; tbb::flow::receiver* flush_port; @@ -87,38 +82,29 @@ namespace phlex::detail { }; using fold_partition_ports_t = std::map; - // Pairs an unfold's input layer name with the child layer name it produces. These - // pairs let establish_layers() extend the static layer hierarchy with the dynamic - // paths that unfolds introduce at runtime, so that downstream path-aware hash - // resolution (e.g. counting_layer_hashes_under) can discover unfold-produced layers. - struct unfold_layer_pair { - phlex::experimental::identifier input; - phlex::experimental::identifier output; + // Keeps an unfold's input layers together with the child layer it produces. Only the deepest + // input on a compatible ancestor chain parents the children and receives the unfold's flush. + // This metadata lets the router discover generated paths before any runtime indices arrive. + struct unfold_layer_spec { + std::string name; + std::vector input_layers; + phlex::experimental::identifier output_layer; }; explicit index_router(tbb::flow::graph& g); data_cell_index_ptr route(data_cell_index_ptr const& index, index_flushes const& flushes); - // Establishes the layer hierarchy, registers unfold metadata, and wires all TBB graph - // edges needed before execution. - // - // `layer_paths_from_driver` supplies the static layer paths (e.g. /job, /job/event). - // `unfold_layer_pairs` lets the router extend those paths with the dynamic layers - // produced by unfolds. For each pair (in, out) and each known path whose trailing - // segment is `in`, a synthetic path with `out` appended is added to the hierarchy. - // The augmentation is iterated to a fixed point so that chained unfolds (out of one - // feeds into another) are fully resolved. - // `unfold_count_per_input_layer` registers how many unfolds produce children from each - // input layer, so that flush_gates are initialized with the correct expected child count - // when they are first created. - struct unfold_data { - std::vector layer_pairs; - std::map count_per_input_layer; - }; + using unfold_data = std::vector; + // Establishes the layer hierarchy, registers unfold flush expectations, and wires the TBB + // graph edges needed before execution. Driver declarations supply the static paths; their + // intermediate parents and implicit /job roots are included in the hierarchy. + // Unfold-generated paths and per-parent flush expectations are resolved together to a fixed + // point so chained unfolds are handled regardless of registration order. Each parent gate + // then waits for every unfold that produces children from that particular hierarchy path. void finalize(tbb::flow::graph& g, - std::vector layer_paths_from_driver, - unfold_data unfolds, + std::vector const& layer_paths_from_driver, + unfold_data const& unfolds, provider_input_ports_t provider_input_ports, fold_partition_ports_t fold_partition_ports, std::map const& multilayer_join_ports); @@ -138,17 +124,15 @@ namespace phlex::detail { bool is_lowest_layer, std::size_t message_id); bool index_is_lowest_layer(data_cell_index_ptr const& index); - // Hash-only lookup, intended for classifying child layer hashes that arrive in flush - // messages (where only the hash is available, not a data_cell_index). Returns the - // cached classification when known; defaults to lowest for unknown hashes, which is - // correct for unfold outputs (the only source of unknown hashes) and consistent with - // index_is_lowest_layer()'s fall-through default. + // Hash-only lookup for child paths from flush messages, where no data_cell_index is available. + // Returns the cached classification for known paths; unknown hashes default to lowest, + // consistently with index_is_lowest_layer()'s fallback. bool is_lowest_layer_hash(std::size_t layer_hash) const; // finalize() helpers — each owns one initialization step. void establish_layer_hierarchy( - std::vector layer_paths_from_driver, - std::vector const& layer_pairs); + std::vector const& layer_paths_from_driver, + unfold_data const& unfolds); void wire_provider_index_sets(tbb::flow::graph& g, provider_input_ports_t provider_input_ports); void wire_fold_partition_index_sets(tbb::flow::graph& g, fold_partition_ports_t fold_partition_ports); @@ -172,22 +156,11 @@ namespace phlex::detail { flush_gate_ptr gate_for(data_cell_index_ptr const& index); void flush_if_done(data_cell_index_ptr index); - // Returns one path-aware layer_hash per static-hierarchy layer path whose trailing - // segment equals `counting_layer_name` and whose path lies strictly under - // `partition_layer_path`. Used by `multilayer_slots_for` to materialize one - // end_token_entry per descendant counting-layer path when a slot's counting layer - // differs from its routing layer. + // Resolve descendant paths ending in the counting layer beneath this partition. std::vector counting_layer_hashes_under( phlex::experimental::layer_path const& partition_layer_path, phlex::experimental::identifier const& counting_layer_name) const; - // Returns the deepest (most-derived) layer name among `layer_names` according to the - // static hierarchy in `sorted_layer_paths_`. Depth is determined by the maximum path - // length of any registered path whose trailing segment equals the layer name; the - // name with the greatest such depth wins, with ties broken by lexicographic order on - // the layer name itself for determinism. Layer names that aren't present in - // `sorted_layer_paths_` (rare — should only occur for genuinely-unknown layers) - // contribute a depth of 0 and are therefore never selected when a known peer exists. phlex::experimental::identifier deepest_layer_name( std::vector const& layer_names) const; @@ -195,7 +168,7 @@ namespace phlex::detail { tbb::flow::function_node unfold_flush_receiver_; std::atomic received_indices_; tbb::concurrent_unordered_map is_lowest_layer_hashes_; - // Layer paths from the driver, sorted lexicographically. Used to resolve a slot's + // Driver and unfold paths, sorted lexicographically. Used to resolve a slot's // counting-layer name into the set of path-aware layer hashes that the flush gate // will populate, when the counting layer differs from the routing layer. std::vector sorted_layer_paths_; @@ -217,8 +190,8 @@ namespace phlex::detail { multilayer_join_slots_; // This struct lets multilayer_slots_for return message slots and end-token entries together, - // instead of passing concurrent_hash_map accessors as output parameters. End-token entries are - // generated only for fold partition slots: one entry per descendant path at the counting layer. + // instead of passing concurrent_hash_map accessors as output parameters. End-token entries + // contain one combined descendant count per receiving slot. struct multilayer_slot_cache_entry { internal::multilayer_slots_ptr message_slots; internal::end_token_entries_ptr end_token_entries; @@ -237,9 +210,10 @@ namespace phlex::detail { using const_accessor = gates_t::const_accessor; gates_t flush_gates_; - // Number of unfolds that will send flush messages for each input layer. Used to - // initialize flush_gates with the correct expected child count. - std::map unfold_count_per_input_layer_; + // Layer-path hashes, not bare names: repeated names can have different unfold parents. + // These counts are passed to flush_gate; see its expected_flush_count documentation for why + // a gate may need to wait for flushes from multiple unfolds. + std::map number_unfolds_per_parent_path_; }; } diff --git a/phlex/model/layer_path.cpp b/phlex/model/layer_path.cpp index 4aaf4232d..fbd584e06 100644 --- a/phlex/model/layer_path.cpp +++ b/phlex/model/layer_path.cpp @@ -79,6 +79,11 @@ namespace phlex::experimental { return layer_path_.back() == name; } + bool layer_path::contains(identifier const& name) const noexcept + { + return std::ranges::contains(layer_path_, name); + } + std::string layer_path::to_string() const { return fmt::format("{}{}", is_complete() ? "/" : "", fmt::join(layer_path_, "/")); diff --git a/phlex/model/layer_path.hpp b/phlex/model/layer_path.hpp index 6fb38675b..af8ee0838 100644 --- a/phlex/model/layer_path.hpp +++ b/phlex/model/layer_path.hpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -36,6 +37,11 @@ namespace phlex::experimental { bool ends_with(identifier const& name) const noexcept; + /// Check stored components only, excluding any implicit job root. + bool contains(identifier const& name) const noexcept; + + std::span components() const noexcept { return layer_path_; } + std::string to_string() const; /// This function assumes incomplete paths have an implicit job root diff --git a/phlex/utilities/simple_ptr_map.hpp b/phlex/utilities/simple_ptr_map.hpp index beb4e49a7..e1c5ed1f3 100644 --- a/phlex/utilities/simple_ptr_map.hpp +++ b/phlex/utilities/simple_ptr_map.hpp @@ -11,6 +11,7 @@ // - try_emplace(string, Ptr) // - begin() // - end() +// - size() // - get(std::string const& key) [[not provided by std::map]] // // The 'get(...') function returns a bare pointer to the stored element if it exists; @@ -44,8 +45,9 @@ namespace phlex::detail { return data_.try_emplace(std::move(node_name), std::move(node_ptr)); } - auto begin() const { return data_.begin(); } - auto end() const { return data_.end(); } + auto begin() const noexcept { return data_.begin(); } + auto end() const noexcept { return data_.end(); } + auto size() const noexcept { return data_.size(); } ptr::element_type* get(std::string const& node_name) const { diff --git a/test/layer_path_test.cpp b/test/layer_path_test.cpp index 6919a68cc..ebac51d36 100644 --- a/test/layer_path_test.cpp +++ b/test/layer_path_test.cpp @@ -28,6 +28,21 @@ TEST_CASE("Layer path tests", "[layer_path]") CHECK(event.ends_with(partial_event)); CHECK_FALSE(subrun.ends_with(partial_event)); + CHECK(event.contains("job")); + CHECK(event.contains("run")); + CHECK(event.contains(event_id)); + CHECK_FALSE(event.contains("lumiblock")); + CHECK(partial_event.contains("subrun")); + CHECK(partial_event.contains(event_id)); + CHECK_FALSE(partial_event.contains("job")); + + auto const components = event.components(); + REQUIRE(components.size() == 4); + CHECK(components[0] == "job"); + CHECK(components[1] == "run"); + CHECK(components[2] == "subrun"); + CHECK(components[3] == "event"); + CHECK(subrun.to_string() == "/job/run/subrun"); auto event_hashes = event.hashes(); From 058e70cbc53cc06373720863485763f5f4532a99 Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Tue, 13 Oct 2026 09:12:45 -0500 Subject: [PATCH 2/6] fix(core): make partition completion aggregate and order-independent Deliver one completion token per receiving slot, summing all applicable descendant-path counts so a partial contribution cannot retire cached state prematurely. Retain zero-count completion state until the product arrives. Cover combined counts across repeated layer names, empty unfold branches, and both product/completion arrival orders. These completion-protocol prerequisites support #955 and #973. --- phlex/core/detail/repeater_node.cpp | 5 +- phlex/core/index_router.cpp | 12 +- phlex/core/index_router.hpp | 8 +- phlex/core/message.hpp | 5 +- test/CMakeLists.txt | 9 ++ test/fold_duplicate_layer_name_test.cpp | 184 ++++++++++++++++++------ test/index_router_test.cpp | 66 +++++++++ test/repeater_node_test.cpp | 26 ++-- 8 files changed, 254 insertions(+), 61 deletions(-) create mode 100644 test/index_router_test.cpp diff --git a/phlex/core/detail/repeater_node.cpp b/phlex/core/detail/repeater_node.cpp index a8aecb4e6..d01eec671 100644 --- a/phlex/core/detail/repeater_node.cpp +++ b/phlex/core/detail/repeater_node.cpp @@ -191,7 +191,10 @@ namespace phlex::detail::internal { assert(entry->data_msg); output_port<0>(repeater_).try_put(*entry->data_msg); cached_products_.erase(a); - } else if (entry->flush_received.test() and entry->pending_invocations == 0) { + } else if (entry->flush_received.test() and entry->pending_invocations == 0 and + entry->data_msg) { + // A zero-count flush can precede the product. Keep its completion state until the + // product arrives, otherwise that late product would be cached without a future flush. cached_products_.erase(a); } } diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index f785d71e3..0f4f2a3d2 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -8,6 +8,7 @@ #include "phlex/model/identifier.hpp" #include "phlex/model/layer_path.hpp" #include "phlex/utilities/bulleted_list.hpp" +#include "phlex/utilities/signed_size.hpp" #include #include @@ -317,7 +318,10 @@ namespace phlex::detail { gate_for(index)->set_flush_callback( [end_token_entries = std::move(end_token_entries)](flush_gate const& fc) { for (auto const& entry : *end_token_entries) { - auto const count = fc.committed_count_for_layer(entry.counting_layer_hash); + signed_size_t count = 0; + for (auto const hash : entry.counting_layer_hashes) { + count += fc.committed_count_for_layer(hash); + } entry.flush_port->try_put({.index = fc.index(), .count = count}); } }); @@ -425,7 +429,7 @@ namespace phlex::detail { // Message entries: All slots from a node are appended if at least one slot exactly matches the // current layer and every slot either exactly matches or is a parent of the routed index. // - // End-token entries select descendants ending in the slot's counting layer. + // End-token entries combine descendants ending in the slot's counting layer. auto index_router::resolve_join_slots(data_cell_index_ptr const& index, layer_path const& layer_path, internal::join_node_slots const& node_slots) const @@ -448,9 +452,9 @@ namespace phlex::detail { has_exact_match = true; if (flush.counting_layer != slot->layer()) { auto hashes = counting_layer_hashes_under(layer_path, flush.counting_layer); - for (auto const hash : hashes) { + if (not hashes.empty()) { end_token_entries.push_back( - {.counting_layer_hash = hash, .flush_port = flush.flush_port}); + {.counting_layer_hashes = std::move(hashes), .flush_port = flush.flush_port}); } } message_slots.push_back(slot); diff --git a/phlex/core/index_router.hpp b/phlex/core/index_router.hpp index be8bd275d..1f9957f98 100644 --- a/phlex/core/index_router.hpp +++ b/phlex/core/index_router.hpp @@ -35,8 +35,8 @@ namespace phlex::detail { using multilayer_slots_ptr = std::shared_ptr; // Flush metadata paired positionally with the routing slot in join_node_slots. - // Descendants ending in the counting layer supply the committed counts forwarded - // individually as indexed_end_tokens to flush_port. + // Descendants ending in the counting layer supply one combined completion count + // forwarded as an indexed_end_token to flush_port. struct flush_spec { phlex::experimental::identifier counting_layer; tbb::flow::receiver* flush_port; @@ -50,9 +50,9 @@ namespace phlex::detail { std::vector flush_specs; }; - // Each entry identifies one descendant count in the partition gate's committed_counts_. + // One completion token per slot combines all applicable descendant counting paths. struct end_token_entry { - std::size_t counting_layer_hash; + std::vector counting_layer_hashes; tbb::flow::receiver* flush_port; }; using end_token_entries = std::vector; diff --git a/phlex/core/message.hpp b/phlex/core/message.hpp index 6c0f5d25f..61c94b7ca 100644 --- a/phlex/core/message.hpp +++ b/phlex/core/message.hpp @@ -31,11 +31,12 @@ namespace phlex::detail { struct indexed_end_token { data_cell_index_ptr index; - // The count is the number of direct children processed for this index. It uses signed_size_t + // The count is the number of invocations covered by this partition's completion token. + // It combines all applicable descendant counting paths and uses signed_size_t // because it is subtracted from a pending-invocations counter, which may be negative when the // indexed_end_token arrives before all pending invocations are processed. // (See the pending_invocations members in repeater_node and accumulator_node.) - signed_size_t count; + signed_size_t count{}; }; struct message { diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 2a9ea79f9..506363db8 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -174,6 +174,15 @@ cet_test(data_cell_index USE_CATCH2_MAIN SOURCE data_cell_index_test.cpp LIBRARI cet_test(index_generator USE_CATCH2_MAIN SOURCE index_generator_test.cpp LIBRARIES phlex::model_internal ) +cet_test( + index_router + USE_CATCH2_MAIN + SOURCE + index_router_test.cpp + LIBRARIES + phlex::core_internal + TBB::tbb +) cet_test(source_unit USE_CATCH2_MAIN SOURCE source_test.cpp LIBRARIES phlex::core_internal ) diff --git a/test/fold_duplicate_layer_name_test.cpp b/test/fold_duplicate_layer_name_test.cpp index 659df45c7..090fa3767 100644 --- a/test/fold_duplicate_layer_name_test.cpp +++ b/test/fold_duplicate_layer_name_test.cpp @@ -34,12 +34,19 @@ #include "phlex/core/product_selector.hpp" #include "phlex/model/data_cell_index.hpp" #include "plugins/layer_generator.hpp" +#include "test/ostream_logger.hpp" +#include #include +#include +#include #include #include +#include #include +#include +#include using namespace phlex; @@ -48,48 +55,145 @@ namespace { unsigned int provide_number(data_cell_index const& index) { return index.number(); } void add(std::atomic& counter, unsigned int number) { counter += number; } + + class split { + public: + explicit split(unsigned int count) : count_{count} {} + static unsigned int initial_value() { return 0; } + bool predicate(unsigned int i) const { return i < count_; } + static auto unfold(unsigned int i) { return std::make_pair(i + 1, i + 1); } + + private: + unsigned int count_; + }; + + void check_no_cached_entries(std::ostringstream const& output) + { + auto const messages = output.str(); + CHECK_FALSE(messages.contains("Cached")); + } } TEST_CASE("Fold different layer paths with same trailing name", "[graph]") { - // job -> run -> event layers - constexpr auto index_limit = 2u; - constexpr auto number_limit = 5u; - - // job -> event layers - constexpr auto top_level_event_limit = 10u; - - auto gen = experimental::layer_generator::make(); - gen->add_layer("run", {.parent_layer = "job", .count = index_limit}); - gen->add_layer("event", {.parent_layer = "run", .count = number_limit}); - gen->add_layer("event", {.parent_layer = "job", .count = top_level_event_limit}); - - auto g = phlex::detail::framework_graph::without_driver("test"); - g.add_driver(gen); - - // Register provider - g.provide("provide_number", provide_number, concurrency::unlimited) - .output_product("input", "number", "event"); - - g.fold("run_add", add, concurrency::unlimited, "run", 0u) - .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "number"}) - .output_product_suffixes("run_sum"); - g.fold("job_add", add, concurrency::unlimited) - .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "number"}) - .output_product_suffixes("job_sum"); - - g.observe("verify_run_sum", [](unsigned int actual) { CHECK(actual == 10u); }) - .input_family(product_selector{.creator = "run_add", .layer = "run", .suffix = "run_sum"}); - g.observe("verify_job_sum", - [](unsigned int actual) { - CHECK(actual == 20u + 45u); // 20u from nested events, 45u from top-level events - }) - .input_family(product_selector{.creator = "job_add", .layer = "job", .suffix = "job_sum"}); - - g.execute(); - - CHECK(g.execution_count("run_add") == std::size_t{index_limit} * number_limit); - CHECK(g.execution_count("job_add") == (index_limit * number_limit) + top_level_event_limit); - CHECK(g.execution_count("verify_run_sum") == index_limit); - CHECK(g.execution_count("verify_job_sum") == 1); + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [] { + std::vector run_sums; + std::vector job_sums; + // job -> run -> event layers + constexpr auto number_runs = 2u; + constexpr auto number_events_per_run = 5u; + + // job -> event layers + constexpr auto number_top_level_events = 10u; + + auto gen = experimental::layer_generator::make(); + gen->add_layer("run", {.parent_layer = "job", .count = number_runs}); + gen->add_layer("event", {.parent_layer = "run", .count = number_events_per_run}); + gen->add_layer("event", {.parent_layer = "job", .count = number_top_level_events}); + + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(gen); + + // Register provider + g.provide("provide_number", provide_number, concurrency::unlimited) + .output_product("input", "number", "event"); + + g.fold("run_add", add, concurrency::unlimited, "run", 0u) + .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "number"}) + .output_product_suffixes("run_sum"); + g.fold("job_add", add, concurrency::unlimited) + .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "number"}) + .output_product_suffixes("job_sum"); + + g.observe( + "verify_run_sum", + [&run_sums](unsigned int actual) { run_sums.push_back(actual); }, + concurrency::serial) + .input_family(product_selector{.creator = "run_add", .layer = "run", .suffix = "run_sum"}); + g.observe( + "verify_job_sum", + [&job_sums](unsigned int actual) { job_sums.push_back(actual); }, + concurrency::serial) + .input_family(product_selector{.creator = "job_add", .layer = "job", .suffix = "job_sum"}); + + g.execute(); + + CHECK(g.execution_count("run_add") == std::size_t{number_runs} * number_events_per_run); + CHECK(g.execution_count("job_add") == + (number_runs * number_events_per_run) + number_top_level_events); + CHECK(g.execution_count("verify_run_sum") == number_runs); + CHECK(g.execution_count("verify_job_sum") == 1); + return std::make_pair(std::move(run_sums), std::move(job_sums)); + }; + // Destroy the graph before checking for cached-entry warnings. + auto [run_sums, job_sums] = run_graph(); + std::ranges::sort(run_sums); + CHECK(run_sums == std::vector{10, 10}); + CHECK(job_sums == std::vector{65}); + check_no_cached_entries(output); +} + +TEST_CASE("Fold duplicate event paths with empty unfolds", "[graph][issue955]") +{ + auto const [description, top_count, run_count] = + GENERATE(Catch::Generators::table( + {{"Both paths populated", 3u, 2u}, + {"Empty top branch events", 0u, 2u}, + {"Empty nested events", 3u, 0u}, + {"Both paths empty", 0u, 0u}})); + CAPTURE(description, top_count, run_count); + + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [](unsigned int top_count, unsigned int run_count) { + std::vector sums; + auto gen = experimental::layer_generator::make(); + gen->add_layer("top", {.parent_layer = "job", .count = 1}); + gen->add_layer("run", {.parent_layer = "job", .count = 2}); + auto g = phlex::detail::framework_graph::without_driver("test", 1); + g.add_driver(gen); + g.provide( + "top_count", + [top_count](data_cell_index const&) { return top_count; }, + concurrency::unlimited) + .output_product("input", "top_count", "top"); + g.provide( + "run_count", + [run_count](data_cell_index const&) { return run_count; }, + concurrency::unlimited) + .output_product("input", "run_count", "run"); + g.unfold( + "top_events", &split::predicate, &split::unfold, concurrency::unlimited, "event") + .input_family(product_selector{.creator = "input", .layer = "top", .suffix = "top_count"}) + .output_product_suffixes("top_seed"); + g.unfold( + "run_events", &split::predicate, &split::unfold, concurrency::unlimited, "event") + .input_family(product_selector{.creator = "input", .layer = "run", .suffix = "run_count"}) + .output_product_suffixes("nested_seed"); + g.provide( + "event_number", + [](data_cell_index const& index) { return static_cast(index.number() + 1); }, + concurrency::unlimited) + .output_product("input", "number", "event"); + g.fold("job_add", add, concurrency::unlimited, "job", 0u) + .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "number"}) + .output_product_suffixes("sum"); + g.observe( + "collect_sum", [&sums](unsigned int sum) { sums.push_back(sum); }, concurrency::serial) + .input_family(product_selector{.creator = "job_add", .layer = "job", .suffix = "sum"}); + g.execute(); + + CHECK(g.execution_count("top_events") == 1); + CHECK(g.execution_count("run_events") == 2); + CHECK(g.execution_count("job_add") == top_count + (2 * run_count)); + CHECK(g.execution_count("collect_sum") == 1); + return sums; + }; + // Destroy the graph before checking for cached-entry warnings. + auto const sums = run_graph(top_count, run_count); + auto const expected = (top_count * (top_count + 1) / 2) + (run_count * (run_count + 1)); + CHECK(sums == std::vector{expected}); + check_no_cached_entries(output); } diff --git a/test/index_router_test.cpp b/test/index_router_test.cpp new file mode 100644 index 000000000..6ebffbc28 --- /dev/null +++ b/test/index_router_test.cpp @@ -0,0 +1,66 @@ +#include "phlex/core/index_router.hpp" +#include "phlex/core/message.hpp" +#include "phlex/model/data_cell_counts.hpp" +#include "phlex/model/data_cell_index.hpp" +#include "phlex/model/flush_messages.hpp" +#include "phlex/model/identifier.hpp" +#include "phlex/model/layer_path.hpp" + +#include +#include +#include +#include + +#include +#include + +using namespace phlex; + +TEST_CASE("Duplicate counting paths deliver one combined end token", "[graph][issue955]") +{ + using namespace phlex::experimental::literals; + auto const [description, top_count, nested_count] = + GENERATE(Catch::Generators::table( + {{"Both event paths populated", 2u, 3u}, + {"Top-level event path empty", 0u, 3u}, + {"Nested event path empty", 2u, 0u}, + {"Both event paths empty", 0u, 0u}})); + CAPTURE(description, top_count, nested_count); + + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + tbb::flow::queue_node tokens{graph}; + detail::index_router router{graph}; + router.finalize( + graph, + {experimental::layer_path{"/job/event"}, experimental::layer_path{"/job/run/event"}}, + {}, + {{"dummy", {.input_product = {.layer = "job"}, .port = &indices}}}, + {}, + {{"fold", + {{.layer = "job"_id, + .counting_layer = "event"_id, + .token_port = &tokens, + .index_port = &indices}}}}); + + auto job = data_cell_index::job(); + auto run = job->make_child("run", 0); + router.route(job, {}); + router.route(run, {}); + auto job_counts = std::make_shared(); + job_counts->emplace(job->make_child("event", 0)->layer_hash(), top_count); + job_counts->emplace(run->layer_hash(), 1); + router.drain({{.index = job, .counts = job_counts}}); + graph.wait_for_all(); + detail::indexed_end_token token; + CHECK_FALSE(tokens.try_get(token)); + + auto run_counts = std::make_shared(); + run_counts->emplace(run->make_child("event", 0)->layer_hash(), nested_count); + router.drain({{.index = run, .counts = run_counts}}); + graph.wait_for_all(); + REQUIRE(tokens.try_get(token)); + CHECK(token.index == job); + CHECK(std::cmp_equal(token.count, top_count + nested_count)); + CHECK_FALSE(tokens.try_get(token)); +} diff --git a/test/repeater_node_test.cpp b/test/repeater_node_test.cpp index b85d9fdfd..b7fb4af95 100644 --- a/test/repeater_node_test.cpp +++ b/test/repeater_node_test.cpp @@ -343,16 +343,22 @@ TEST_CASE("Test data-only with flush count zero", "[multithreading]") repeater_test_fixture fixture{"test_repeater_data_only_flush_zero"}; auto [run1, store1] = make_run_with_product(1, 42); - // Send data only (no index messages) - fixture.put_data_message({.store = store1, .id = 0}); - fixture.wait_for_all(); - - // Consumer should not receive anything and cache should contain the product - CHECK(fixture.consumed_messages().empty()); - CHECK(fixture.cache_size() == 1); - - // Emit a flush token with count=0 — this should evict the cached product without emitting messages - fixture.put_flush_token({.index = run1, .count = 0}); + SECTION("Product before completion") + { + fixture.put_data_message({.store = store1, .id = 0}); + fixture.wait_for_all(); + CHECK(fixture.consumed_messages().empty()); + CHECK(fixture.cache_size() == 1); + fixture.put_flush_token({.index = run1, .count = 0}); + } + SECTION("Completion before product") + { + fixture.put_flush_token({.index = run1, .count = 0}); + fixture.wait_for_all(); + CHECK(fixture.consumed_messages().empty()); + CHECK(fixture.cache_size() == 1); + fixture.put_data_message({.store = store1, .id = 0}); + } fixture.wait_for_all(); CHECK(fixture.consumed_messages().empty()); From 3ec0b48756ad223393e96c880378b75f8bc8af17 Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Tue, 13 Oct 2026 09:14:34 -0500 Subject: [PATCH 3/6] fix(fold): require at least one data product Reject folds without product inputs, including resource-only registrations, and distinguish the product count from the total argument count in fold-join templates. Use the nonempty-input invariant to remove the empty-layer fallback. Invocation-layer selection and multi-layer fold regressions follow in the shared join-domain change. --- phlex/core/fold_join_node.hpp | 53 ++++++++++++++------------------- phlex/core/registration_api.hpp | 14 ++++++--- 2 files changed, 33 insertions(+), 34 deletions(-) diff --git a/phlex/core/fold_join_node.hpp b/phlex/core/fold_join_node.hpp index 7c02e49c1..106ee5baf 100644 --- a/phlex/core/fold_join_node.hpp +++ b/phlex/core/fold_join_node.hpp @@ -5,9 +5,9 @@ #include "phlex/core/detail/repeater_node.hpp" #include "phlex/core/message.hpp" +#include #include -#include #include #include @@ -36,28 +36,30 @@ namespace phlex::detail { // ⋮ ├──► join_node ──► (accum, data[0..N-1]) // data[N-1] ──► repeater[N-1] ────► port(join) ───┘ - template + template using partition_index_with_messages = - decltype(std::tuple_cat(std::tuple{}, message_tuple{})); + decltype(std::tuple_cat(std::tuple{}, message_tuple{})); - template + template using accumulator_with_messages = decltype(std::tuple_cat( - std::tuple>{}, message_tuple{})); + std::tuple>{}, message_tuple{})); - template + template using fold_join_node_base_t = - tbb::flow::composite_node, - std::tuple>>; - - template - class fold_join_node : public fold_join_node_base_t { - using args_t = partition_index_with_messages; - using base_t = fold_join_node_base_t; - using join_args_t = accumulator_with_messages; + tbb::flow::composite_node, + std::tuple>>; + + template + class fold_join_node : public fold_join_node_base_t { + using args_t = partition_index_with_messages; + using base_t = fold_join_node_base_t; + using join_args_t = accumulator_with_messages; using result_initializer_t = std::function(data_cell_index const&)>; using input_t = base_t::input_ports_type; using output_t = base_t::output_ports_type; + static_assert(NProducts >= 1, "fold_join_node requires at least one data product."); + template static auto make_join(tbb::flow::graph& g, std::index_sequence) { @@ -86,17 +88,17 @@ namespace phlex::detail { partition_layer_name, std::move(output), std::move(result_initializer)}, - join_{make_join(g, std::make_index_sequence{})}, + join_{make_join(g, std::make_index_sequence{})}, name_{node_name}, partition_layer_{partition_layer_name}, layers_{std::move(layer_names)} { - assert(NInputs == layers_.size()); + Expects(NProducts == layers_.size()); // We do not worry about collapsing layers for the fold-join node as the accumulator // repeater will always belong to a higher layer than (at least one of) the data products // that serve as input to the fold operation. - repeaters_.reserve(NInputs); + repeaters_.reserve(NProducts); for (auto const& layer : layers_) { repeaters_.push_back(std::make_unique(g, name_, layer)); } @@ -110,7 +112,7 @@ namespace phlex::detail { (make_edge(*repeaters_[Is], input_port(join_)), ...); }; - set_ports(std::make_index_sequence{}); + set_ports(std::make_index_sequence{}); } tbb::flow::receiver& partition_port() @@ -130,27 +132,18 @@ namespace phlex::detail { // // The first entry is the *partition slot* — it routes the partition index to the // accumulator's index_port and delivers the partition-level flush token to the - // accumulator's flush_port. Its routing layer is the partition layer (e.g. "job"), - // but its counting layer is the fold's most-derived input data layer (e.g. "event"), - // because the accumulator is incremented once per fold call (= once per cell at the - // most-derived input layer) and the flush count must balance against that. + // accumulator's flush_port. Its routing layer is the partition layer (e.g. "job"), + // but its counting layer is currently the first input data layer. // // The remaining entries are the data-input slots, one per repeater. For these the // routing layer and counting layer are the same: the repeater both routes and counts // at its own input layer. - // - // FIXME: For multi-input folds, the "most-derived" input layer is the deepest of - // `layers_` in the data hierarchy. Until the router exposes a depth comparison, we - // pick `layers_[0]`; all current tests have single-input folds so this is correct in - // practice. std::vector index_ports() { std::vector result; result.reserve(1 + repeaters_.size()); // +1 for the result repeater - phlex::experimental::identifier const counting_layer_for_partition = - layers_.empty() ? partition_layer_ : layers_[0]; result.emplace_back(partition_layer_, - counting_layer_for_partition, + layers_[0], &result_repeater_.flush_port(), &result_repeater_.index_port()); for (std::size_t i = 0; i != repeaters_.size(); ++i) { diff --git a/phlex/core/registration_api.hpp b/phlex/core/registration_api.hpp index 7c275c316..763f6b693 100644 --- a/phlex/core/registration_api.hpp +++ b/phlex/core/registration_api.hpp @@ -261,7 +261,8 @@ namespace phlex::detail { static constexpr auto num_inputs = AlgorithmBits::number_inputs; static constexpr auto num_outputs = 1; // For now - static_assert(num_inputs > 1, "input_family requires at least one product selector."); + static_assert(num_inputs > 1, + "A fold algorithm requires the accumulator and at least one data product."); public: fold_api(configuration const* config, @@ -291,11 +292,16 @@ namespace phlex::detail { template auto input_family(std::array input_args, std::tuple) { + static_assert(NProducts >= 1, + "A fold requires at least one data product in addition to the accumulator."); static_assert((is_resource && ...), "The second argument to input_family(...) must be a tuple of resources."); - static_assert(num_inputs - 1 == NProducts + sizeof...(Resources), - "The number of fold product parameters is not the same as the number of " - "specified input arguments."); + static_assert( + num_inputs - 1 == NProducts + sizeof...(Resources), + "The number of product selectors and resources specified to input_family(...) must match " + "the number of data products and resources expected by the fold algorithm. The first " + "parameter of a fold algorithm is always the accumulator, followed by one or more data " + "products, followed by any resources required by the algorithm."); using product_parameter_types = boost::mp11::mp_take_c; populate_types(input_args); From 60cc4e6bb31b71c2f709dcbca20159e47decb755 Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Tue, 13 Oct 2026 09:15:55 -0500 Subject: [PATCH 4/6] fix(core): unify join routing and completion accounting Replace independent slot counting-layer preferences with product-driven join invocation domains and participants declaring their storage layers and receivers. Use the same eligible paths for routing and completion, respecting fold partitions and nearest-ancestor projection. Complete cached ancestor products even for excluded cells with zero invocations, and reject domains requiring mixed caching/pass-through roles. Enable multi-layer unfolds now that their hierarchy and completion accounting agree. Cover fold input ordering, nested unfolds, empty partitions, excluded branches, repeated layer names, leaf-only hierarchies, and unsupported participant roles. Fixes #955. Fixes #973. --- phlex/core/CMakeLists.txt | 1 + phlex/core/declared_unfold.cpp | 14 - phlex/core/fold_join_node.hpp | 30 +- phlex/core/fold_node.hpp | 3 +- phlex/core/index_router.cpp | 246 ++++++-------- phlex/core/index_router.hpp | 47 +-- phlex/core/join_registration.hpp | 29 ++ phlex/core/make_computational_edges.cpp | 9 +- phlex/core/make_computational_edges.hpp | 3 +- phlex/core/message.hpp | 22 -- phlex/core/multilayer_join_node.hpp | 19 +- phlex/core/observer_node.hpp | 3 +- phlex/core/predicate_node.hpp | 3 +- phlex/core/products_consumer.hpp | 3 +- phlex/core/transform_node.hpp | 3 +- phlex/core/unfold_node.hpp | 3 +- test/fold_test.cpp | 113 ++++++ test/framework_graph_test.cpp | 105 ++++++ test/index_router_test.cpp | 264 +++++++++++++- test/multilayer_join_test.cpp | 9 +- test/unfold_test.cpp | 435 ++++++++++++++++++------ 21 files changed, 1006 insertions(+), 358 deletions(-) create mode 100644 phlex/core/join_registration.hpp diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt index 0496d0e4e..ab7b67518 100644 --- a/phlex/core/CMakeLists.txt +++ b/phlex/core/CMakeLists.txt @@ -61,6 +61,7 @@ install( glue.hpp graph_proxy.hpp input_arguments.hpp + join_registration.hpp message.hpp multilayer_join_node.hpp index_router.hpp diff --git a/phlex/core/declared_unfold.cpp b/phlex/core/declared_unfold.cpp index 0407b0059..066d37435 100644 --- a/phlex/core/declared_unfold.cpp +++ b/phlex/core/declared_unfold.cpp @@ -9,17 +9,11 @@ #include "phlex/model/products.hpp" #include "phlex/utilities/hashing.hpp" -#include -// IWYU wants to remove the include below, but it is needed for the fmt::format call. -#include // IWYU pragma: keep #include #include #include #include -#include -#include -#include #include #include #include @@ -60,14 +54,6 @@ namespace phlex::detail { require_layers::always}, child_layer_{std::move(child_layer)} { - // Forbid multi-layer inputs until Phlex v0.5. - std::set collapsed_layers{std::from_range, layers()}; - if (collapsed_layers.size() > 1u) { - throw std::invalid_argument( - fmt::format("Unfolds cannot yet accept multi-layer inputs (node: {}, layers: {})", - this->name().to_string(), - collapsed_layers)); - } } declared_unfold::~declared_unfold() = default; diff --git a/phlex/core/fold_join_node.hpp b/phlex/core/fold_join_node.hpp index 106ee5baf..a115d14a2 100644 --- a/phlex/core/fold_join_node.hpp +++ b/phlex/core/fold_join_node.hpp @@ -3,6 +3,7 @@ #include "phlex/core/detail/accumulator_node.hpp" #include "phlex/core/detail/repeater_node.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include @@ -128,27 +129,18 @@ namespace phlex::detail { return input_port<3>(result_repeater_); } - // Returns one named_index_port per slot the index router must wire up. - // - // The first entry is the *partition slot* — it routes the partition index to the - // accumulator's index_port and delivers the partition-level flush token to the - // accumulator's flush_port. Its routing layer is the partition layer (e.g. "job"), - // but its counting layer is currently the first input data layer. - // - // The remaining entries are the data-input slots, one per repeater. For these the - // routing layer and counting layer are the same: the repeater both routes and counts - // at its own input layer. - std::vector index_ports() + // The partition constrains the product-driven invocation domain. Its initialization + // connection remains separate from the accumulator's per-invocation index receiver. + // Ancestor data repeaters and the accumulator complete against the same invocations. + join_registration index_ports() { - std::vector result; - result.reserve(1 + repeaters_.size()); // +1 for the result repeater - result.emplace_back(partition_layer_, - layers_[0], - &result_repeater_.flush_port(), - &result_repeater_.index_port()); + join_registration result{.input_layers = layers_, .participants = {}}; + result.participants.reserve(1 + repeaters_.size()); // +1 for the result repeater + result.participants.emplace_back( + partition_layer_, &result_repeater_.index_port(), &result_repeater_.flush_port()); for (std::size_t i = 0; i != repeaters_.size(); ++i) { - result.emplace_back( - layers_[i], layers_[i], &repeaters_[i]->flush_port(), &repeaters_[i]->index_port()); + result.participants.emplace_back( + layers_[i], &repeaters_[i]->index_port(), &repeaters_[i]->flush_port()); } return result; } diff --git a/phlex/core/fold_node.hpp b/phlex/core/fold_node.hpp index fe8499f18..7ba31d706 100644 --- a/phlex/core/fold_node.hpp +++ b/phlex/core/fold_node.hpp @@ -8,6 +8,7 @@ #include "phlex/core/fold_join_node.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/node_builder.hpp" #include "phlex/core/product_selector.hpp" @@ -129,7 +130,7 @@ namespace phlex::detail { tbb::flow::sender& output_port() override { return join_.output_port(); } phlex::experimental::product_specifications const& output() const override { return output_; } - named_index_ports index_ports() final { return join_.index_ports(); } + join_registration index_ports() final { return join_.index_ports(); } std::size_t num_calls() const final { return calls_.load(); } std::size_t apply_fold( diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index 0f4f2a3d2..8d4beafd6 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -1,5 +1,6 @@ #include "phlex/core/index_router.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/model/data_cell_index.hpp" #include "phlex/model/flush_gate.hpp" @@ -66,46 +67,62 @@ namespace phlex::detail { return std::ranges::all_of(input_layers, [&path](auto const& input) { return path.contains(input); }); } + + // Select strict descendant invocation paths whose nearest-name projection is this + // partition. Full path hashes keep repeated names on branches and within chains distinct. + std::vector counting_layer_hashes_under( + layer_path const& partition_layer_path, internal::join_node_slots const& node_slots) + { + std::vector result; + for (auto const& candidate : node_slots.invocation_paths) { + // candidate must be a strict descendant of the partition layer path + if (not partition_layer_path.is_strict_prefix_of(candidate)) { + continue; + } + // Routing projects to the nearest ancestor with this storage name. A deeper + // occurrence shadows this partition, which therefore owns none of its invocations. + auto const descendants = candidate.components().subspan(partition_layer_path.depth()); + if (std::ranges::contains(descendants, partition_layer_path.components().back())) { + continue; + } + result.push_back(candidate.hash()); + } + return result; + } } //======================================================================================== // multilayer_slot implementation // - // A multilayer_slot is a static, pre-registered routing component created once per - // `named_index_port` in `finalize()`. Each slot is responsible *only for message routing*: it - // decides whether a routed index covers this slot (via `matches_exactly` / `is_parent_of`) and - // forwards index messages to a single downstream `input_port` at the appropriate layer. - // - // The flush-side metadata for the same `named_index_port` — namely the `counting_layer` name - // used to look up `committed_counts_` entries at flush time, and the downstream `flush_port` - // that receives `indexed_end_token`s — lives in a paired `flush_spec` (see below). The two are - // stored side-by-side in `join_node_slots` so that `multilayer_slots_for` can pair them up - // while resolving end-token entries for a routed partition index. + // A static participant slot projects invocation indices to its storage layer and carries + // the receiver that completes cached products or accumulators against the join's domain. namespace internal { class multilayer_slot { public: - multilayer_slot(tbb::flow::graph& g, - identifier layer, - tbb::flow::receiver* input_port); + multilayer_slot(tbb::flow::graph& g, join_participant const& participant, bool cached); void put_message(data_cell_index_ptr const& index, std::size_t message_id); bool matches_exactly(layer_path const& path) const; - bool is_parent_of(data_cell_index_ptr const& index) const; - - identifier const& layer() const { return layer_; } + bool cached() const { return cached_; } + tbb::flow::receiver* completion_port() const { return completion_port_; } private: identifier layer_; index_set_node broadcaster_; + tbb::flow::receiver* completion_port_; + bool cached_; }; multilayer_slot::multilayer_slot(tbb::flow::graph& g, - identifier layer, - tbb::flow::receiver* input_port) : - layer_{std::move(layer)}, broadcaster_{g} + join_participant const& participant, + bool cached) : + layer_{participant.layer}, + broadcaster_{g}, + completion_port_{participant.completion_port}, + cached_{cached} { - make_edge(broadcaster_, *input_port); + make_edge(broadcaster_, *participant.index_port); } void multilayer_slot::put_message(data_cell_index_ptr const& index, std::size_t message_id) @@ -122,12 +139,42 @@ namespace phlex::detail { { return layer_path.ends_with(layer_); } + } // namespace internal + + namespace { + struct join_slot_resolution { + internal::multilayer_slots message_slots; + internal::end_token_entries end_token_entries; + }; - bool multilayer_slot::is_parent_of(data_cell_index_ptr const& index) const + // Route all participants only for an eligible invocation. At a participant's partition, + // combine the same domain's strict descendant counts into one completion token. A deepest + // product passes through and receives no token; applicable empty partitions still get zero. + join_slot_resolution resolve_join_slots(layer_path const& layer_path, + internal::join_node_slots const& node_slots) { - return index->parent(layer_) != nullptr; + auto const& slots = node_slots.slots; + internal::multilayer_slots message_slots; + internal::end_token_entries end_token_entries; + + if (std::ranges::contains(node_slots.invocation_paths, layer_path)) { + message_slots = slots; + return {.message_slots = std::move(message_slots), .end_token_entries = {}}; + } + + for (auto const& slot : slots) { + if (slot->cached() and slot->matches_exactly(layer_path)) { + // Name-based product wiring also supplies excluded storage cells. They need a + // zero completion even when no eligible invocation projects to this cell. + end_token_entries.push_back( + {.counting_layer_hashes = counting_layer_hashes_under(layer_path, node_slots), + .flush_port = slot->completion_port()}); + } + } + return {.message_slots = std::move(message_slots), + .end_token_entries = std::move(end_token_entries)}; } - } + } // namespace //======================================================================================== // index_router implementation @@ -154,7 +201,7 @@ namespace phlex::detail { unfold_data const& unfolds, provider_input_ports_t provider_input_ports, fold_partition_ports_t fold_partition_ports, - std::map const& multilayer_join_ports) + std::map const& multilayer_join_ports) { using namespace phlex::experimental::literals; // We must have at least one provider port, or there can be no data to process. @@ -261,29 +308,44 @@ namespace phlex::detail { } // -------------------------------------------------------------------------------------------- - // Resolve unset counting layers to the node's deepest slot layer. + // Product inputs drive invocations only on paths satisfying every participant's ancestry. + // Resolve the domain once so routing and completion cannot select different branches. void index_router::build_multilayer_join_slots( - tbb::flow::graph& g, std::map const& multilayer_join_ports) + tbb::flow::graph& g, std::map const& multilayer_join_ports) { - for (auto const& [node_name, join_ports] : multilayer_join_ports) { - spdlog::trace("Making multilayer slots for {}", node_name); + for (auto const& [node_name, registration] : multilayer_join_ports) { + assert(not registration.participants.empty()); + assert(not registration.input_layers.empty()); - std::vector slot_layers; - slot_layers.reserve(join_ports.size()); - for (auto const& port : join_ports) { - slot_layers.push_back(port.layer); - } - identifier const node_deepest_layer = deepest_layer_name(slot_layers); + spdlog::trace("Making multilayer slots for {}", node_name); internal::join_node_slots node_slots; - node_slots.slots.reserve(join_ports.size()); - node_slots.flush_specs.reserve(join_ports.size()); - for (auto const& [layer, counting_layer, flush_port, input_port] : join_ports) { - identifier const effective_counting_layer = counting_layer.value_or(node_deepest_layer); + for (auto const& path : sorted_layer_paths_) { + if (matches_input_chain(path, registration.input_layers) and + std::ranges::all_of(registration.participants, [&path](auto const& participant) { + return path.contains(participant.layer); + })) { + node_slots.invocation_paths.push_back(path); + } + } + node_slots.slots.reserve(registration.participants.size()); + for (auto const& participant : registration.participants) { + bool const pass_through = + std::ranges::any_of(node_slots.invocation_paths, + [&](auto const& p) { return p.ends_with(participant.layer); }); + bool const cached = std::ranges::any_of(node_slots.invocation_paths, [&](auto const& p) { + return not p.ends_with(participant.layer); + }); + // A repeater's caching mode is fixed, even when the same layer name occurs on + // multiple branches. Do not register a domain requiring both modes for one slot. + if (pass_through and cached) { + throw std::runtime_error( + fmt::format("Join '{}' participant '{}' requires mixed caching and pass-through modes.", + node_name, + participant.layer)); + } node_slots.slots.push_back( - std::make_shared(g, layer, input_port)); - node_slots.flush_specs.push_back( - {.counting_layer = effective_counting_layer, .flush_port = flush_port}); + std::make_shared(g, participant, cached)); } multilayer_join_slots_.emplace(identifier{node_name}, std::move(node_slots)); } @@ -309,9 +371,13 @@ namespace phlex::detail { slot->put_message(index, message_id); } - // Lowest-layer indices have no flush gate and contribute to their parent's readiness solely - // through the expected-count message that announced them — nothing to do here for them. + // Lowest-layer indices have no flush gate. Excluded storage cells can nevertheless + // receive products in a globally cached participant, so retire those with a zero token. if (is_lowest_layer) { + for (auto const& entry : *end_token_entries) { + assert(entry.counting_layer_hashes.empty()); + entry.flush_port->try_put({.index = index, .count = 0}); + } return index; } @@ -411,7 +477,7 @@ namespace phlex::detail { for (auto const& node_slots : multilayer_join_slots_ | std::views::values) { auto [resolved_message_slots, resolved_end_token_entries] = - resolve_join_slots(index, layer_path, node_slots); + resolve_join_slots(layer_path, node_slots); message_slots.append_range(std::views::as_rvalue(resolved_message_slots)); end_token_entries.append_range(std::views::as_rvalue(resolved_end_token_entries)); } @@ -423,98 +489,6 @@ namespace phlex::detail { return {acc->second.message_slots, acc->second.end_token_entries}; } - // Resolve the message slots and end-token entries contributed by one multi-layer join node for a - // routed index. - // - // Message entries: All slots from a node are appended if at least one slot exactly matches the - // current layer and every slot either exactly matches or is a parent of the routed index. - // - // End-token entries combine descendants ending in the slot's counting layer. - auto index_router::resolve_join_slots(data_cell_index_ptr const& index, - layer_path const& layer_path, - internal::join_node_slots const& node_slots) const - -> join_slot_resolution - { - auto const& slots = node_slots.slots; - auto const& flush_specs = node_slots.flush_specs; - assert(slots.size() == flush_specs.size()); - - internal::multilayer_slots message_slots; - message_slots.reserve(slots.size()); - internal::end_token_entries end_token_entries; - - bool has_exact_match = false; - std::size_t matched_count = 0; - for (std::size_t i = 0; i != slots.size(); ++i) { - auto const& slot = slots[i]; - auto const& flush = flush_specs[i]; - if (slot->matches_exactly(layer_path)) { - has_exact_match = true; - if (flush.counting_layer != slot->layer()) { - auto hashes = counting_layer_hashes_under(layer_path, flush.counting_layer); - if (not hashes.empty()) { - end_token_entries.push_back( - {.counting_layer_hashes = std::move(hashes), .flush_port = flush.flush_port}); - } - } - message_slots.push_back(slot); - ++matched_count; - } else if (slot->is_parent_of(index)) { - message_slots.push_back(slot); - ++matched_count; - } - } - - // Add all matching slots only when the node has an exact slot and every slot matches. - if (not has_exact_match or matched_count != slots.size()) { - message_slots.clear(); - } - return {.message_slots = std::move(message_slots), - .end_token_entries = std::move(end_token_entries)}; - } - - std::vector index_router::counting_layer_hashes_under( - layer_path const& partition_layer_path, identifier const& counting_layer_name) const - { - std::vector result; - for (auto const& candidate : sorted_layer_paths_) { - // candidate must be a strict descendant of the partition layer path - if (not partition_layer_path.is_strict_prefix_of(candidate)) { - continue; - } - if (not candidate.ends_with(counting_layer_name)) { - continue; - } - result.push_back(candidate.hash()); - } - return result; - } - - identifier index_router::deepest_layer_name(std::vector const& layer_names) const - { - auto depth_of = [this](identifier const& name) -> std::size_t { - std::size_t best = 0; - for (auto const& path : sorted_layer_paths_) { - if (path.ends_with(name)) { - best = std::max(path.depth(), best); - } - } - return best; - }; - - assert(not layer_names.empty()); - auto const* result = &layer_names.front(); - std::size_t best_depth = depth_of(*result); - for (auto const& name : layer_names | std::views::drop(1)) { - auto const d = depth_of(name); - if (d > best_depth or (d == best_depth and name < *result)) { - result = &name; - best_depth = d; - } - } - return *result; - } - void index_router::update_flush_counts(index_flushes const& flushes) { for (auto const& [index, flush_counts] : flushes) { diff --git a/phlex/core/index_router.hpp b/phlex/core/index_router.hpp index 1f9957f98..82d247561 100644 --- a/phlex/core/index_router.hpp +++ b/phlex/core/index_router.hpp @@ -2,6 +2,7 @@ #define PHLEX_CORE_INDEX_ROUTER_HPP #include "phlex/core/fwd.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/model/data_cell_index.hpp" #include "phlex/model/flush_gate.hpp" @@ -25,32 +26,21 @@ namespace phlex::detail { using index_set_node_ptr = std::shared_ptr; // ========================================================================================== - // A multilayer_slot (one per registered named_index_port) captures the routing decision - // for one input slot of a multi-layer join node. See the implementation file for the full - // description. The slot is concerned purely with message routing; flush-side metadata is - // carried separately in a paired flush_spec (see below) so that the slot need not know - // about counting layers or flush ports. + // Each participant slot owns routing and its required non-owning completion receiver. class multilayer_slot; using multilayer_slots = std::vector>; using multilayer_slots_ptr = std::shared_ptr; - // Flush metadata paired positionally with the routing slot in join_node_slots. - // Descendants ending in the counting layer supply one combined completion count - // forwarded as an indexed_end_token to flush_port. - struct flush_spec { - phlex::experimental::identifier counting_layer; - tbb::flow::receiver* flush_port; - }; - - // Per multi-layer join node: the slots, and the flush metadata paired positionally - // with each slot (slots[i] corresponds to flush_specs[i]). Stored together so - // that multilayer_slots_for can iterate the two in lockstep. + // Resolve one invocation-path set from product inputs and every participant's ancestry. + // Routing and completion counts both use this set, including unfold-generated paths. struct join_node_slots { multilayer_slots slots; - std::vector flush_specs; + std::vector invocation_paths; }; // One completion token per slot combines all applicable descendant counting paths. + // The hashes identify entries in the partition gate's committed_counts_; their sum balances + // the receiving slot's pending invocations when the partition flushes. struct end_token_entry { std::vector counting_layer_hashes; tbb::flow::receiver* flush_port; @@ -107,7 +97,7 @@ namespace phlex::detail { unfold_data const& unfolds, provider_input_ports_t provider_input_ports, fold_partition_ports_t fold_partition_ports, - std::map const& multilayer_join_ports); + std::map const& multilayer_join_ports); void drain(index_flushes const& flushes); tbb::flow::function_node& unfold_index_receiver() @@ -137,18 +127,11 @@ namespace phlex::detail { void wire_fold_partition_index_sets(tbb::flow::graph& g, fold_partition_ports_t fold_partition_ports); void build_multilayer_join_slots( - tbb::flow::graph& g, std::map const& multilayer_join_ports); + tbb::flow::graph& g, std::map const& multilayer_join_ports); internal::index_set_node_ptr index_set_node_for(phlex::experimental::layer_path const& layer); internal::index_set_node_ptr index_set_node_for(data_cell_index_ptr const& index); std::pair multilayer_slots_for( data_cell_index_ptr const& index); - struct join_slot_resolution { - internal::multilayer_slots message_slots; - internal::end_token_entries end_token_entries; - }; - join_slot_resolution resolve_join_slots(data_cell_index_ptr const& index, - phlex::experimental::layer_path const& layer_path, - internal::join_node_slots const& node_slots) const; void update_flush_counts(index_flushes const& flushes); void apply_expected_count(flush_gate& gate, data_cell_index::hash_type child_layer_hash, @@ -156,21 +139,11 @@ namespace phlex::detail { flush_gate_ptr gate_for(data_cell_index_ptr const& index); void flush_if_done(data_cell_index_ptr index); - // Resolve descendant paths ending in the counting layer beneath this partition. - std::vector counting_layer_hashes_under( - phlex::experimental::layer_path const& partition_layer_path, - phlex::experimental::identifier const& counting_layer_name) const; - - phlex::experimental::identifier deepest_layer_name( - std::vector const& layer_names) const; - tbb::flow::function_node unfold_index_receiver_; tbb::flow::function_node unfold_flush_receiver_; std::atomic received_indices_; tbb::concurrent_unordered_map is_lowest_layer_hashes_; - // Driver and unfold paths, sorted lexicographically. Used to resolve a slot's - // counting-layer name into the set of path-aware layer hashes that the flush gate - // will populate, when the counting layer differs from the routing layer. + // Driver and unfold paths, sorted lexicographically, used to resolve invocation domains. std::vector sorted_layer_paths_; // ========================================================================================== diff --git a/phlex/core/join_registration.hpp b/phlex/core/join_registration.hpp new file mode 100644 index 000000000..e324e2971 --- /dev/null +++ b/phlex/core/join_registration.hpp @@ -0,0 +1,29 @@ +#ifndef PHLEX_CORE_JOIN_REGISTRATION_HPP +#define PHLEX_CORE_JOIN_REGISTRATION_HPP + +#include "phlex/core/message.hpp" +#include "phlex/model/identifier.hpp" + +#include +#include + +#include + +namespace phlex::detail { + // Required non-owning receivers for one product repeater or fold accumulator. + // The layer identifies storage and invocation-index projection, not an independent count. + struct join_participant { + phlex::experimental::identifier layer; + gsl::not_null*> index_port; + gsl::not_null*> completion_port; + }; + + // Product inputs drive invocations; all participants constrain their ancestry. + // Empty participants mean that no router registration is needed. + struct join_registration { + std::vector input_layers; + std::vector participants; + }; +} + +#endif // PHLEX_CORE_JOIN_REGISTRATION_HPP diff --git a/phlex/core/make_computational_edges.cpp b/phlex/core/make_computational_edges.cpp index 859a2d70a..9342ab370 100644 --- a/phlex/core/make_computational_edges.cpp +++ b/phlex/core/make_computational_edges.cpp @@ -3,6 +3,7 @@ #include "phlex/core/declared_output.hpp" #include "phlex/core/filter.hpp" #include "phlex/core/index_router.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/node_catalog.hpp" #include "phlex/core/producer_catalog.hpp" @@ -215,12 +216,12 @@ namespace phlex::detail { return result; } - std::map multilayer_ports( + std::map multilayer_ports( std::span consumers) { - std::map result; + std::map result; for (auto* node : consumers) { - if (auto const& ports = node->index_ports(); not ports.empty()) { + if (auto const& ports = node->index_ports(); not ports.participants.empty()) { result.try_emplace(node->name().to_string(), ports); } } @@ -242,7 +243,7 @@ namespace phlex::detail { } } - std::tuple> + std::tuple> make_computational_edges(node_catalog& nodes, std::map& filters, tbb::flow::graph& g, diff --git a/phlex/core/make_computational_edges.hpp b/phlex/core/make_computational_edges.hpp index 864bead14..d696aedc1 100644 --- a/phlex/core/make_computational_edges.hpp +++ b/phlex/core/make_computational_edges.hpp @@ -15,6 +15,7 @@ #include "phlex/core/filter.hpp" #include "phlex/core/index_router.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/node_catalog.hpp" #include "phlex/phlex_core_export.hpp" @@ -28,7 +29,7 @@ namespace phlex::detail { PHLEX_CORE_EXPORT - std::tuple> + std::tuple> make_computational_edges(node_catalog& nodes, std::map& filters, tbb::flow::graph& g, diff --git a/phlex/core/message.hpp b/phlex/core/message.hpp index 61c94b7ca..9654b34c1 100644 --- a/phlex/core/message.hpp +++ b/phlex/core/message.hpp @@ -11,10 +11,7 @@ #include "phlex/utilities/signed_size.hpp" #include "phlex/utilities/sized_tuple.hpp" -#include // <-- belongs somewhere else - #include -#include #include #include #include @@ -55,25 +52,6 @@ namespace phlex::detail { template using messages_t = std::conditional_t>; - // A named_index_port describes one input slot of a multi-layer join node from the - // perspective of the index router. Two distinct layer concepts are carried: - // - // - `layer` — the *routing* layer. The router decides whether a routed index - // feeds this slot's `index_port` (and whether the slot should - // receive a flush token) using this layer name via - // `matches_exactly` / `is_parent_of` checks. - // - // - `counting_layer` — the *counting* layer preference. `std::nullopt` selects the - // node's deepest layer, while a populated value selects that - // explicit layer name. - struct named_index_port { - phlex::experimental::identifier layer; - std::optional counting_layer; - tbb::flow::receiver* token_port; - tbb::flow::receiver* index_port; - }; - using named_index_ports = std::vector; - // Overload for use with most_derived PHLEX_CORE_EXPORT message const& more_derived(message const& a, message const& b); diff --git a/phlex/core/multilayer_join_node.hpp b/phlex/core/multilayer_join_node.hpp index 7666cbdec..71b789725 100644 --- a/phlex/core/multilayer_join_node.hpp +++ b/phlex/core/multilayer_join_node.hpp @@ -2,6 +2,7 @@ #define PHLEX_CORE_MULTILAYER_JOIN_NODE_HPP #include "phlex/core/detail/repeater_node.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/product_selector.hpp" #include "phlex/utilities/sized_tuple.hpp" @@ -9,7 +10,6 @@ #include #include -#include #include #include #include @@ -105,17 +105,14 @@ namespace phlex::detail { set_ports(std::make_index_sequence{}); } - // 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 index_ports() + // Register the product-driven invocation domain and each repeater's receivers. + // No participants are needed when all inputs share the same layer. + join_registration index_ports() { - std::vector result; - result.reserve(repeaters_.size()); + join_registration result{.input_layers = layers_, .participants = {}}; + result.participants.reserve(repeaters_.size()); for (auto const& [layer, repeater] : std::views::zip(layers_, repeaters_)) { - // Leave counting layer unset so the router balances this slot's flush token - // against the node's deepest layer. - result.emplace_back(layer, std::nullopt, &repeater->flush_port(), &repeater->index_port()); + result.participants.emplace_back(layer, &repeater->index_port(), &repeater->flush_port()); } return result; } @@ -134,7 +131,7 @@ namespace phlex::detail { // Stateless placeholder used instead of multilayer_join_node when a node has only a // single input (no joining is required). struct no_join { - static named_index_ports index_ports() { return {}; } + static join_registration index_ports() { return {}; } }; // Maps the number of inputs to the appropriate join type: a real multilayer_join_node diff --git a/phlex/core/observer_node.hpp b/phlex/core/observer_node.hpp index 768c43cbe..6ea680924 100644 --- a/phlex/core/observer_node.hpp +++ b/phlex/core/observer_node.hpp @@ -5,6 +5,7 @@ #include "phlex/core/declared_observer.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" @@ -94,7 +95,7 @@ namespace phlex::detail { } } - named_index_ports index_ports() final { return join_.index_ports(); } + join_registration index_ports() final { return join_.index_ports(); } std::size_t num_calls() const final { return calls_.load(); } input_retriever_types input_{input_arguments()}; diff --git a/phlex/core/predicate_node.hpp b/phlex/core/predicate_node.hpp index ffa8483c4..a8562b983 100644 --- a/phlex/core/predicate_node.hpp +++ b/phlex/core/predicate_node.hpp @@ -6,6 +6,7 @@ #include "phlex/core/detail/filter_impl.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" @@ -107,7 +108,7 @@ namespace phlex::detail { } } - named_index_ports index_ports() final { return join_.index_ports(); } + join_registration index_ports() final { return join_.index_ports(); } std::size_t num_calls() const final { return calls_.load(); } input_retriever_types input_{input_arguments()}; diff --git a/phlex/core/products_consumer.hpp b/phlex/core/products_consumer.hpp index 1ac80dd78..45abac5db 100644 --- a/phlex/core/products_consumer.hpp +++ b/phlex/core/products_consumer.hpp @@ -4,6 +4,7 @@ #include "phlex/core/consumer.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/product_selector.hpp" #include "phlex/model/algorithm_name.hpp" @@ -36,7 +37,7 @@ namespace phlex::detail { std::vector const& layers() const noexcept; tbb::flow::receiver& port(product_selector const& input_product); - virtual named_index_ports index_ports() = 0; + virtual join_registration index_ports() = 0; virtual std::size_t num_calls() const = 0; protected: diff --git a/phlex/core/transform_node.hpp b/phlex/core/transform_node.hpp index 996a74f53..c4581dcbb 100644 --- a/phlex/core/transform_node.hpp +++ b/phlex/core/transform_node.hpp @@ -8,6 +8,7 @@ #include "phlex/core/declared_transform.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" @@ -121,7 +122,7 @@ namespace phlex::detail { } } - named_index_ports index_ports() final { return join_.index_ports(); } + join_registration index_ports() final { return join_.index_ports(); } std::size_t num_calls() const final { return calls_.load(); } input_retriever_types input_{input_arguments()}; diff --git a/phlex/core/unfold_node.hpp b/phlex/core/unfold_node.hpp index 703b97e86..526851e6d 100644 --- a/phlex/core/unfold_node.hpp +++ b/phlex/core/unfold_node.hpp @@ -5,6 +5,7 @@ #include "phlex/core/declared_unfold.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/node_builder.hpp" @@ -191,7 +192,7 @@ namespace phlex::detail { } } - named_index_ports index_ports() final { return join_.index_ports(); } + join_registration index_ports() final { return join_.index_ports(); } std::size_t num_calls() const final { return calls_.load(); } input_retriever_types input_{input_arguments()}; diff --git a/test/fold_test.cpp b/test/fold_test.cpp index 04999f472..dae418584 100644 --- a/test/fold_test.cpp +++ b/test/fold_test.cpp @@ -30,15 +30,21 @@ #include "phlex/utilities/sleep_for.hpp" #include "phlex/utilities/thread_counter.hpp" #include "plugins/layer_generator.hpp" +#include "test/ostream_logger.hpp" +#include #include +#include +#include #include // Provides chrono literals used below but attributed to a transitive include by include-cleaner. #include // IWYU pragma: keep #include #include +#include #include +#include #include using namespace phlex; @@ -115,6 +121,113 @@ TEST_CASE("Different data layers of fold", "[graph]") CHECK(g.execution_count("verify_job_sum") == 1); } +TEST_CASE("Multi-layer fold with parent and child inputs", "[graph][fold]") +{ + auto const [first, second] = GENERATE( + table({{"event", "subevent"}, {"subevent", "event"}})); + CAPTURE(first, second); + + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + { + std::vector sums; + constexpr auto number_events = 2u; + constexpr auto number_subevents = 10u; + + auto gen = experimental::layer_generator::make(); + gen->add_layer("event", {.parent_layer = "job", .count = number_events}); + gen->add_layer("subevent", {.parent_layer = "event", .count = number_subevents}); + + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(gen); + g.provide( + "provide_parent_number", + [](data_cell_index const& id) -> unsigned int { return 10u * (id.number() + 1); }, + concurrency::unlimited) + .output_product("input", "number", "event"); + g.provide("provide_child_number", provide_number, concurrency::unlimited) + .output_product("input", "number", "subevent"); + + // The accumulator is first. Bookkeeping must count subevents in either input order. + g.fold( + "add_parent_and_child", + [](unsigned int& sum, unsigned int first, unsigned int second) { sum += first + second; }, + concurrency::serial, + "event") + .input_family(product_selector{.layer = first}, product_selector{.layer = second}); + + g.observe( + "check_parent_and_child_sum", + [&sums](unsigned int sum) { sums.push_back(sum); }, + concurrency::serial) + .input_family({.creator = "add_parent_and_child", .layer = "event"}); + + g.execute(); + + CHECK(g.execution_count("provide_parent_number") == number_events); + CHECK(g.execution_count("provide_child_number") == + std::size_t{number_events} * number_subevents); + CHECK(g.execution_count("add_parent_and_child") == + std::size_t{number_events} * number_subevents); + CHECK(g.execution_count("check_parent_and_child_sum") == number_events); + + std::ranges::sort(sums); + CHECK(sums == std::vector{145u, 245u}); + } + CHECK_FALSE(output.str().contains("Cached")); +} + +TEST_CASE("Multi-layer fold excludes products outside its partition", "[graph][fold]") +{ + auto const [first, second] = GENERATE( + table({{"event", "subevent"}, {"subevent", "event"}})); + CAPTURE(first, second); + + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + { + std::vector sums; + constexpr auto number_subevents = 10u; + + auto gen = experimental::layer_generator::make(); + gen->add_layer("run", {.parent_layer = "job", .count = 1}); + gen->add_layer("event", {.parent_layer = "run", .count = 2}); + gen->add_layer("event", {.parent_layer = "job", .count = 1}); + gen->add_layer("subevent", {.parent_layer = "/job/run/event", .count = number_subevents}); + gen->add_layer("subevent", {.parent_layer = "/job/event", .count = number_subevents}); + + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(gen); + g.provide( + "provide_parent_number", + [](data_cell_index const& id) -> unsigned int { return 10u * (id.number() + 1); }, + concurrency::unlimited) + .output_product("input", "number", "event"); + g.provide("provide_child_number", provide_number, concurrency::unlimited) + .output_product("input", "number", "subevent"); + g.fold( + "add_parent_and_child", + [](unsigned int& sum, unsigned int first, unsigned int second) { sum += first + second; }, + concurrency::serial, + "run") + .input_family(product_selector{.layer = first}, product_selector{.layer = second}); + g.observe( + "check_parent_and_child_sum", + [&sums](unsigned int sum) { sums.push_back(sum); }, + concurrency::serial) + .input_family({.creator = "add_parent_and_child", .layer = "run"}); + + g.execute(); + + CHECK(g.execution_count("provide_parent_number") == 3uz); + CHECK(g.execution_count("provide_child_number") == 3uz * number_subevents); + CHECK(g.execution_count("add_parent_and_child") == 2uz * number_subevents); + CHECK(g.execution_count("check_parent_and_child_sum") == 1uz); + CHECK(sums == std::vector{390u}); + } + CHECK_FALSE(output.str().contains("Cached")); +} + TEST_CASE("Fold output without send consumed downstream", "[graph]") { constexpr auto index_limit = 2u; diff --git a/test/framework_graph_test.cpp b/test/framework_graph_test.cpp index 61d273858..2f439853e 100644 --- a/test/framework_graph_test.cpp +++ b/test/framework_graph_test.cpp @@ -7,17 +7,23 @@ #include "phlex/model/fixed_hierarchy.hpp" #include "phlex/model/fwd.hpp" #include "plugins/layer_generator.hpp" +#include "test/ostream_logger.hpp" #include +#include #include +#include #include #include #include +#include #include #include #include +#include #include +#include #include #include #include @@ -43,6 +49,22 @@ namespace { return [](data_cell_yielder const /*yielder*/) {}; } }; + + class split { + public: + explicit split(unsigned int count) : count_{count} {} + static unsigned int initial_value() { return 0; } + bool predicate(unsigned int i) const { return i < count_; } + static auto unfold(unsigned int i) { return std::make_pair(i + 1, 1u); } + + private: + unsigned int count_; + }; + + class split_with_run : public split { + public: + split_with_run(unsigned int count, unsigned int /*run*/) : split{count} {} + }; } TEST_CASE("Catch STL exceptions", "[graph]") @@ -315,3 +337,86 @@ TEST_CASE("Throw when configuring driver twice", "[graph]") g.add_driver(gen), Catch::Matchers::ContainsSubstring("Driver has already been configured for framework_graph")); } + +TEST_CASE("Leaf-only hierarchy resolves ancestor chains", "[graph][issue955]") +{ + auto const [description, hierarchy] = + GENERATE(Catch::Generators::table( + {{"Implicit job prefix", fixed_hierarchy{{"run", "spill"}}}, + {"Explicit job prefix", fixed_hierarchy{{"job", "run", "spill"}}}})); + CAPTURE(description); + + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [](fixed_hierarchy const& hierarchy) { + std::vector sums; + std::vector joined_sums; + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(g.driver_proxy({}).driver(hierarchy, [](data_cell_cursor job) { + for (auto r = 0u; r != 2; ++r) { + auto run = job.yield_child("run", r); + for (auto s = 0u; s != 3; ++s) { + run.yield_child("spill", s); + } + } + })); + g.provide( + "run_number", [](data_cell_index const&) { return 2u; }, concurrency::unlimited) + .output_product("input", "run_number", "run"); + g.provide( + "spill_number", [](data_cell_index const&) { return 3u; }, concurrency::unlimited) + .output_product("input", "spill_number", "spill"); + + // The dependent name sorts before its producer and consumes an implicit ancestor prefix. + g.unfold("a_split", + &split_with_run::predicate, + &split_with_run::unfold, + concurrency::unlimited, + "unit") + .input_family(product_selector{.creator = "z_split", .layer = "piece", .suffix = "number"}, + product_selector{.creator = "input", .layer = "run", .suffix = "run_number"}) + .output_product_suffixes("number"); + g.unfold("z_split", &split::predicate, &split::unfold, concurrency::unlimited, "piece") + .input_family( + product_selector{.creator = "input", .layer = "spill", .suffix = "spill_number"}) + .output_product_suffixes("number"); + + auto add = [](std::atomic& sum, unsigned int number) { sum += number; }; + g.fold("sum", add, concurrency::unlimited, "job", 0u) + .input_family(product_selector{.creator = "a_split", .layer = "unit", .suffix = "number"}) + .output_product_suffixes("sum"); + g.observe( + "collect_sum", [&sums](unsigned int sum) { sums.push_back(sum); }, concurrency::serial) + .input_family(product_selector{.creator = "sum", .layer = "job", .suffix = "sum"}); + + g.transform("join", [](unsigned int run, unsigned int spill) { return run + spill; }) + .input_family( + product_selector{.creator = "input", .layer = "run", .suffix = "run_number"}, + product_selector{.creator = "input", .layer = "spill", .suffix = "spill_number"}) + .output_product_suffixes("number"); + g.fold("joined_sum", add, concurrency::unlimited, "job", 0u) + .input_family(product_selector{.creator = "join", .layer = "spill", .suffix = "number"}) + .output_product_suffixes("sum"); + g.observe( + "collect_joined_sum", + [&joined_sums](unsigned int sum) { joined_sums.push_back(sum); }, + concurrency::serial) + .input_family(product_selector{.creator = "joined_sum", .layer = "job", .suffix = "sum"}); + + g.execute(); + CHECK(g.execution_count("z_split") == 6); + CHECK(g.execution_count("a_split") == 18); + CHECK(g.execution_count("sum") == 18); + CHECK(g.execution_count("join") == 6); + CHECK(g.execution_count("collect_sum") == 1); + CHECK(g.execution_count("collect_joined_sum") == 1); + return std::make_pair(std::move(sums), std::move(joined_sums)); + }; + // Destroy the graph before checking for cached-entry warnings. + auto const [sums, joined_sums] = run_graph(hierarchy); + CHECK(sums == std::vector{18}); + CHECK(joined_sums == std::vector{30}); + + auto const messages = output.str(); + CHECK_FALSE(messages.contains("Cached")); +} diff --git a/test/index_router_test.cpp b/test/index_router_test.cpp index 6ebffbc28..c60687808 100644 --- a/test/index_router_test.cpp +++ b/test/index_router_test.cpp @@ -1,4 +1,5 @@ #include "phlex/core/index_router.hpp" +#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/model/data_cell_counts.hpp" #include "phlex/model/data_cell_index.hpp" @@ -9,9 +10,12 @@ #include #include #include +#include #include #include +#include +#include #include using namespace phlex; @@ -30,6 +34,7 @@ TEST_CASE("Duplicate counting paths deliver one combined end token", "[graph][is tbb::flow::graph graph; tbb::flow::queue_node indices{graph}; tbb::flow::queue_node tokens{graph}; + tbb::flow::queue_node data_tokens{graph}; detail::index_router router{graph}; router.finalize( graph, @@ -38,10 +43,10 @@ TEST_CASE("Duplicate counting paths deliver one combined end token", "[graph][is {{"dummy", {.input_product = {.layer = "job"}, .port = &indices}}}, {}, {{"fold", - {{.layer = "job"_id, - .counting_layer = "event"_id, - .token_port = &tokens, - .index_port = &indices}}}}); + {.input_layers = {"event"_id}, + .participants = { + {.layer = "job"_id, .index_port = &indices, .completion_port = &tokens}, + {.layer = "event"_id, .index_port = &indices, .completion_port = &data_tokens}}}}}); auto job = data_cell_index::job(); auto run = job->make_child("run", 0); @@ -53,6 +58,7 @@ TEST_CASE("Duplicate counting paths deliver one combined end token", "[graph][is router.drain({{.index = job, .counts = job_counts}}); graph.wait_for_all(); detail::indexed_end_token token; + // Completion must wait for the nested counting path, even when its count will be zero. CHECK_FALSE(tokens.try_get(token)); auto run_counts = std::make_shared(); @@ -62,5 +68,255 @@ TEST_CASE("Duplicate counting paths deliver one combined end token", "[graph][is REQUIRE(tokens.try_get(token)); CHECK(token.index == job); CHECK(std::cmp_equal(token.count, top_count + nested_count)); + // Combine both counting paths into one token; the pass-through participant gets none. CHECK_FALSE(tokens.try_get(token)); + CHECK_FALSE(data_tokens.try_get(token)); +} + +TEST_CASE("Join participants share partition-constrained invocation accounting", "[graph][fold]") +{ + using namespace phlex::experimental::literals; + auto const count = GENERATE(0u, 3u); + tbb::flow::graph graph; + tbb::flow::queue_node provider_indices{graph}; + tbb::flow::queue_node accumulator_indices{graph}; + tbb::flow::queue_node parent_indices{graph}; + tbb::flow::queue_node child_indices{graph}; + tbb::flow::queue_node accumulator_tokens{graph}; + tbb::flow::queue_node parent_tokens{graph}; + tbb::flow::queue_node child_tokens{graph}; + detail::index_router router{graph}; + router.finalize( + graph, + {experimental::layer_path{"/job/run/event/subevent"}, + experimental::layer_path{"/job/event/subevent"}}, + {}, + {{"dummy", {.input_product = {.layer = "job"}, .port = &provider_indices}}}, + {}, + {{"fold", + {.input_layers = {"event"_id, "subevent"_id}, + .participants = { + {.layer = "run"_id, + .index_port = &accumulator_indices, + .completion_port = &accumulator_tokens}, + {.layer = "event"_id, .index_port = &parent_indices, .completion_port = &parent_tokens}, + {.layer = "subevent"_id, + .index_port = &child_indices, + .completion_port = &child_tokens}}}}}); + + auto job = data_cell_index::job(); + auto run = job->make_child("run", 0); + auto event = run->make_child("event", 0); + auto excluded = job->make_child("event", 0); + router.route(job, {}); + router.route(run, {}); + router.route(event, {}); + router.route(excluded, {}); + router.route(excluded->make_child("subevent", 0), {}); + for (unsigned int i = 0; i != count; ++i) { + router.route(event->make_child("subevent", i), {}); + } + auto event_counts = std::make_shared(); + event_counts->emplace(event->make_child("subevent", 0)->layer_hash(), count); + router.drain({{.index = event, .counts = event_counts}}); + auto run_counts = std::make_shared(); + run_counts->emplace(event->layer_hash(), 1); + router.drain({{.index = run, .counts = run_counts}}); + auto excluded_counts = std::make_shared(); + excluded_counts->emplace(excluded->make_child("subevent", 0)->layer_hash(), 1); + router.drain({{.index = excluded, .counts = excluded_counts}}); + graph.wait_for_all(); + + detail::indexed_end_token token; + // Cached partitions get exactly one completion each; pass-through children get none. + REQUIRE(accumulator_tokens.try_get(token)); + CHECK(token.index == run); + CHECK(std::cmp_equal(token.count, count)); + CHECK_FALSE(accumulator_tokens.try_get(token)); + REQUIRE(parent_tokens.try_get(token)); + CHECK(token.index == event); + CHECK(std::cmp_equal(token.count, count)); + REQUIRE(parent_tokens.try_get(token)); + CHECK(token.index == excluded); + CHECK(token.count == 0); + CHECK_FALSE(parent_tokens.try_get(token)); + CHECK_FALSE(child_tokens.try_get(token)); + + detail::index_message index; + for (unsigned int i = 0; i != count; ++i) { + REQUIRE(accumulator_indices.try_get(index)); + CHECK(index.index == run); + CHECK(index.cache); + auto const message_id = index.msg_id; + REQUIRE(parent_indices.try_get(index)); + CHECK(index.index == event); + CHECK(index.cache); + CHECK(index.msg_id == message_id); + REQUIRE(child_indices.try_get(index)); + CHECK(index.index->parent() == event); + CHECK_FALSE(index.cache); + CHECK(index.msg_id == message_id); + } + // The excluded branch must not add invocations, including when the eligible count is zero. + CHECK_FALSE(accumulator_indices.try_get(index)); + CHECK_FALSE(parent_indices.try_get(index)); + CHECK_FALSE(child_indices.try_get(index)); +} + +TEST_CASE("Join rejects participants with mixed caching roles", "[graph]") +{ + using namespace phlex::experimental::literals; + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + tbb::flow::queue_node tokens{graph}; + detail::index_router router{graph}; + CHECK_THROWS_AS( + router.finalize( + graph, + {experimental::layer_path{"/job/a/b"}, experimental::layer_path{"/job/b/a"}}, + {}, + {{"dummy", {.input_product = {.layer = "job"}, .port = &indices}}}, + {}, + {{"join", + {.input_layers = {"a"_id, "b"_id}, + .participants = + {{.layer = "a"_id, .index_port = &indices, .completion_port = &tokens}, + {.layer = "b"_id, .index_port = &indices, .completion_port = &tokens}}}}}), + std::runtime_error); +} + +TEST_CASE("Join completion follows nearest repeated ancestor", "[graph]") +{ + using namespace phlex::experimental::literals; + tbb::flow::graph graph; + tbb::flow::queue_node provider_indices{graph}; + tbb::flow::queue_node parent_indices{graph}; + tbb::flow::queue_node child_indices{graph}; + tbb::flow::queue_node parent_tokens{graph}; + tbb::flow::queue_node child_tokens{graph}; + detail::index_router router{graph}; + router.finalize( + graph, + {experimental::layer_path{"/job/a/a/b"}}, + {}, + {{"dummy", {.input_product = {.layer = "job"}, .port = &provider_indices}}}, + {}, + {{"join", + {.input_layers = {"a"_id, "b"_id}, + .participants = { + {.layer = "a"_id, .index_port = &parent_indices, .completion_port = &parent_tokens}, + {.layer = "b"_id, .index_port = &child_indices, .completion_port = &child_tokens}}}}}); + auto job = data_cell_index::job(); + auto outer = job->make_child("a", 0); + auto inner = outer->make_child("a", 0); + auto child = inner->make_child("b", 0); + router.route(job, {}); + router.route(outer, {}); + router.route(inner, {}); + router.route(child, {}); + auto inner_counts = std::make_shared(); + inner_counts->emplace(child->layer_hash(), 1); + router.drain({{.index = inner, .counts = inner_counts}}); + graph.wait_for_all(); + detail::indexed_end_token token; + REQUIRE(parent_tokens.try_get(token)); + CHECK(token.index == inner); + CHECK(token.count == 1); + auto outer_counts = std::make_shared(); + outer_counts->emplace(inner->layer_hash(), 1); + router.drain({{.index = outer, .counts = outer_counts}}); + graph.wait_for_all(); + REQUIRE(parent_tokens.try_get(token)); + CHECK(token.index == outer); + CHECK(token.count == 0); + CHECK_FALSE(parent_tokens.try_get(token)); + // The child is pass-through and needs no completion token. + CHECK_FALSE(child_tokens.try_get(token)); + detail::index_message index; + REQUIRE(parent_indices.try_get(index)); + CHECK(index.index == inner); + CHECK(index.cache); + // One invocation projects only to the nearest "a", not to both occurrences. + CHECK_FALSE(parent_indices.try_get(index)); +} + +TEST_CASE("Unfold hierarchy resolves dependencies before their producers", "[graph][unfold]") +{ + using namespace phlex::experimental::literals; + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + tbb::flow::queue_node tokens{graph}; + tbb::flow::queue_node data_tokens{graph}; + detail::index_router router{graph}; + router.finalize( + graph, + {experimental::layer_path{"/job/spill"}}, + // The dependent comes first both in this list and in lexical node-name order. + {{.name = "a_split_b", .input_layers = {"a"_id, "spill"_id}, .output_layer = "b"_id}, + {.name = "z_split_a", .input_layers = {"spill"_id}, .output_layer = "a"_id}}, + {{"dummy", {.input_product = {.layer = "spill"}, .port = &indices}}}, + {}, + {{"fold", + {.input_layers = {"b"_id}, + .participants = { + {.layer = "spill"_id, .index_port = &indices, .completion_port = &tokens}, + {.layer = "b"_id, .index_port = &indices, .completion_port = &data_tokens}}}}}); + + auto spill = data_cell_index::job()->make_child("spill", 0); + auto a = spill->make_child("a", 0); + auto b = a->make_child("b", 0); + router.route(spill, {}); + router.route(a, {}); + auto spill_counts = std::make_shared(); + spill_counts->emplace(a->layer_hash(), 1); + router.drain({{.index = spill, .counts = spill_counts}}); + graph.wait_for_all(); + detail::indexed_end_token token; + // The spill cannot complete until the nested unfold supplies its child count. + CHECK_FALSE(tokens.try_get(token)); + + router.unfold_flush_receiver().try_put({.index = a, .layer_hash = b->layer_hash(), .count = 3}); + graph.wait_for_all(); + REQUIRE(tokens.try_get(token)); + CHECK(token.index == spill); + CHECK(token.count == 3); + // Emit one spill completion and none for the pass-through b participant. + CHECK_FALSE(tokens.try_get(token)); + CHECK_FALSE(data_tokens.try_get(token)); +} + +TEST_CASE("Unfold hierarchy rejects cyclic layer production", "[graph][unfold]") +{ + using namespace phlex::experimental::literals; + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + detail::index_router router{graph}; + detail::index_router::unfold_data unfolds; + std::string expected_error; + + SECTION("Self-cycle") + { + unfolds = {{.name = "split_run", .input_layers = {"run"_id}, .output_layer = "run"_id}}; + expected_error = + "Unfold layer hierarchy expansion exceeded max depth 3 (initial deepest 2 + 1 unfold " + "node(s)). Offending unfold(s):\n - split_run"; + } + + SECTION("Two-unfold cycle") + { + unfolds = {{.name = "split_spill", .input_layers = {"run"_id}, .output_layer = "spill"_id}, + {.name = "split_run", .input_layers = {"spill"_id}, .output_layer = "run"_id}}; + expected_error = + "Unfold layer hierarchy expansion exceeded max depth 4 (initial deepest 2 + 2 unfold " + "node(s)). Offending unfold(s):\n - split_spill"; + } + + CHECK_THROWS_WITH( + router.finalize(graph, + {experimental::layer_path{"/job/run"}}, + unfolds, + {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, + {}, + {}), + expected_error); } diff --git a/test/multilayer_join_test.cpp b/test/multilayer_join_test.cpp index dac4a4c70..efc79e646 100644 --- a/test/multilayer_join_test.cpp +++ b/test/multilayer_join_test.cpp @@ -85,7 +85,9 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]") graph.wait_for_all(); CHECK_FALSE(sink.try_get(output)); - auto index_ports = join.index_ports(); + auto registration = join.index_ports(); + CHECK(registration.input_layers == std::vector{"left_layer", "right_layer"}); + auto const& index_ports = registration.participants; REQUIRE(index_ports.size() == 2u); REQUIRE(index_ports[0].index_port->try_put( {.index = left_store->index(), .msg_id = message_id, .cache = true})); @@ -97,7 +99,6 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]") graph.wait_for_all(); REQUIRE(sink.try_get(output)); - CHECK_FALSE(sink.try_get(output)); CHECK(std::get<0>(output).id == message_id); CHECK(std::get<1>(output).id == message_id); @@ -108,7 +109,7 @@ TEST_CASE("multilayer_join_node joins multiple input products", "[join]") // Do what is necessary to have the tokens flushed, so that the test does not generate // warnings. - REQUIRE(index_ports[0].token_port->try_put({.index = left_store->index(), .count = 1})); - REQUIRE(index_ports[1].token_port->try_put({.index = right_store->index(), .count = 1})); + REQUIRE(index_ports[0].completion_port->try_put({.index = left_store->index(), .count = 1})); + REQUIRE(index_ports[1].completion_port->try_put({.index = right_store->index(), .count = 1})); graph.wait_for_all(); } diff --git a/test/unfold_test.cpp b/test/unfold_test.cpp index 3c849b306..6036ce6ae 100644 --- a/test/unfold_test.cpp +++ b/test/unfold_test.cpp @@ -19,18 +19,25 @@ #include "phlex/core/resource_api.hpp" #include "phlex/model/data_cell_index.hpp" #include "phlex/model/handle.hpp" +#include "phlex/model/layer_path.hpp" #include "phlex/utilities/sleep_for.hpp" #include "phlex/utilities/thread_counter.hpp" #include "plugins/layer_generator.hpp" +#include "test/ostream_logger.hpp" #include "test/products_for_output.hpp" +#include #include -#include +#include #include +#include #include // Provides chrono literals used below but attributed to a transitive include by include-cleaner. #include // IWYU pragma: keep +#include +#include +#include #include #include #include @@ -110,6 +117,17 @@ namespace { numbers_t numbers_; }; + class iota_with_ancestor : public iota { + public: + iota_with_ancestor(unsigned int max_number, unsigned int /*ancestor*/) : iota{max_number} {} + }; + + void check_no_cached_entries(std::ostringstream const& output) + { + auto const messages = output.str(); + CHECK_FALSE(messages.contains("Cached")); + } + // Test load-bearing resource for unfold operations. struct unfold_resource { using token_type = unfold_resource*; @@ -138,68 +156,79 @@ namespace { TEST_CASE("Splitting the processing", "[graph]") { - constexpr auto index_limit = 2u; - - auto gen = experimental::layer_generator::make(); - gen->add_layer("event", {.parent_layer = "job", .count = index_limit}); - - auto g = phlex::detail::framework_graph::without_driver("test"); - g.add_driver(gen); - - g.provide("provide_max_number", provide_max_number, concurrency::unlimited) - .output_product("input", "max_number", "event"); - g.provide("provide_ten_numbers", provide_ten_numbers, concurrency::unlimited) - .output_product("input", "ten_numbers", "event"); - - g.unfold("iota", &iota::predicate, &iota::unfold, concurrency::unlimited, "lower1") - .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "max_number"}) - .output_product_suffixes("new_number"); - g.fold("add", add, concurrency::unlimited, "event") - .input_family(product_selector{.creator = "iota", .layer = "lower1", .suffix = "new_number"}) - .output_product_suffixes("sum1"); - g.observe("check_sum", check_sum, concurrency::unlimited) - .input_family(product_selector{.creator = "add", .layer = "event", .suffix = "sum1"}); - - g.unfold("iterate_through", - &iterate_through::predicate, - &iterate_through::unfold, - concurrency::unlimited, - "lower2") - .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "ten_numbers"}) - .output_product_suffixes("each_number"); - g.fold("add_numbers", add_numbers, concurrency::unlimited, "event") - .input_family( - product_selector{.creator = "iterate_through", .layer = "lower2", .suffix = "each_number"}) - .output_product_suffixes("sum2"); - g.observe("check_sum_same", check_sum_same, concurrency::unlimited) - .input_family(product_selector{.creator = "add_numbers", .layer = "event", .suffix = "sum2"}); - - g.make().output( - "save", &experimental::test::products_for_output::save, concurrency::serial); - - g.execute(); - - CHECK(g.execution_count("iota") == index_limit); - CHECK(g.execution_count("add") == 30); - CHECK(g.execution_count("check_sum") == index_limit); - - CHECK(g.execution_count("iterate_through") == index_limit); - CHECK(g.execution_count("add_numbers") == 20); - CHECK(g.execution_count("check_sum_same") == index_limit); -} - -TEST_CASE("Multi-layer unfolds not supported until v0.5", "[graph][unfold]") -{ - auto g = phlex::detail::framework_graph::without_driver("test"); - CHECK_THROWS_WITH( - g.unfold("iota_two_inputs", - &iota_two_inputs::predicate, - &iota_two_inputs::unfold, + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [] { + std::vector sums1; + std::vector sums2; + constexpr auto number_events = 2u; + auto gen = experimental::layer_generator::make(); + gen->add_layer("event", {.parent_layer = "job", .count = number_events}); + + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(gen); + + g.provide("provide_max_number", provide_max_number, concurrency::unlimited) + .output_product("input", "max_number", "event"); + g.provide("provide_ten_numbers", provide_ten_numbers, concurrency::unlimited) + .output_product("input", "ten_numbers", "event"); + + g.unfold("iota", &iota::predicate, &iota::unfold, concurrency::unlimited, "lower1") + .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "max_number"}) + .output_product_suffixes("new_number"); + g.fold("add", add, concurrency::unlimited, "event") + .input_family(product_selector{.creator = "iota", .layer = "lower1", .suffix = "new_number"}) + .output_product_suffixes("sum1"); + g.observe( + "check_sum", + [&sums1](handle sum) { + check_sum(sum); + sums1.push_back(*sum); + }, + concurrency::serial) + .input_family(product_selector{.creator = "add", .layer = "event", .suffix = "sum1"}); + + g.unfold("iterate_through", + &iterate_through::predicate, + &iterate_through::unfold, concurrency::unlimited, - "subevent") - .input_family(product_selector{.layer = "run"}, product_selector{.layer = "event"}), - "Unfolds cannot yet accept multi-layer inputs (node: iota_two_inputs, layers: {\"run\", " - "\"event\"})"); + "lower2") + .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "ten_numbers"}) + .output_product_suffixes("each_number"); + g.fold("add_numbers", add_numbers, concurrency::unlimited, "event") + .input_family( + product_selector{.creator = "iterate_through", .layer = "lower2", .suffix = "each_number"}) + .output_product_suffixes("sum2"); + g.observe( + "check_sum_same", + [&sums2](handle sum) { + check_sum_same(sum); + sums2.push_back(*sum); + }, + concurrency::serial) + .input_family(product_selector{.creator = "add_numbers", .layer = "event", .suffix = "sum2"}); + + g.make().output( + "save", &experimental::test::products_for_output::save, concurrency::serial); + + g.execute(); + + CHECK(g.execution_count("iota") == number_events); + CHECK(g.execution_count("add") == 30); + CHECK(g.execution_count("check_sum") == number_events); + + CHECK(g.execution_count("iterate_through") == number_events); + CHECK(g.execution_count("add_numbers") == 20); + CHECK(g.execution_count("check_sum_same") == number_events); + return std::make_pair(std::move(sums1), std::move(sums2)); + }; + // Destroy the graph before checking for cached-entry warnings. + auto [sums1, sums2] = run_graph(); + std::ranges::sort(sums1); + std::ranges::sort(sums2); + CHECK(sums1 == std::vector{45, 190}); + CHECK(sums2 == std::vector{10, 20}); + check_no_cached_entries(output); } // ======================================================================================= @@ -220,10 +249,10 @@ TEST_CASE("Multi-layer unfolds not supported until v0.5", "[graph][unfold]") // ======================================================================================= TEST_CASE("Multi-layer transform with one input from an unfold", "[graph]") { - constexpr auto index_limit = 2u; + constexpr auto number_events = 2u; auto gen = experimental::layer_generator::make(); - gen->add_layer("event", {.parent_layer = "job", .count = index_limit}); + gen->add_layer("event", {.parent_layer = "job", .count = number_events}); auto g = phlex::detail::framework_graph::without_driver("test"); g.add_driver(gen); @@ -247,48 +276,61 @@ TEST_CASE("Multi-layer transform with one input from an unfold", "[graph]") // event 0: max_number=10, new_number in [0,9] -> 10 executions // event 1: max_number=20, new_number in [0,19] -> 20 executions - CHECK(g.execution_count("iota") == index_limit); + CHECK(g.execution_count("iota") == number_events); CHECK(g.execution_count("add_max_and_new") == 30u); } TEST_CASE("Unfold deduplicates same-layer inputs for bookkeeping", "[graph]") { - constexpr auto index_limit = 2u; + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [] { + std::vector sums; + constexpr auto number_events = 2u; + auto gen = experimental::layer_generator::make(); + gen->add_layer("event", {.parent_layer = "job", .count = number_events}); + + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(gen); + + g.provide("provide_max_number", provide_max_number, concurrency::unlimited) + .output_product("input", "max_number", "event"); + g.provide("provide_ten_numbers", provide_ten_numbers, concurrency::unlimited) + .output_product("input", "ten_numbers", "event"); - auto gen = experimental::layer_generator::make(); - gen->add_layer("event", {.parent_layer = "job", .count = index_limit}); - - auto g = phlex::detail::framework_graph::without_driver("test"); - g.add_driver(gen); - - g.provide("provide_max_number", provide_max_number, concurrency::unlimited) - .output_product("input", "max_number", "event"); - g.provide("provide_ten_numbers", provide_ten_numbers, concurrency::unlimited) - .output_product("input", "ten_numbers", "event"); - - g.unfold("iota_two_inputs", - &iota_two_inputs::predicate, - &iota_two_inputs::unfold, - concurrency::unlimited, - "subevent") - .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "max_number"}, - product_selector{.creator = "input", .layer = "event", .suffix = "ten_numbers"}) - .output_product_suffixes("new_number"); - - g.fold("add_dual_input_unfold", add, concurrency::unlimited, "event") - .input_family( - product_selector{.creator = "iota_two_inputs", .layer = "subevent", .suffix = "new_number"}) - .output_product_suffixes("sum"); - - g.observe("check_dual_input_unfold_sum", check_sum, concurrency::unlimited) - .input_family( - product_selector{.creator = "add_dual_input_unfold", .layer = "event", .suffix = "sum"}); - - g.execute(); - - CHECK(g.execution_count("iota_two_inputs") == index_limit); - CHECK(g.execution_count("add_dual_input_unfold") == 30u); - CHECK(g.execution_count("check_dual_input_unfold_sum") == index_limit); + g.unfold("iota_two_inputs", + &iota_two_inputs::predicate, + &iota_two_inputs::unfold, + concurrency::unlimited, + "subevent") + .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "max_number"}, + product_selector{.creator = "input", .layer = "event", .suffix = "ten_numbers"}) + .output_product_suffixes("new_number"); + + g.fold("add_dual_input_unfold", add, concurrency::unlimited, "event") + .input_family( + product_selector{.creator = "iota_two_inputs", .layer = "subevent", .suffix = "new_number"}) + .output_product_suffixes("sum"); + + g.observe( + "check_dual_input_unfold_sum", + [&sums](unsigned int sum) { sums.push_back(sum); }, + concurrency::serial) + .input_family( + product_selector{.creator = "add_dual_input_unfold", .layer = "event", .suffix = "sum"}); + + g.execute(); + + CHECK(g.execution_count("iota_two_inputs") == number_events); + CHECK(g.execution_count("add_dual_input_unfold") == 30u); + CHECK(g.execution_count("check_dual_input_unfold_sum") == number_events); + return sums; + }; + // Destroy the graph before checking for cached-entry warnings. + auto sums = run_graph(); + std::ranges::sort(sums); + CHECK(sums == std::vector{45, 190}); + check_no_cached_entries(output); } TEST_CASE("Unfold receives a resource token", "[graph][unfold][resource]") @@ -319,3 +361,196 @@ TEST_CASE("Unfold receives a resource token", "[graph][unfold][resource]") CHECK(g.execution_count("resource_iota") == num_events); } + +// Reproducer for issue #955: job -> spill --split_a--> a --split_b--> b. +// Joining the spill product in split_b must not prevent the spill fold from flushing. +TEST_CASE("Nested unfold with a higher-layer input flushes the outer layer", "[graph][unfold]") +{ + using data_cell_index_number = unsigned int; + using counts_t = std::vector>; + struct scenario { + char const* description; + std::vector expected_b_counts; + }; + auto const test = + GENERATE(scenario{.description = "Two-input nested unfold (issue reproducer)", + .expected_b_counts = {2u, 2u, 2u}}, + scenario{.description = "Some empty b children, including an empty spill partition", + .expected_b_counts = {0u, 1u, 2u}}, + scenario{.description = "All b partitions empty", .expected_b_counts = {0u, 0u, 0u}}); + CAPTURE(test.description); + + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [](scenario const& test) { + counts_t counts; + counts_t sums; + auto const num_spills = static_cast(test.expected_b_counts.size()); + constexpr int threads = 1; + auto const num_b = + std::accumulate(test.expected_b_counts.begin(), test.expected_b_counts.end(), std::size_t{0}); + + auto gen = experimental::layer_generator::make(); + gen->add_layer("spill", {.parent_layer = "job", .count = num_spills}); + + auto g = phlex::detail::framework_graph::without_driver("test", threads); + g.add_driver(gen); + g.provide( + "provide_n", [](data_cell_index const&) { return 2u; }, concurrency::unlimited) + .output_product("src", "n", "spill"); + + product_selector const spill_n{.creator = "src", .layer = "spill", .suffix = "n"}; + product_selector const a_n{.creator = "split_a", .layer = "a", .suffix = "n"}; + product_selector const b_n{.creator = "split_b", .layer = "b", .suffix = "n"}; + product_selector const split_b_input{ + .creator = "set_b_child_count", .layer = "a", .suffix = "n"}; + + g.transform( + "set_b_child_count", + [expected_b_counts = test.expected_b_counts](handle n) { + auto const& index = n.data_cell_index(); + auto const spill_number = index.parent()->number(); + return static_cast(index.number() < expected_b_counts.at(spill_number)); + }, + concurrency::unlimited) + .input_family(a_n) + .output_product_suffixes("n"); + + g.unfold("split_a", &iota::predicate, &iota::unfold, concurrency::unlimited, "a") + .input_family(spill_n) + .output_product_suffixes("n"); + + g.unfold( + "split_b", &iota::predicate, &iota::unfold, concurrency::unlimited, "b") + .input_family(split_b_input, spill_n) + .output_product_suffixes("n"); + + g.fold( + "count", [](int& count, unsigned int) { ++count; }, concurrency::serial, "spill") + .input_family(b_n) + .output_product_suffixes("count"); + g.observe( + "check_count", + [&counts](handle count) { + counts.emplace_back(count.data_cell_index().number(), *count); + }, + concurrency::serial) + .input_family(product_selector{.creator = "count", .layer = "spill", .suffix = "count"}); + + g.transform( + "join_b_spill", + [](unsigned int b, unsigned int spill) { return static_cast(b + spill); }, + concurrency::unlimited) + .input_family(b_n, spill_n) + .output_product_suffixes("n"); + g.fold( + "sum_joined", [](int& sum, int n) { sum += n; }, concurrency::serial, "spill") + .input_family(product_selector{.creator = "join_b_spill", .layer = "b", .suffix = "n"}) + .output_product_suffixes("sum"); + g.observe( + "check_joined", + [&sums](handle sum) { sums.emplace_back(sum.data_cell_index().number(), *sum); }, + concurrency::serial) + .input_family(product_selector{.creator = "sum_joined", .layer = "spill", .suffix = "sum"}); + + g.execute(); + + CHECK(g.execution_count("split_a") == num_spills); + CHECK(g.execution_count("split_b") == static_cast(2 * num_spills)); + CHECK(g.execution_count("count") == num_b); + CHECK(g.execution_count("join_b_spill") == num_b); + CHECK(g.execution_count("sum_joined") == num_b); + // Check emission separately: checks inside the observer cannot catch it never running. + CHECK(g.execution_count("check_count") == num_spills); + CHECK(g.execution_count("check_joined") == num_spills); + return std::make_pair(std::move(counts), std::move(sums)); + }; + // Destroy the graph before checking for cached-entry warnings. + auto [counts, sums] = run_graph(test); + std::ranges::sort(counts); + std::ranges::sort(sums); + counts_t expected_counts; + counts_t expected_sums; + for (data_cell_index_number spill = 0; spill != test.expected_b_counts.size(); ++spill) { + auto const count = static_cast(test.expected_b_counts[spill]); + expected_counts.emplace_back(spill, count); + // Each b product is 0, and the spill product is 2. + expected_sums.emplace_back(spill, count * 2); + } + CHECK(counts == expected_counts); + CHECK(sums == expected_sums); + check_no_cached_entries(output); +} + +TEST_CASE("Unfold flush expectations distinguish repeated layer names", "[graph][unfold]") +{ + std::ostringstream output; + auto logger = test::use_ostream_logger(output); + auto const run_graph = [] { + std::vector> counts; + std::vector extra_counts; + auto gen = experimental::layer_generator::make(); + gen->add_layer("run", {.parent_layer = "job", .count = 1}); + gen->add_layer("spill", {.parent_layer = "run", .count = 2}); + gen->add_layer("spill", {.parent_layer = "job", .count = 3}); + auto g = phlex::detail::framework_graph::without_driver("test"); + g.add_driver(gen); + g.provide( + "provide_n", + [](data_cell_index const& index) { return index.depth() == 1 ? 2u : 3u; }, + concurrency::unlimited) + .output_product("src", "n", "spill"); + g.provide( + "provide_run", [](data_cell_index const&) { return 7u; }, concurrency::unlimited) + .output_product("src", "n", "run"); + + product_selector const spill_n{.creator = "src", .layer = "spill", .suffix = "n"}; + g.unfold("split_a", &iota::predicate, &iota::unfold, concurrency::unlimited, "a") + .input_family(spill_n) + .output_product_suffixes("n"); + // Only /job/run/spill has a run ancestor, so its flush gate expects a second unfold. + g.unfold( + "split_c", &iota::predicate, &iota::unfold, concurrency::unlimited, "c") + .input_family(spill_n, product_selector{.creator = "src", .layer = "run", .suffix = "n"}) + .output_product_suffixes("n"); + g.fold( + "count_a", [](int& count, unsigned int) { ++count; }, concurrency::serial, "spill") + .input_family(product_selector{.creator = "split_a", .layer = "a", .suffix = "n"}) + .output_product_suffixes("count"); + g.observe( + "check_a", + [&counts](handle count) { + counts.emplace_back(count.data_cell_index().layer_path().to_string(), *count); + }, + concurrency::serial) + .input_family(product_selector{.creator = "count_a", .layer = "spill", .suffix = "count"}); + g.fold( + "count_c", [](int& count, unsigned int) { ++count; }, concurrency::serial, "run") + .input_family(product_selector{.creator = "split_c", .layer = "c", .suffix = "n"}) + .output_product_suffixes("count"); + g.observe( + "check_c", + [&extra_counts](int count) { extra_counts.push_back(count); }, + concurrency::serial) + .input_family(product_selector{.creator = "count_c", .layer = "run", .suffix = "count"}); + + g.execute(); + CHECK(g.execution_count("split_a") == 5); + CHECK(g.execution_count("split_c") == 2); + CHECK(g.execution_count("count_a") == 12); + CHECK(g.execution_count("count_c") == 6); + CHECK(g.execution_count("check_a") == 5); + CHECK(g.execution_count("check_c") == 1); + return std::make_pair(std::move(counts), std::move(extra_counts)); + }; + // Destroy the graph before checking for cached-entry warnings. + auto [counts, extra_counts] = run_graph(); + std::ranges::sort(counts); + CHECK(counts == std::vector>{{"/job/run/spill", 3}, + {"/job/run/spill", 3}, + {"/job/spill", 2}, + {"/job/spill", 2}, + {"/job/spill", 2}}); + CHECK(extra_counts == std::vector{6}); + check_no_cached_entries(output); +} From 0df499d36bc6a5b8e7abad3b8d8c05714d8344ee Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Wed, 14 Oct 2026 13:02:32 -0500 Subject: [PATCH 5/6] fix(core): simplify unfold path expansion and reject output collisions --- phlex/core/index_router.cpp | 125 ++++++++++++++++++++---------------- phlex/core/index_router.hpp | 1 + test/index_router_test.cpp | 49 +++++++++++++- 3 files changed, 119 insertions(+), 56 deletions(-) diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index 8d4beafd6..ad68ddffe 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -21,8 +21,8 @@ #include #include #include +#include #include -#include #include #include #include @@ -36,9 +36,10 @@ namespace phlex::detail { // Build the complete set of paths implied by the driver's declarations by adding each path's // intermediate parents. Incomplete paths are rooted at /job; for example, declarations // {"/job/run/spill", "spill"} produce /job, /job/run, /job/run/spill, and /job/spill. - std::set driver_layer_paths(std::vector const& layer_paths_from_driver) + std::map> driver_layer_paths( + std::vector const& layer_paths_from_driver) { - std::set paths{layer_path{"/job"}}; + std::map> paths{{layer_path{"/job"}, std::nullopt}}; for (auto const& path : layer_paths_from_driver) { // Driver declarations may contain only leaves. Include their intermediate parents too. std::vector prefix; @@ -47,7 +48,7 @@ namespace phlex::detail { } for (auto const& component : path.components()) { prefix.push_back(component); - paths.emplace(prefix); + paths.try_emplace(layer_path{prefix}, std::nullopt); } } return paths; @@ -213,67 +214,83 @@ namespace phlex::detail { build_multilayer_join_slots(g, multilayer_join_ports); } - // -------------------------------------------------------------------------------------------- - // Resolve each unfold on compatible ancestor chains. Its deepest input is the only parent - // that receives its flush, so paths and expectations must be derived from the same application. - // Append the output layer to each matching parent path and iterate to a fixed point, allowing - // chained unfolds to resolve even when a dependent is registered before its producer. - // An acyclic chain adds at most one layer per unfold node; expansion beyond the initial deepest - // path plus the number of unfolds indicates a cycle and fails with the offending node names. - void index_router::establish_layer_hierarchy( - std::vector const& layer_paths_from_driver, unfold_data const& unfolds) - { - auto paths = driver_layer_paths(layer_paths_from_driver); - sorted_layer_paths_.assign_range(paths); - assert(!sorted_layer_paths_.empty()); + namespace { + struct unfold_layer_paths { + std::vector sorted_paths; + std::map number_unfolds_per_parent_path; + }; - std::size_t const initial_deepest_path_depth = - std::ranges::max_element(sorted_layer_paths_, {}, &layer_path::depth)->depth(); - std::size_t const max_allowed_depth = initial_deepest_path_depth + unfolds.size(); - std::set> counted_unfold_parents; - - bool changed = true; - while (changed) { - changed = false; - std::set offending_unfolds; - for (auto const& unfold : unfolds) { - // Snapshot the current size so paths appended for this unfold are not revisited until - // the next fixed-point iteration. In particular, self-referential unfolds stay bounded. - std::size_t const snapshot = sorted_layer_paths_.size(); - for (std::size_t i = 0; i < snapshot; ++i) { - auto const& parent_path = sorted_layer_paths_[i]; + // Resolve each unfold on compatible ancestor chains. Its deepest input is the only parent + // that receives its flush, so paths and expectations must be derived from the same application. + // Visit each discovered path once and append new children to the worklist, allowing chained + // unfolds to resolve even when a dependent is registered before its producer. + // An acyclic chain adds at most one layer per unfold node; expansion beyond the initial deepest + // path plus the number of unfolds indicates a cycle and fails with the offending node name. + unfold_layer_paths expand_unfold_layer_paths( + std::vector const& layer_paths_from_driver, + index_router::unfold_data const& unfolds) + { + unfold_layer_paths result; + auto paths = driver_layer_paths(layer_paths_from_driver); + auto layer_paths = paths | std::views::keys | std::ranges::to(); + + assert(not layer_paths.empty()); + std::size_t const initial_deepest_path_depth = + std::ranges::max_element(layer_paths, {}, &layer_path::depth)->depth(); + std::size_t const max_allowed_depth = initial_deepest_path_depth + unfolds.size(); + + // N.B. layer_paths.size() is purposefully evaluated on each iteration, since new children + // are appended to the worklist. + for (std::size_t i = 0; i < layer_paths.size(); ++i) { + // Appending children can reallocate the worklist, so keep a copy of the parent. + auto const parent_path = layer_paths[i]; + for (auto const& unfold : unfolds) { if (not matches_input_chain(parent_path, unfold.input_layers)) { continue; } - layer_path candidate{fmt::format("{}/{}", parent_path, unfold.output_layer)}; + auto components = parent_path.components() | std::ranges::to(); + components.push_back(unfold.output_layer); + layer_path candidate{std::move(components)}; if (candidate.depth() > max_allowed_depth) { - offending_unfolds.insert(unfold.name); - continue; - } - // Count each unfold once per parent across iterations, even when multiple unfolds - // produce the same child path: each still sends its own flush message. - if (counted_unfold_parents.emplace(unfold.name, parent_path.hash()).second) { - ++number_unfolds_per_parent_path_[parent_path.hash()]; + throw std::runtime_error(fmt::format("Unfold layer hierarchy expansion exceeded max " + "depth {} (initial deepest {} + {} unfold " + "node(s)). Offending unfold: {}", + max_allowed_depth, + initial_deepest_path_depth, + unfolds.size(), + unfold.name)); } - if (paths.insert(candidate).second) { - sorted_layer_paths_.push_back(std::move(candidate)); - changed = true; + auto const [producer, inserted] = paths.try_emplace(candidate, unfold.name); + if (not inserted) { + if (not producer->second) { + throw std::runtime_error(fmt::format( + "Unfold '{}' specifies output layer '{}' already declared by the driver.", + unfold.name, + candidate)); + } + throw std::runtime_error( + fmt::format("Unfolds '{}' and '{}' specify the same output layer '{}'.", + *producer->second, + unfold.name, + candidate)); } + ++result.number_unfolds_per_parent_path[parent_path.hash()]; + layer_paths.push_back(std::move(candidate)); } } - - if (not offending_unfolds.empty()) { - throw std::runtime_error(fmt::format( - "Unfold layer hierarchy expansion exceeded max depth {} (initial deepest {} + {} unfold " - "node(s)). Offending unfold(s):\n{}", - max_allowed_depth, - initial_deepest_path_depth, - unfolds.size(), - bulleted_list(offending_unfolds))); - } + result.sorted_paths.assign_range(paths | std::views::keys); + return result; } + } // namespace - std::ranges::sort(sorted_layer_paths_); + // -------------------------------------------------------------------------------------------- + void index_router::establish_layer_hierarchy( + std::vector const& layer_paths_from_driver, unfold_data const& unfolds) + { + auto result = expand_unfold_layer_paths(layer_paths_from_driver, unfolds); + sorted_layer_paths_ = std::move(result.sorted_paths); + number_unfolds_per_parent_path_ = std::move(result.number_unfolds_per_parent_path); + assert(!sorted_layer_paths_.empty()); // In sorted order, a path can only be a prefix of paths that follow it. for (std::size_t i = 0; i < sorted_layer_paths_.size(); ++i) { diff --git a/phlex/core/index_router.hpp b/phlex/core/index_router.hpp index 82d247561..06da526b7 100644 --- a/phlex/core/index_router.hpp +++ b/phlex/core/index_router.hpp @@ -84,6 +84,7 @@ namespace phlex::detail { explicit index_router(tbb::flow::graph& g); data_cell_index_ptr route(data_cell_index_ptr const& index, index_flushes const& flushes); + // One entry per uniquely named unfold node, as guaranteed by framework registration. using unfold_data = std::vector; // Establishes the layer hierarchy, registers unfold flush expectations, and wires the TBB diff --git a/test/index_router_test.cpp b/test/index_router_test.cpp index c60687808..e0a1630d2 100644 --- a/test/index_router_test.cpp +++ b/test/index_router_test.cpp @@ -299,7 +299,7 @@ TEST_CASE("Unfold hierarchy rejects cyclic layer production", "[graph][unfold]") unfolds = {{.name = "split_run", .input_layers = {"run"_id}, .output_layer = "run"_id}}; expected_error = "Unfold layer hierarchy expansion exceeded max depth 3 (initial deepest 2 + 1 unfold " - "node(s)). Offending unfold(s):\n - split_run"; + "node(s)). Offending unfold: split_run"; } SECTION("Two-unfold cycle") @@ -308,7 +308,7 @@ TEST_CASE("Unfold hierarchy rejects cyclic layer production", "[graph][unfold]") {.name = "split_run", .input_layers = {"spill"_id}, .output_layer = "run"_id}}; expected_error = "Unfold layer hierarchy expansion exceeded max depth 4 (initial deepest 2 + 2 unfold " - "node(s)). Offending unfold(s):\n - split_spill"; + "node(s)). Offending unfold: split_spill"; } CHECK_THROWS_WITH( @@ -320,3 +320,48 @@ TEST_CASE("Unfold hierarchy rejects cyclic layer production", "[graph][unfold]") {}), expected_error); } + +TEST_CASE("Unfold hierarchy rejects driver output layer collisions", "[graph][unfold]") +{ + using namespace phlex::experimental::literals; + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + detail::index_router router{graph}; + auto const driver_path = GENERATE("/job/run/spill", "/job/run/spill/event"); + CAPTURE(driver_path); + + CHECK_THROWS_WITH( + router.finalize(graph, + {experimental::layer_path{driver_path}}, + {{.name = "split", .input_layers = {"run"_id}, .output_layer = "spill"_id}}, + {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, + {}, + {}), + "Unfold 'split' specifies output layer '/job/run/spill' already declared by the driver."); +} + +TEST_CASE("Unfold hierarchy rejects duplicate output layers", "[graph][unfold]") +{ + using namespace phlex::experimental::literals; + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + detail::index_router router{graph}; + + auto const generated_parent = GENERATE(false, true); + CAPTURE(generated_parent); + detail::index_router::unfold_data unfolds{ + {.name = "split_a", .input_layers = {"run"_id}, .output_layer = "spill"_id}, + {.name = "split_b", .input_layers = {"run"_id}, .output_layer = "spill"_id}}; + if (generated_parent) { + unfolds.push_back({.name = "split_run", .input_layers = {"job"_id}, .output_layer = "run"_id}); + } + + CHECK_THROWS_WITH( + router.finalize(graph, + {experimental::layer_path{generated_parent ? "/job" : "/job/run"}}, + unfolds, + {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, + {}, + {}), + "Unfolds 'split_a' and 'split_b' specify the same output layer '/job/run/spill'."); +} From 20df5ad343581b2c136e17c1626367144c4a0936 Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Wed, 14 Oct 2026 16:19:41 -0500 Subject: [PATCH 6/6] fix(core): reject undeclared index paths and initialize driver-only graphs --- phlex/core/framework_graph.cpp | 6 ----- phlex/core/index_router.cpp | 14 ++++-------- phlex/core/index_router.hpp | 5 ++-- test/index_router_test.cpp | 42 ++++++++++++++++++++++++++++------ 4 files changed, 41 insertions(+), 26 deletions(-) diff --git a/phlex/core/framework_graph.cpp b/phlex/core/framework_graph.cpp index ebeff0080..381942207 100644 --- a/phlex/core/framework_graph.cpp +++ b/phlex/core/framework_graph.cpp @@ -249,12 +249,6 @@ namespace phlex::detail { auto [provider_input_ports, multilayer_join_index_ports] = make_computational_edges(nodes_, filters_, graph_, stage_); - if (provider_input_ports.empty()) { - assert(multilayer_join_index_ports.empty()); - // No algorithms downstream of source. - return; - } - // Index-router finalization makes edges between the index-set nodes and the provider nodes. index_router_.finalize(graph_, fixed_hierarchy_.layer_paths(), diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index ad68ddffe..f4a0ef957 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -205,9 +205,6 @@ namespace phlex::detail { std::map const& multilayer_join_ports) { using namespace phlex::experimental::literals; - // We must have at least one provider port, or there can be no data to process. - assert(!provider_input_ports.empty()); - establish_layer_hierarchy(layer_paths_from_driver, unfolds); wire_provider_index_sets(g, std::move(provider_input_ports)); wire_fold_partition_index_sets(g, std::move(fold_partition_ports)); @@ -299,7 +296,7 @@ namespace phlex::detail { i + 1 == sorted_layer_paths_.size() or not sorted_layer_paths_[i].is_strict_prefix_of(sorted_layer_paths_[i + 1]); // Record every known layer, both lowest and non-lowest. Pre-populating the lowest entries - // lets index_is_lowest_layer() return a definitive answer without its unknown-path fallback, + // lets index_is_lowest_layer() return a definitive answer for every declared path, // including for unfold outputs that themselves parent another unfold's children. is_lowest_layer_hashes_.emplace(layer_hash, is_lowest_layer); } @@ -416,18 +413,15 @@ namespace phlex::detail { void index_router::drain(index_flushes const& flushes) { update_flush_counts(flushes); } - bool index_router::index_is_lowest_layer(data_cell_index_ptr const& index) + bool index_router::index_is_lowest_layer(data_cell_index_ptr const& index) const { auto it = is_lowest_layer_hashes_.find(index->layer_hash()); if (it != is_lowest_layer_hashes_.end()) { return it->second; } - // Unknown layer hash: establish_layer_hierarchy() includes driver paths and unfold-produced - // descendants, so a hash absent from is_lowest_layer_hashes_ corresponds to a layer the router - // was never told about. Treating it as lowest skips the rollup/expected-count bookkeeping that - // requires path knowledge and matches the prior behavior for unfold output layers. - return is_lowest_layer_hashes_.emplace(index->layer_hash(), true).first->second; + throw std::runtime_error(fmt::format( + "Index layer path '{}' is not in the established layer hierarchy.", index->layer_path())); } internal::index_set_node_ptr index_router::index_set_node_for(data_cell_index_ptr const& index) diff --git a/phlex/core/index_router.hpp b/phlex/core/index_router.hpp index 06da526b7..fbfbd5934 100644 --- a/phlex/core/index_router.hpp +++ b/phlex/core/index_router.hpp @@ -114,10 +114,9 @@ namespace phlex::detail { data_cell_index_ptr route(data_cell_index_ptr const& index, bool is_lowest_layer, std::size_t message_id); - bool index_is_lowest_layer(data_cell_index_ptr const& index); + bool index_is_lowest_layer(data_cell_index_ptr const& index) const; // Hash-only lookup for child paths from flush messages, where no data_cell_index is available. - // Returns the cached classification for known paths; unknown hashes default to lowest, - // consistently with index_is_lowest_layer()'s fallback. + // Returns the cached classification for known paths; unknown hashes return true. bool is_lowest_layer_hash(std::size_t layer_hash) const; // finalize() helpers — each owns one initialization step. diff --git a/test/index_router_test.cpp b/test/index_router_test.cpp index e0a1630d2..0f44c36a4 100644 --- a/test/index_router_test.cpp +++ b/test/index_router_test.cpp @@ -1,5 +1,4 @@ #include "phlex/core/index_router.hpp" -#include "phlex/core/join_registration.hpp" #include "phlex/core/message.hpp" #include "phlex/model/data_cell_counts.hpp" #include "phlex/model/data_cell_index.hpp" @@ -20,6 +19,30 @@ using namespace phlex; +TEST_CASE("Index routing rejects undeclared layer paths", "[graph]") +{ + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + detail::index_router router{graph}; + router.finalize( + graph, + {experimental::layer_path{"/job/run/event"}, experimental::layer_path{"/job/spill"}}, + {}, + {{"dummy", {.input_product = {.layer = "event"}, .port = &indices}}}, + {}, + {}); + + auto job = data_cell_index::job(); + auto run = job->make_child("run", 0); + CHECK_NOTHROW(router.route(run->make_child("event", 0), {})); + auto unknown = job->make_child("spill", 0)->make_child("event", 0); + // A known terminal name does not make a different parent path eligible. + CHECK_THROWS_WITH( + router.route(unknown, {}), + "Index layer path '/job/spill/event' is not in the established layer hierarchy."); + graph.wait_for_all(); +} + TEST_CASE("Duplicate counting paths deliver one combined end token", "[graph][issue955]") { using namespace phlex::experimental::literals; @@ -324,15 +347,17 @@ TEST_CASE("Unfold hierarchy rejects cyclic layer production", "[graph][unfold]") TEST_CASE("Unfold hierarchy rejects driver output layer collisions", "[graph][unfold]") { using namespace phlex::experimental::literals; + using experimental::layer_path; tbb::flow::graph graph; tbb::flow::queue_node indices{graph}; detail::index_router router{graph}; - auto const driver_path = GENERATE("/job/run/spill", "/job/run/spill/event"); + auto const driver_path = + GENERATE(layer_path{"/job/run/spill"}, layer_path{"/job/run/spill/event"}); CAPTURE(driver_path); CHECK_THROWS_WITH( router.finalize(graph, - {experimental::layer_path{driver_path}}, + {driver_path}, {{.name = "split", .input_layers = {"run"_id}, .output_layer = "spill"_id}}, {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, {}, @@ -343,22 +368,25 @@ TEST_CASE("Unfold hierarchy rejects driver output layer collisions", "[graph][un TEST_CASE("Unfold hierarchy rejects duplicate output layers", "[graph][unfold]") { using namespace phlex::experimental::literals; + using experimental::layer_path; tbb::flow::graph graph; tbb::flow::queue_node indices{graph}; detail::index_router router{graph}; - auto const generated_parent = GENERATE(false, true); - CAPTURE(generated_parent); + auto const driver_path = GENERATE(layer_path{"/job/run"}, layer_path{"/job"}); + CAPTURE(driver_path); detail::index_router::unfold_data unfolds{ {.name = "split_a", .input_layers = {"run"_id}, .output_layer = "spill"_id}, {.name = "split_b", .input_layers = {"run"_id}, .output_layer = "spill"_id}}; - if (generated_parent) { + if (driver_path == layer_path{"/job"}) { + // When the driver path is '/job', an unfold is needed to create the 'run' layer as + // 'split_a' and 'split_b' both depend on it. unfolds.push_back({.name = "split_run", .input_layers = {"job"_id}, .output_layer = "run"_id}); } CHECK_THROWS_WITH( router.finalize(graph, - {experimental::layer_path{generated_parent ? "/job" : "/job/run"}}, + {driver_path}, unfolds, {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, {},