From 601224b3abcd541dad78e0ddf1ad4b5cb69c4ed8 Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Fri, 16 Oct 2026 09:20:42 -0500 Subject: [PATCH 1/2] fix(core): resolve hierarchy paths and enforce driver declarations Normalize driver paths and their intermediate ancestors, and expand grouped unfold inputs with a worklist. Record expectations per parent path and reject cycles, output collisions, and undeclared runtime paths. Give default drivers a job-only hierarchy, reject explicitly empty declarations, and initialize routing for driver-only graphs. Include the hierarchy, collision, and declaration regressions with these prerequisites. --- phlex/core/framework_graph.cpp | 43 ++---- phlex/core/index_router.cpp | 225 ++++++++++++++++------------- phlex/core/index_router.hpp | 98 +++++-------- phlex/model/fixed_hierarchy.cpp | 8 +- phlex/model/fixed_hierarchy.hpp | 2 +- phlex/model/layer_path.cpp | 5 + phlex/model/layer_path.hpp | 6 + phlex/utilities/simple_ptr_map.hpp | 6 +- test/CMakeLists.txt | 9 ++ test/fixed_hierarchy_test.cpp | 65 ++++++++- test/index_router_test.cpp | 175 ++++++++++++++++++++++ test/layer_generator_test.cpp | 8 + test/layer_path_test.cpp | 15 ++ 13 files changed, 467 insertions(+), 198 deletions(-) create mode 100644 test/index_router_test.cpp diff --git a/phlex/core/framework_graph.cpp b/phlex/core/framework_graph.cpp index 69f9125c0..381942207 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; } @@ -256,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 d187dcc7f..7b1ea44bd 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 @@ -18,11 +19,10 @@ #include #include #include +#include #include -#include #include #include -#include #include #include @@ -30,6 +30,43 @@ 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::map> driver_layer_paths( + std::vector const& layer_paths_from_driver) + { + 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; + if (not path.is_complete()) { + prefix.emplace_back("job"); + } + for (auto const& component : path.components()) { + prefix.push_back(component); + paths.try_emplace(layer_path{prefix}, std::nullopt); + } + } + 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,82 +150,99 @@ 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) { 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(std::move(layer_paths_from_driver), unfolds.layer_pairs); - unfold_count_per_input_layer_ = std::move(unfolds.count_per_input_layer); + if (layer_paths_from_driver.empty()) { + throw std::runtime_error("Cannot finalize index_router with an empty driver hierarchy."); + } + 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) - { - 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(); - - 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::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)) { + namespace { + struct unfold_layer_paths { + std::vector sorted_paths; + std::map number_unfolds_per_parent_path; + }; + + // 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{parent_path.to_string() + "/" + std::string(output_name)}; + 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_pairs.insert(fmt::format("{} -> {}", input_layer, output_layer)); - continue; + 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 (not std::ranges::contains(sorted_layer_paths_, candidate)) { - 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_pairs.empty()) { - throw std::runtime_error(fmt::format( - "Unfold layer hierarchy expansion exceeded max depth {} (initial deepest {} + {} unfold " - "pair(s)). Offending unfold pair(s):\n{}", - max_allowed_depth, - initial_deepest_path_depth, - layer_pairs.size(), - bulleted_list(offending_pairs))); - } + 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) { @@ -197,9 +251,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 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); } } @@ -224,24 +277,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) { @@ -304,19 +346,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_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. - 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) @@ -401,9 +439,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 +461,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 +492,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 +502,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 +579,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..18d7f6be4 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,30 @@ 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; - }; + // 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 + // 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); @@ -137,18 +124,15 @@ 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); - // 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. + 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 return true. 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 descendant count per receiving slot and counting path. 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/fixed_hierarchy.cpp b/phlex/model/fixed_hierarchy.cpp index 9ac859e7b..da4c18d4c 100644 --- a/phlex/model/fixed_hierarchy.cpp +++ b/phlex/model/fixed_hierarchy.cpp @@ -106,6 +106,8 @@ namespace phlex { // ================================================================================ // fixed_hierarchy implementation + fixed_hierarchy::fixed_hierarchy() : fixed_hierarchy{{"job"}} {} + fixed_hierarchy::fixed_hierarchy(std::initializer_list> layer_paths) : fixed_hierarchy(std::vector>(layer_paths)) { @@ -115,6 +117,9 @@ namespace phlex { layer_paths_(convert_vector_vector_string(std::move(layer_paths))), layer_hashes_(std::from_range, build_hashes(layer_paths_)) { + if (layer_paths_.empty()) { + throw std::runtime_error("A fixed hierarchy must declare at least one layer path."); + } } // The layer_paths variable is directly moved from when creating the merged set. @@ -130,9 +135,6 @@ namespace phlex { void fixed_hierarchy::validate(data_cell_index_ptr const& index) const { - if (layer_hashes_.empty()) { - return; - } if (std::ranges::binary_search(layer_hashes_, index->layer_hash())) { return; } diff --git a/phlex/model/fixed_hierarchy.hpp b/phlex/model/fixed_hierarchy.hpp index 086544de9..e944a1b74 100644 --- a/phlex/model/fixed_hierarchy.hpp +++ b/phlex/model/fixed_hierarchy.hpp @@ -66,7 +66,7 @@ namespace phlex { class PHLEX_MODEL_EXPORT fixed_hierarchy { public: - fixed_hierarchy() = default; + fixed_hierarchy(); // Using an std::initializer_list removes one set of braces that the // user must provide; leaving it explicit preserves clarity of // expression for callers. 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/CMakeLists.txt b/test/CMakeLists.txt index c38165f23..aa63aa115 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/fixed_hierarchy_test.cpp b/test/fixed_hierarchy_test.cpp index de6e2c8ce..71432f8d0 100644 --- a/test/fixed_hierarchy_test.cpp +++ b/test/fixed_hierarchy_test.cpp @@ -6,9 +6,12 @@ #include "phlex/utilities/resumable_driver.hpp" // IWYU pragma: keep #include +#include #include #include +#include +#include #include using namespace phlex; @@ -36,18 +39,72 @@ namespace { TEST_CASE("Ill-formed paths result in an exception", "[fixed_hierarchy]") { + CHECK_THROWS_WITH(fixed_hierarchy(std::vector>{}), + ContainsSubstring("A fixed hierarchy must declare at least one layer path")); + CHECK_THROWS_WITH(fixed_hierarchy(std::initializer_list>{}), + ContainsSubstring("A fixed hierarchy must declare at least one layer path")); CHECK_THROWS_WITH(fixed_hierarchy{{}}, ContainsSubstring("Layer paths cannot be empty")); CHECK_THROWS_WITH((fixed_hierarchy{{"job", "run", "job"}}), ContainsSubstring("Layer paths may only contain 'job' as the first element")); } -TEST_CASE("Default-constructed fixed_hierarchy accepts any index", "[fixed_hierarchy]") +TEST_CASE("Default-constructed fixed_hierarchy accepts only job", "[fixed_hierarchy]") { - fixed_hierarchy const h; + fixed_hierarchy h; auto const job = data_cell_index::job(); + CHECK(h.layer_paths() == std::vector{"job"}); CHECK_NOTHROW(h.validate(job)); - CHECK_NOTHROW(h.validate(job->make_child("run", 0))); - CHECK_NOTHROW(h.validate(job->make_child("unrelated", 0))); + CHECK_THROWS(h.validate(job->make_child("run", 0))); + CHECK_THROWS(h.validate(job->make_child("unrelated", 0))); + + SECTION("Empty update preserves job-only hierarchy") + { + h.update({}); + CHECK(h.layer_paths() == std::vector{"job"}); + CHECK_NOTHROW(h.validate(job)); + CHECK_THROWS(h.validate(job->make_child("run", 0))); + } + + SECTION("Valid update extends the hierarchy") + { + h.update({"run/subrun"}); + CHECK(h.layer_paths() == std::vector{"run/subrun", "job"}); + CHECK_NOTHROW(h.validate(job)); + CHECK_NOTHROW(h.validate(job->make_child("run", 0))); + CHECK_NOTHROW(h.validate(job->make_child("run", 0)->make_child("subrun", 0))); + CHECK_THROWS(h.validate(job->make_child("unrelated", 0))); + } + + SECTION("Job-only update preserves job-only hierarchy") + { + h.update({"/job"}); + CHECK(h.layer_paths() == std::vector{"job"}); + CHECK_NOTHROW(h.validate(job)); + CHECK_THROWS(h.validate(job->make_child("run", 0))); + } +} + +TEST_CASE("Job-only hierarchy accepts and yields only job", "[fixed_hierarchy]") +{ + auto const h = GENERATE(fixed_hierarchy{}, fixed_hierarchy{{"job"}}); + auto const job = data_cell_index::job(); + CHECK(h.layer_paths() == std::vector{"job"}); + CHECK_NOTHROW(h.validate(job)); + CHECK_THROWS(h.validate(job->make_child("run", 0))); + + auto const use_cursor = GENERATE(false, true); + detail::framework_driver driver{[&h, use_cursor](detail::framework_driver& d) { + if (use_cursor) { + h.yield_job(d); + } else { + h.yielder(d)(data_cell_index::job()); + } + }}; + std::vector yielded; + while (auto index = driver()) { + yielded.push_back((*index)->layer_path()); + } + CHECK(yielded == std::vector{"/job"}); } TEST_CASE("Fixed hierarchy accepts valid layer indices", "[fixed_hierarchy]") diff --git a/test/index_router_test.cpp b/test/index_router_test.cpp new file mode 100644 index 000000000..7971a5626 --- /dev/null +++ b/test/index_router_test.cpp @@ -0,0 +1,175 @@ +#include "phlex/core/index_router.hpp" +#include "phlex/core/message.hpp" +#include "phlex/model/data_cell_index.hpp" +#include "phlex/model/layer_path.hpp" + +#include +#include +#include +#include + +#include +#include + +using namespace phlex; + +TEST_CASE("Index routing requires an explicit driver hierarchy", "[graph]") +{ + tbb::flow::graph graph; + detail::index_router router{graph}; + CHECK_THROWS_WITH(router.finalize(graph, {}, {}, {}, {}, {}), + "Cannot finalize index_router with an empty driver hierarchy."); + CHECK_NOTHROW(router.finalize(graph, {experimental::layer_path{"/job"}}, {}, {}, {}, {})); + CHECK_NOTHROW(router.route(data_cell_index::job(), {})); + CHECK_THROWS_WITH(router.route(data_cell_index::job()->make_child("run", 0), {}), + "Index layer path '/job/run' is not in the established layer hierarchy."); + graph.wait_for_all(); +} + +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); + 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("Index routing roots relative driver paths at the job layer", "[graph]") +{ + tbb::flow::graph graph; + detail::index_router router{graph}; + router.finalize(graph, {experimental::layer_path{"run/event"}}, {}, {}, {}, {}); + + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + auto const event = run->make_child("event", 0); + CHECK_NOTHROW(router.route(job, {})); + CHECK_NOTHROW(router.route(run, {})); + CHECK_NOTHROW(router.route(event, {})); + CHECK_THROWS_WITH(router.route(job->make_child("event", 0), {}), + "Index layer path '/job/event' is not in the established layer hierarchy."); + graph.wait_for_all(); +} + +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}; + detail::index_router router{graph}; + router.finalize( + graph, + {experimental::layer_path{"/job/spill"}}, + {{.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}}}, + {}, + {}); + + auto spill = data_cell_index::job()->make_child("spill", 0); + auto a = spill->make_child("a", 0); + auto b = a->make_child("b", 0); + CHECK_NOTHROW(router.route(spill, {})); + CHECK_NOTHROW(router.route(a, {})); + CHECK_NOTHROW(router.route(b, {})); + graph.wait_for_all(); +} + +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: 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: split_spill"; + } + + CHECK_THROWS_WITH( + router.finalize(graph, + {experimental::layer_path{"/job/run"}}, + unfolds, + {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, + {}, + {}), + expected_error); +} + +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(layer_path{"/job/run/spill"}, layer_path{"/job/run/spill/event"}); + CAPTURE(driver_path); + + CHECK_THROWS_WITH( + router.finalize(graph, + {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; + using experimental::layer_path; + tbb::flow::graph graph; + tbb::flow::queue_node indices{graph}; + detail::index_router router{graph}; + + 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 (driver_path == layer_path{"/job"}) { + unfolds.push_back({.name = "split_run", .input_layers = {"job"_id}, .output_layer = "run"_id}); + } + + CHECK_THROWS_WITH( + router.finalize(graph, + {driver_path}, + unfolds, + {{"dummy", {.input_product = {.layer = "run"}, .port = &indices}}}, + {}, + {}), + "Unfolds 'split_a' and 'split_b' specify the same output layer '/job/run/spill'."); +} diff --git a/test/layer_generator_test.cpp b/test/layer_generator_test.cpp index adb7bdcc2..aa844ad59 100644 --- a/test/layer_generator_test.cpp +++ b/test/layer_generator_test.cpp @@ -1,10 +1,14 @@ #include "phlex/core/framework_graph.hpp" +#include "phlex/model/data_cell_index.hpp" +#include "phlex/model/layer_path.hpp" #include "plugins/layer_generator.hpp" #include #include #include +#include + using namespace phlex::experimental; using phlex::detail::framework_graph; using namespace Catch::Matchers; @@ -12,6 +16,10 @@ using namespace Catch::Matchers; TEST_CASE("Only job layer", "[layer-generation]") { auto gen = layer_generator::make(); + auto const hierarchy = gen->hierarchy(); + CHECK(hierarchy.layer_paths() == std::vector{"job"}); + CHECK_NOTHROW(hierarchy.validate(phlex::data_cell_index::job())); + CHECK_THROWS(hierarchy.validate(phlex::data_cell_index::job()->make_child("spill", 0))); auto g = framework_graph::without_driver("test"); g.add_driver(gen); 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 4815e3f6bdc123899d351e6038554653a4ffef4f Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Fri, 16 Oct 2026 09:26:21 -0500 Subject: [PATCH 2/2] fix(core): make partition completion aggregate and order-independent Deliver one completion token per receiving slot, summing applicable descendant counts. Retain zero-count completion state until its product arrives so arrival order cannot strand cached products. Report zero-count driver-owned children when empty partitions close, without synthesizing counts for children owned by unfolds. Cover repeated layer names, empty partitions, sparse counts, and both arrival orders. Preserves work originally aided by GPT-6 Luna. --- phlex/core/detail/repeater_node.cpp | 6 +- phlex/core/framework_graph.cpp | 11 +- phlex/core/framework_graph.hpp | 2 +- phlex/core/index_router.cpp | 12 +- phlex/core/index_router.hpp | 8 +- phlex/core/message.hpp | 5 +- phlex/model/data_cell_tracker.cpp | 188 ++++++++--------- phlex/model/data_cell_tracker.hpp | 16 +- test/data_cell_tracker_test.cpp | 258 ++++++++++++++++++++++-- test/fold_duplicate_layer_name_test.cpp | 184 +++++++++++++---- test/index_router_test.cpp | 97 ++++++++- test/repeater_node_test.cpp | 26 ++- 12 files changed, 623 insertions(+), 190 deletions(-) diff --git a/phlex/core/detail/repeater_node.cpp b/phlex/core/detail/repeater_node.cpp index a8aecb4e6..879746387 100644 --- a/phlex/core/detail/repeater_node.cpp +++ b/phlex/core/detail/repeater_node.cpp @@ -191,7 +191,11 @@ 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) { + // Every flush, including a zero-count flush, must be matched with a product. + // 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/framework_graph.cpp b/phlex/core/framework_graph.cpp index 381942207..64920ad3e 100644 --- a/phlex/core/framework_graph.cpp +++ b/phlex/core/framework_graph.cpp @@ -107,7 +107,7 @@ namespace phlex::detail { [this](tbb::flow_control& fc) mutable -> ready_flushes_then_emit { assert(driver_); if (auto item = (*driver_)()) { - return {.ready_flushes = cell_tracker_.report_and_evict_ready_flushes(*item), + return {.ready_flushes = cell_tracker_->report_and_evict_ready_flushes(*item), .index_to_emit = *item}; } fc.stop(); @@ -153,9 +153,9 @@ namespace phlex::detail { framework_graph::~framework_graph() { - if (shutdown_on_error_) { + if (shutdown_on_error_ and cell_tracker_) { // When in an error state, we need to pop the layer stack and wait for any tasks to finish. - auto remaining_flushes = cell_tracker_.report_and_evict_ready_flushes(nullptr); + auto remaining_flushes = cell_tracker_->report_and_evict_ready_flushes(nullptr); index_router_.drain(remaining_flushes); graph_.wait_for_all(); } @@ -201,7 +201,9 @@ namespace phlex::detail { graph_.wait_for_all(); // Now back out of all remaining layers - index_router_.drain(cell_tracker_.report_and_evict_ready_flushes(nullptr)); + assert(cell_tracker_); + // NOLINTNEXTLINE(bugprone-unchecked-optional-access) + index_router_.drain(cell_tracker_->report_and_evict_ready_flushes(nullptr)); graph_.wait_for_all(); } @@ -243,6 +245,7 @@ namespace phlex::detail { void framework_graph::finalize() { throw_if_registration_errors(); + cell_tracker_.emplace(fixed_hierarchy_.layer_paths()); make_filter_edges(); make_bookkeeping_edges(); diff --git a/phlex/core/framework_graph.hpp b/phlex/core/framework_graph.hpp index b38eb45dc..9a951eb5b 100644 --- a/phlex/core/framework_graph.hpp +++ b/phlex/core/framework_graph.hpp @@ -227,7 +227,7 @@ namespace phlex::detail { tbb::flow::graph graph_{}; std::optional driver_; std::vector registration_errors_; - data_cell_tracker cell_tracker_; + std::optional cell_tracker_; tbb::flow::input_node src_; index_router index_router_; tbb::flow::function_node diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index 7b1ea44bd..623b29801 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 @@ -334,7 +335,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}); } }); @@ -439,7 +443,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 @@ -462,9 +466,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 18d7f6be4..0cb2d6ede 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/phlex/model/data_cell_tracker.cpp b/phlex/model/data_cell_tracker.cpp index 532c28f67..64acdfd80 100644 --- a/phlex/model/data_cell_tracker.cpp +++ b/phlex/model/data_cell_tracker.cpp @@ -3,36 +3,71 @@ #include "phlex/model/data_cell_counts.hpp" #include "phlex/model/flush_messages.hpp" #include "phlex/model/fwd.hpp" +#include "phlex/model/identifier.hpp" +#include "phlex/model/layer_path.hpp" +#include "phlex/utilities/hashing.hpp" #include +#include #include -#include +#include +#include +#include +#include #include #include #include #include #include +using phlex::experimental::layer_path; +using layer_hash_type = phlex::data_cell_index::hash_type; + namespace { - auto make_data_cell_counts(phlex::data_cell_index_ptr const& index) + auto make_driver_child_layer_hashes(std::vector const& layer_paths) { - auto result = std::make_shared(); - result->emplace(index->layer_hash(), 1); + std::map> result; + for (auto const& path : layer_paths) { + auto parent_hash = phlex::experimental::identifier{"job"}.hash(); + auto components = path.components(); + if (path.is_complete()) { + components = components.subspan(1); + } + for (auto const& component : components) { + auto const child_hash = phlex::detail::hash(parent_hash, component.hash()); + result[parent_hash].push_back(child_hash); + parent_hash = child_hash; + } + } + for (auto& child_hashes : result | std::views::values) { + std::ranges::sort(child_hashes); + child_hashes.erase(std::ranges::unique(child_hashes).begin(), child_hashes.end()); + } return result; } } namespace phlex::detail { + data_cell_tracker::data_cell_tracker(std::initializer_list const layer_paths) : + data_cell_tracker{std::vector(layer_paths)} + { + } + + data_cell_tracker::data_cell_tracker(std::vector const& layer_paths) : + driver_child_layer_hashes_{make_driver_child_layer_hashes(layer_paths)} + { + } + data_cell_tracker::~data_cell_tracker() { - if (pending_flushes_.empty()) { + if (active_families_.empty()) { return; } - spdlog::warn("Cached pending flushes at destruction:"); - for (auto const& [index, flush_counts] : pending_flushes_ | std::views::values) { - spdlog::warn(" Index: {}", index->to_string()); - for (auto const& [layer_hash, count] : *flush_counts) { + spdlog::warn("Active families at destruction:"); + for (auto const& family : active_families_) { + spdlog::warn(" Index: {}", family.index->to_string()); + for (auto const& [layer_hash, count] : *family.counts) { spdlog::warn(" {} = {}", layer_hash, count.load()); } } @@ -41,109 +76,64 @@ namespace phlex::detail { index_flushes data_cell_tracker::report_and_evict_ready_flushes( data_cell_index_ptr const& received_index) { - // Always update the cached index. The logic below uses the previous cached index to - // determine what flushes to emit. - auto cached_index = std::exchange(cached_index_, received_index); - - // Just beginning job (or ending a job that immediately threw an exception) - if (cached_index == nullptr) { - return {}; - } - - // Ending job. Backout to the job layer and emit flush tokens for all closed-out indices. - // - // Example: - // Current index: [run: 4, spill: 7] - // Received index: nullptr (end of job) - // Actions: - // a. Emit flush token for [run: 4, spill: 7] - // b. Emit flush token for [run: 4] - // c. Remove remaining flushes from cache if (received_index == nullptr) { - // The "std::move" empties the cache, so we don't need to manually clear it after. - return std::move(pending_flushes_) | std::views::values | std::ranges::to(); + index_flushes result; + result.reserve(active_families_.size()); + while (not active_families_.empty()) { + result.push_back(close_active_family()); + } + return result; } - assert(received_index); - assert(cached_index); - - // A parent must exist at this point - auto received_parent = received_index->parent(); - assert(received_parent); - - // Received index is immediate child of current index. - // - // Example: - // Current index: [run: 4] - // Received index: [run: 4, spill: 6] - // Actions: - // a. Initialize count for [run: 4] - if (received_parent->hash() == cached_index->hash()) { - create_parent_count(received_parent, received_index); + // The job index starts the stream and is the only valid index without a parent. + if (not received_index->has_parent()) { + Expects(received_index->layer_path() == layer_path{"/job"}); + if (not active_families_.empty()) { + throw std::runtime_error(fmt::format( + "Received root index {} while driver families are active", received_index->to_string())); + } + if (driver_child_layer_hashes_.contains(received_index->layer_hash())) { + active_families_.push_back( + {.index = received_index, .counts = std::make_shared()}); + } return {}; } - auto cached_parent = cached_index->parent(); - - // Received index is a sibling of the current index. Increment parent count and move - // current index to the new sibling. - // - // Example: - // Current index: [run: 4, spill: 6] - // Received index: [run: 4, spill: 7] - // Actions: - // a. Increment count for [run: 4] - if (cached_parent and received_parent->hash() == cached_parent->hash()) { - increment_parent_count(received_parent, received_index); - return {}; + auto const parent = received_index->parent(); + auto const parent_family = + std::ranges::find_if(active_families_, [&parent](index_flush const& family) { + return family.index->hash() == parent->hash(); + }); + if (parent_family == active_families_.end()) { + throw std::runtime_error( + fmt::format("Received index {}, whose parent {} does not identify an active driver family", + received_index->to_string(), + parent->to_string())); } - // Received index is a parent of the cached index. This means we've closed out the - // cached index and all of its siblings, and are moving back up the hierarchy. We need - // to emit flush tokens for all closed-out indices. - // - // Example: - // Cached index: [run: 4, spill: 6, subspill: 2, subsubspill: 3] - // Received index: [run: 4, spill: 7] - // Actions: - // a. Emit flush token for [run: 4, spill: 6, subspill: 2] - // b. Emit flush token for [run: 4, spill: 6] - // c. Remove relevant flush tokens from cache - // d. Increment count for [run: 4] index_flushes result; - auto recursive_parent = cached_parent; - while (recursive_parent != received_parent) { - if (recursive_parent == nullptr) { - // We will get here if the received parent is at a layer lower than the cached parent and is - // not an immediate child of the cached parent. The recursive parent walks all the way back - // up to the root without finding the received parent. This means the received index is not - // an ancestor of the cached index, and is invalid. - throw std::runtime_error( - fmt::format("Received index {}, which is not an immediate child of {}", - received_index->to_string(), - cached_index->to_string())); - } - auto fh = pending_flushes_.extract(recursive_parent->hash()); - result.push_back(fh.mapped()); - recursive_parent = recursive_parent->parent(); + while (std::next(parent_family) != active_families_.end()) { + result.push_back(close_active_family()); } - increment_parent_count(received_parent, received_index); - return result; - } - void data_cell_tracker::create_parent_count(data_cell_index_ptr const& parent, - data_cell_index_ptr const& child) - { - pending_flushes_.emplace(parent->hash(), - index_flush{.index = parent, .counts = make_data_cell_counts(child)}); + parent_family->counts->increment(received_index->layer_hash()); + if (driver_child_layer_hashes_.contains(received_index->layer_hash())) { + active_families_.push_back( + {.index = received_index, .counts = std::make_shared()}); + } + return result; } - void data_cell_tracker::increment_parent_count(data_cell_index_ptr const& parent, - data_cell_index_ptr const& child) + index_flush data_cell_tracker::close_active_family() { - auto it = pending_flushes_.find(parent->hash()); - // This is only called for siblings, so the parent count must already exist in the cache. - assert(it != pending_flushes_.end()); - it->second.counts->increment(child->layer_hash()); + auto family = std::move(active_families_.back()); + active_families_.pop_back(); + if (family.counts->size() == 0) { + for (auto const child_layer_hash : + driver_child_layer_hashes_.at(family.index->layer_hash())) { + family.counts->emplace(child_layer_hash, 0); + } + } + return family; } } diff --git a/phlex/model/data_cell_tracker.hpp b/phlex/model/data_cell_tracker.hpp index 349bbd0b4..c9f5f7c22 100644 --- a/phlex/model/data_cell_tracker.hpp +++ b/phlex/model/data_cell_tracker.hpp @@ -3,14 +3,18 @@ #include "phlex/model/data_cell_index.hpp" #include "phlex/model/flush_messages.hpp" +#include "phlex/model/layer_path.hpp" #include "phlex/phlex_model_export.hpp" +#include #include +#include namespace phlex::detail { class PHLEX_MODEL_EXPORT data_cell_tracker { public: - data_cell_tracker() = default; + data_cell_tracker(std::initializer_list layer_paths); + explicit data_cell_tracker(std::vector const& layer_paths); ~data_cell_tracker(); data_cell_tracker(data_cell_tracker const&) = delete; @@ -24,12 +28,12 @@ namespace phlex::detail { index_flushes report_and_evict_ready_flushes(data_cell_index_ptr const& index); private: - void create_parent_count(data_cell_index_ptr const& parent, data_cell_index_ptr const& child); - void increment_parent_count(data_cell_index_ptr const& parent, - data_cell_index_ptr const& child); + index_flush close_active_family(); - data_cell_index_ptr cached_index_{nullptr}; - std::map pending_flushes_; + std::map> + driver_child_layer_hashes_; + // Each family is identified by its parent index and may span multiple child layers. + std::vector active_families_; }; } diff --git a/test/data_cell_tracker_test.cpp b/test/data_cell_tracker_test.cpp index 74c0794d5..2fc1aeeed 100644 --- a/test/data_cell_tracker_test.cpp +++ b/test/data_cell_tracker_test.cpp @@ -1,20 +1,23 @@ #include "phlex/model/data_cell_index.hpp" #include "phlex/model/data_cell_tracker.hpp" +#include "phlex/model/layer_path.hpp" #include "test/ostream_logger.hpp" #include #include #include +#include #include #include using namespace phlex; using namespace phlex::detail; +using phlex::experimental::layer_path; TEST_CASE("Test data-cell tracker", "[graph]") { - data_cell_tracker tracker; + data_cell_tracker tracker{layer_path{"run/spill/subspill"}}; auto job_index = data_cell_index::job(); auto run4 = job_index->make_child("run", 4); @@ -27,10 +30,13 @@ TEST_CASE("Test data-cell tracker", "[graph]") CHECK(tracker.report_and_evict_ready_flushes(job_index).empty()); CHECK(tracker.report_and_evict_ready_flushes(run4).empty()); CHECK(tracker.report_and_evict_ready_flushes(spill5).empty()); - CHECK(tracker.report_and_evict_ready_flushes(spill6).empty()); + auto flushes = tracker.report_and_evict_ready_flushes(spill6); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == spill5); + CHECK(flushes[0].counts->count(spill5->make_child("subspill", 0)->layer_hash()) == 0); CHECK(tracker.report_and_evict_ready_flushes(subspill2).empty()); - auto flushes = tracker.report_and_evict_ready_flushes(run5); + flushes = tracker.report_and_evict_ready_flushes(run5); REQUIRE(flushes.size() == 2); auto spill6_flush = flushes[0]; @@ -44,17 +50,192 @@ TEST_CASE("Test data-cell tracker", "[graph]") CHECK(run4_flush.counts->count(spill5->layer_hash()) == 2); // spills 5 and 6 flushes = tracker.report_and_evict_ready_flushes(nullptr); - REQUIRE(flushes.size() == 1); // only job should have a flush count + REQUIRE(flushes.size() == 2); // job and the empty second run - auto job_flush = flushes[0]; + CHECK(flushes[0].index == run5); + CHECK(flushes[0].counts->count(run5->make_child("spill", 0)->layer_hash()) == 0); + auto job_flush = flushes[1]; CHECK(job_flush.index == job_index); REQUIRE(job_flush.counts->size() == 1); // Should only be "run" layer CHECK(job_flush.counts->count(run4->layer_hash()) == 2); // runs 4 and 5 } -TEST_CASE("Test data-cell tracker with multiple hierarchy branches", "[graph]") +TEST_CASE("Data-cell tracker reports zero-count driver children", "[graph]") { - data_cell_tracker tracker; + data_cell_tracker tracker{layer_path{"run"}, layer_path{"run/event"}, layer_path{"event"}}; + + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + auto const foreign_event = job->make_child("event", 0); + + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(run).empty()); + + auto flushes = tracker.report_and_evict_ready_flushes(foreign_event); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == run); + REQUIRE(flushes[0].counts->size() == 1); + CHECK(flushes[0].counts->count(run->make_child("event", 0)->layer_hash()) == 0); + + flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == job); + CHECK(flushes[0].counts->count(run->layer_hash()) == 1); + CHECK(flushes[0].counts->count(foreign_event->layer_hash()) == 1); +} + +TEST_CASE("Data-cell tracker closes consecutive empty driver families", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run/spill/event"}, layer_path{"/job/run/spill/event"}}; + // Include duplicate complete/incomplete declarations and implicit intermediate layers. + auto const job = data_cell_index::job(); + auto const run0 = job->make_child("run", 0); + auto const spill = run0->make_child("spill", 0); + auto const run1 = job->make_child("run", 1); + auto const run2 = job->make_child("run", 2); + + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(run0).empty()); + CHECK(tracker.report_and_evict_ready_flushes(spill).empty()); + + auto flushes = tracker.report_and_evict_ready_flushes(run1); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == spill); + REQUIRE(flushes[0].counts->size() == 1); + CHECK(flushes[0].counts->count(spill->make_child("event", 0)->layer_hash()) == 0); + CHECK(flushes[1].index == run0); + CHECK(flushes[1].counts->count(spill->layer_hash()) == 1); + + flushes = tracker.report_and_evict_ready_flushes(run2); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == run1); + CHECK(flushes[0].counts->count(spill->layer_hash()) == 0); + + flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 2); + for (auto const& flush : flushes) { + if (flush.index == job) { + CHECK(flush.counts->count(run0->layer_hash()) == 3); + } else { + CHECK(flush.index == run2); + REQUIRE(flush.counts->size() == 1); + CHECK(flush.counts->count(spill->layer_hash()) == 0); + } + } + CHECK(tracker.report_and_evict_ready_flushes(nullptr).empty()); +} + +TEST_CASE("Data-cell tracker closes an empty job only for driver-owned children", "[graph]") +{ + auto const job = data_cell_index::job(); + SECTION("Declared driver children") + { + data_cell_tracker tracker{layer_path{"run"}}; + CHECK(tracker.report_and_evict_ready_flushes(nullptr).empty()); + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + auto const flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == job); + REQUIRE(flushes[0].counts->size() == 1); + CHECK(flushes[0].counts->count(job->make_child("run", 0)->layer_hash()) == 0); + } + SECTION("Children owned only by unfolds") + { + data_cell_tracker tracker{layer_path{"/job"}}; + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(nullptr).empty()); + } +} + +TEST_CASE("Data-cell tracker keeps populated completion counts sparse", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run"}, layer_path{"event"}}; + auto const job = data_cell_index::job(); + auto const event = job->make_child("event", 0); + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(event).empty()); + + auto const flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == job); + REQUIRE(flushes[0].counts->size() == 1); + CHECK(flushes[0].counts->count(event->layer_hash()) == 1); + CHECK(flushes[0].counts->count(job->make_child("run", 0)->layer_hash()) == 0); +} + +TEST_CASE("Data-cell tracker tracks leaf siblings without retaining leaf indices", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run/event"}}; + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + auto const event0 = run->make_child("event", 0); + auto const event1 = run->make_child("event", 1); + + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(run).empty()); + CHECK(tracker.report_and_evict_ready_flushes(event0).empty()); + CHECK(tracker.report_and_evict_ready_flushes(event1).empty()); + + auto const flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == run); + CHECK(flushes[0].counts->count(event0->layer_hash()) == 2); + CHECK(flushes[1].index == job); + CHECK(flushes[1].counts->count(run->layer_hash()) == 1); +} + +TEST_CASE("Data-cell tracker distinguishes repeated trailing layer names", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run/event"}, layer_path{"calib/event"}}; + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + auto const run_event = run->make_child("event", 0); + auto const calib = job->make_child("calib", 0); + auto const calib_event = calib->make_child("event", 0); + + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(run).empty()); + CHECK(tracker.report_and_evict_ready_flushes(run_event).empty()); + auto flushes = tracker.report_and_evict_ready_flushes(calib); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == run); + CHECK(flushes[0].counts->count(run_event->layer_hash()) == 1); + CHECK(tracker.report_and_evict_ready_flushes(calib_event).empty()); + + flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == calib); + CHECK(flushes[0].counts->count(calib_event->layer_hash()) == 1); + CHECK(flushes[1].index == job); + CHECK(flushes[1].counts->count(run->layer_hash()) == 1); + CHECK(flushes[1].counts->count(calib->layer_hash()) == 1); +} + +TEST_CASE("Data-cell tracker closes multiple active levels deepest first", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run/spill/event"}, layer_path{"run/calib/event"}}; + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + auto const spill = run->make_child("spill", 0); + auto const event = spill->make_child("event", 0); + auto const next_run = job->make_child("run", 1); + + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK(tracker.report_and_evict_ready_flushes(run).empty()); + CHECK(tracker.report_and_evict_ready_flushes(spill).empty()); + CHECK(tracker.report_and_evict_ready_flushes(event).empty()); + auto const flushes = tracker.report_and_evict_ready_flushes(next_run); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == spill); + CHECK(flushes[0].counts->count(event->layer_hash()) == 1); + CHECK(flushes[1].index == run); + CHECK(flushes[1].counts->count(spill->layer_hash()) == 1); + CHECK(flushes[1].counts->count(run->make_child("calib", 0)->layer_hash()) == 0); +} + +TEST_CASE("Data-cell tracker counts a family spanning multiple child layers", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run/event"}, layer_path{"calib/event"}}; auto job_index = data_cell_index::job(); auto run4 = job_index->make_child("run", 4); @@ -63,21 +244,55 @@ TEST_CASE("Test data-cell tracker with multiple hierarchy branches", "[graph]") CHECK(tracker.report_and_evict_ready_flushes(job_index).empty()); CHECK(tracker.report_and_evict_ready_flushes(run4).empty()); - CHECK(tracker.report_and_evict_ready_flushes(calib1).empty()); - CHECK(tracker.report_and_evict_ready_flushes(run5).empty()); + auto flushes = tracker.report_and_evict_ready_flushes(calib1); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == run4); + CHECK(flushes[0].counts->count(run4->make_child("event", 0)->layer_hash()) == 0); + flushes = tracker.report_and_evict_ready_flushes(run5); + REQUIRE(flushes.size() == 1); + CHECK(flushes[0].index == calib1); + CHECK(flushes[0].counts->count(calib1->make_child("event", 0)->layer_hash()) == 0); - auto flushes = tracker.report_and_evict_ready_flushes(nullptr); - REQUIRE(flushes.size() == 1); // only job should have a flush count - auto job_flush = flushes[0]; + flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == run5); + CHECK(flushes[0].counts->count(run5->make_child("event", 0)->layer_hash()) == 0); + auto job_flush = flushes[1]; CHECK(job_flush.index == job_index); REQUIRE(job_flush.counts->size() == 2); // Should have "run" and "calib" layers CHECK(job_flush.counts->count(run4->layer_hash()) == 2); // run 4 and 5 CHECK(job_flush.counts->count(calib1->layer_hash()) == 1); // calib 1 } +TEST_CASE("Data-cell tracker rejects root indices while driver families are active", "[graph]") +{ + data_cell_tracker tracker{layer_path{"run/event"}}; + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + auto const event = run->make_child("event", 0); + + CHECK(tracker.report_and_evict_ready_flushes(job).empty()); + CHECK_THROWS_WITH(tracker.report_and_evict_ready_flushes(job), + "Received root index [] while driver families are active"); + CHECK(tracker.report_and_evict_ready_flushes(run).empty()); + CHECK(tracker.report_and_evict_ready_flushes(event).empty()); + CHECK_THROWS_WITH(tracker.report_and_evict_ready_flushes(job), + "Received root index [] while driver families are active"); + + auto const flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == run); + REQUIRE(flushes[0].counts->size() == 1); + CHECK(flushes[0].counts->count(event->layer_hash()) == 1); + CHECK(flushes[1].index == job); + REQUIRE(flushes[1].counts->size() == 1); + CHECK(flushes[1].counts->count(run->layer_hash()) == 1); + CHECK(tracker.report_and_evict_ready_flushes(nullptr).empty()); +} + TEST_CASE("Test data-cell tracker with missing intermediate layers", "[graph]") { - data_cell_tracker tracker; + data_cell_tracker tracker{layer_path{"run/spill"}}; auto job_index = data_cell_index::job(); auto run4 = job_index->make_child("run", 4); @@ -86,14 +301,21 @@ TEST_CASE("Test data-cell tracker with missing intermediate layers", "[graph]") CHECK(tracker.report_and_evict_ready_flushes(job_index).empty()); CHECK_THROWS_WITH(tracker.report_and_evict_ready_flushes(spill2), - "Received index [run:4, spill:2], which is not an immediate child of []"); + "Received index [run:4, spill:2], whose parent [run:4] does not identify an " + "active driver family"); + CHECK(tracker.report_and_evict_ready_flushes(run4).empty()); + auto const flushes = tracker.report_and_evict_ready_flushes(nullptr); + REQUIRE(flushes.size() == 2); + CHECK(flushes[0].index == run4); + CHECK(flushes[1].index == job_index); } -TEST_CASE("Cached flush counts at destruction generate warning message", "[graph]") +TEST_CASE("Active families at destruction generate warning message", "[graph]") { std::ostringstream oss; auto logger = phlex::test::use_ostream_logger(oss); - auto tracker = std::make_unique(); + auto tracker = + std::make_unique(std::initializer_list{layer_path{"run/event"}}); auto job_index = data_cell_index::job(); auto run4 = job_index->make_child("run", 4); @@ -103,7 +325,7 @@ TEST_CASE("Cached flush counts at destruction generate warning message", "[graph tracker.reset(); // Invoke destructor to trigger warning message auto const warning = oss.str(); - CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Cached pending flushes at destruction:")); + CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Active families at destruction:")); CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Index: []")); - CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("7457871974376244100 = 1")); + CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Index: [run:4]")); } 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 index 7971a5626..12b00af4b 100644 --- a/test/index_router_test.cpp +++ b/test/index_router_test.cpp @@ -1,15 +1,20 @@ #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 +#include #include +#include using namespace phlex; @@ -173,3 +178,93 @@ TEST_CASE("Unfold hierarchy rejects duplicate output layers", "[graph][unfold]") {}), "Unfolds 'split_a' and 'split_b' specify the same output layer '/job/run/spill'."); } + +TEST_CASE("Counting layers on other branches do not produce end tokens", "[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}; + router.finalize( + graph, + {experimental::layer_path{"/job/run/calib"}, experimental::layer_path{"/job/spill/event"}}, + {}, + {}, + {}, + {{"fold", + {{.layer = "run"_id, + .counting_layer = "event"_id, + .token_port = &tokens, + .index_port = &indices}}}}); + + auto const job = data_cell_index::job(); + auto const run = job->make_child("run", 0); + router.route(job, {}); + router.route(run, {}); + graph.wait_for_all(); + detail::index_message message; + REQUIRE(indices.try_get(message)); + CHECK(message.index == run); + CHECK_FALSE(indices.try_get(message)); + detail::indexed_end_token token; + CHECK_FALSE(tokens.try_get(token)); + + auto run_counts = std::make_shared(); + run_counts->emplace(run->make_child("calib", 0)->layer_hash(), 0); + auto job_counts = std::make_shared(); + job_counts->emplace(run->layer_hash(), 1); + job_counts->emplace(job->make_child("spill", 0)->layer_hash(), 0); + router.drain({{.index = run, .counts = run_counts}, {.index = job, .counts = job_counts}}); + graph.wait_for_all(); + CHECK_FALSE(tokens.try_get(token)); +} + +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());