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 69f9125c0..64920ad3e 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; } @@ -114,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(); @@ -160,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(); } @@ -208,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(); } @@ -250,18 +245,13 @@ namespace phlex::detail { void framework_graph::finalize() { throw_if_registration_errors(); + cell_tracker_.emplace(fixed_hierarchy_.layer_paths()); make_filter_edges(); make_bookkeeping_edges(); 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/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 d187dcc7f..623b29801 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -8,8 +8,10 @@ #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 #include #include @@ -18,11 +20,10 @@ #include #include #include +#include #include -#include #include #include -#include #include #include @@ -30,6 +31,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 +151,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 +252,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 +278,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) { @@ -292,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}); } }); @@ -304,19 +350,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 +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: 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 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 @@ -425,10 +465,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); + 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); @@ -456,7 +496,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 +506,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 +583,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..0cb2d6ede 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 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; @@ -52,12 +50,9 @@ 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. + // 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; @@ -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/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/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/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/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/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..12b00af4b --- /dev/null +++ b/test/index_router_test.cpp @@ -0,0 +1,270 @@ +#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 + +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'."); +} + +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/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(); 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());