diff --git a/.amazonq/rules/memory-bank/structure.md b/.amazonq/rules/memory-bank/structure.md index 84ed041fa..ef849e1b0 100644 --- a/.amazonq/rules/memory-bank/structure.md +++ b/.amazonq/rules/memory-bank/structure.md @@ -16,7 +16,7 @@ The main framework implementation containing the execution engine and core abstr - Algorithm declarations: `provider_node.cpp/hpp`, `declared_transform.cpp/hpp`, `declared_observer.cpp/hpp`, `declared_fold.cpp/hpp`, `declared_unfold.cpp/hpp`, `declared_predicate.cpp/hpp`, `declared_output.cpp/hpp` - Graph execution: `framework_graph.cpp/hpp`, `make_computational_edges.cpp/hpp`, `producer_catalog.cpp/hpp` - Data flow: `message.cpp/hpp`, `message_sender.cpp/hpp`, `consumer.cpp/hpp` - - Product management: `product_query.cpp/hpp`, `products_consumer.cpp/hpp` + - Product management: `product_selector.cpp/hpp`, `products_consumer.cpp/hpp` - Registration: `registrar.cpp/hpp`, `registration_api.cpp/hpp`, `node_catalog.cpp/hpp` - Utilities: `filter.cpp/hpp`, `glue.cpp/hpp`, `multiplexer.cpp/hpp` - `fold/`: Fold operation implementations diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt index 1e8f0f457..b35c04c0f 100644 --- a/phlex/core/CMakeLists.txt +++ b/phlex/core/CMakeLists.txt @@ -28,7 +28,6 @@ cet_make_library( provider_node.cpp registrar.cpp registration_api.cpp - store_counters.cpp LIBRARIES PUBLIC TBB::tbb @@ -67,7 +66,6 @@ install( registrar.hpp registration_api.hpp source.hpp - store_counters.hpp upstream_predicates.hpp DESTINATION include/phlex/core ) diff --git a/phlex/core/concepts.hpp b/phlex/core/concepts.hpp index d6ef4ffa6..3ccf7d45e 100644 --- a/phlex/core/concepts.hpp +++ b/phlex/core/concepts.hpp @@ -1,6 +1,7 @@ #ifndef PHLEX_CORE_CONCEPTS_HPP #define PHLEX_CORE_CONCEPTS_HPP +#include "phlex/core/fold/send.hpp" #include "phlex/core/fwd.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/fwd.hpp" @@ -17,9 +18,7 @@ namespace phlex::experimental { concept is_bound_object = not std::same_as; template - concept sendable = std::move_constructible || requires(T& t) { - { send(t) } -> std::move_constructible; - }; + concept sendable = requires { typename sendable_type; }; template concept at_least_n_input_parameters = number_parameters >= N; diff --git a/phlex/core/declared_fold.cpp b/phlex/core/declared_fold.cpp index 8eda7c8f2..bbd943516 100644 --- a/phlex/core/declared_fold.cpp +++ b/phlex/core/declared_fold.cpp @@ -3,8 +3,10 @@ namespace phlex::experimental { declared_fold::declared_fold(algorithm_name name, std::vector predicates, - product_selectors input_products) : - products_consumer{std::move(name), std::move(predicates), std::move(input_products)} + product_selectors input_products, + std::string partition_layer) : + products_consumer{std::move(name), std::move(predicates), std::move(input_products)}, + partition_layer_{std::move(partition_layer)} { } diff --git a/phlex/core/declared_fold.hpp b/phlex/core/declared_fold.hpp index a665e3d38..a19a6f64e 100644 --- a/phlex/core/declared_fold.hpp +++ b/phlex/core/declared_fold.hpp @@ -6,13 +6,12 @@ #include "phlex/concurrency.hpp" #include "phlex/core/concepts.hpp" #include "phlex/core/fold/send.hpp" +#include "phlex/core/fold_join_node.hpp" #include "phlex/core/fwd.hpp" #include "phlex/core/input_arguments.hpp" #include "phlex/core/message.hpp" -#include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/product_selector.hpp" #include "phlex/core/products_consumer.hpp" -#include "phlex/core/store_counters.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/data_cell_index.hpp" #include "phlex/model/handle.hpp" @@ -40,26 +39,44 @@ namespace phlex::experimental { public: declared_fold(algorithm_name name, std::vector predicates, - product_selectors input_products); + product_selectors input_products, + std::string partition_layer); ~declared_fold() override; virtual tbb::flow::sender& output_port() = 0; - virtual tbb::flow::receiver& flush_port() = 0; virtual product_specifications const& output() const = 0; + virtual tbb::flow::receiver& partition_port() = 0; virtual std::size_t product_count() const = 0; + identifier const& partition_layer() const { return partition_layer_; } + + private: + identifier partition_layer_; }; using declared_fold_ptr = std::unique_ptr; using declared_folds = simple_ptr_map; + template + auto make_initializer(InitTuple args, std::index_sequence) + { + // The compiler emits a warning if an empty args tuple is captured by reference and then not expanded inside the lambda expression. We therefore only capture args in the lambda expression if sizeof...(Is) > 0. + if constexpr (sizeof...(Is) == 0) { + return [](data_cell_index const&) { return std::make_unique(); }; + } else { + return [args](data_cell_index const&) { + return std::make_unique(std::get(args)...); + }; + } + } + // ===================================================================================== template - class fold_node : public declared_fold, private count_stores { + class fold_node : public declared_fold { using all_parameter_types = typename AlgorithmBits::input_parameter_types; + using result_type = std::decay_t>; using input_parameter_types = skip_first_type; // Skip fold object static constexpr auto num_inputs = std::tuple_size_v; - using result_type = std::decay_t>; static constexpr std::size_t num_outputs = 1; // hard-coded for now using function_t = typename AlgorithmBits::bound_type; @@ -73,138 +90,69 @@ namespace phlex::experimental { InitTuple initializer, product_selectors input_products, std::vector output, - std::string partition) : - declared_fold{std::move(algo_name), std::move(predicates), std::move(input_products)}, - initializer_{std::move(initializer)}, + std::string partition_layer) : + declared_fold{std::move(algo_name), + std::move(predicates), + std::move(input_products), + std::move(partition_layer)}, output_{to_product_specifications(name(), std::move(output), make_type_ids())}, - partition_{std::move(partition)}, - flush_receiver_{g, - tbb::flow::unlimited, - [this](flush_message const& msg) -> tbb::flow::continue_msg { - auto const& [index, counts, original_message_id] = msg; - if (index->layer_name() != partition_) { - return {}; - } - - counter_for(index->hash()).set_flush_value(counts, original_message_id); - emit_and_evict_if_done(index); - return {}; - }}, - join_{make_join_or_none( - g, name().to_string(), layers())}, // FIXME: This should change to include result product! + join_{g, + name().to_string(), + this->partition_layer(), + layers(), + this->output(), + make_initializer( + std::move(initializer), std::make_index_sequence>{})}, fold_{g, concurrency, - [this, ft = alg.release_algorithm()](messages_t const& messages, auto&) { - // N.B. The assumption is that a fold will *never* need to cache - // the product store it creates. Any flush messages *do not* need - // to be propagated to downstream nodes. - auto const& msg = most_derived(messages); - auto const& index = msg.store->index(); - - auto fold_index = index->parent(partition_); - if (not fold_index) { - return; - } - - auto index_hash_for_counter = fold_index->hash(); + [this, ft = alg.release_algorithm()]( + accumulator_with_messages const& accum_with_msgs, auto&) { + std::size_t const partition_hash = apply_fold(ft, accum_with_msgs); - call(ft, messages, std::make_index_sequence{}); ++calls_; - counter_for(index_hash_for_counter).increment(index->layer_hash()); - - emit_and_evict_if_done(fold_index); + join_.notify_result_repeater_port().try_put(partition_hash); }} { - if constexpr (num_inputs > 1ull) { - make_edge(join_, fold_); - } + make_edge(join_, fold_); } private: - void emit_and_evict_if_done(data_cell_index_ptr const& fold_index) - { - if (auto counter = done_with(fold_index->hash())) { - auto parent = std::make_shared(fold_index, name()); - commit(parent); - ++product_count_; - tbb::flow::output_port<0>(fold_).try_put( - {.store = parent, .id = counter->original_message_id()}); - } - } - tbb::flow::receiver& port_for(product_selector const& input_product) override { - return receiver_for(join_, input(), input_product, fold_); + return receiver_for(join_, input(), input_product); } - std::vector*> ports() override - { - return input_ports(join_, fold_); - } + std::vector*> ports() override { return input_ports(join_); } - tbb::flow::receiver& flush_port() override { return flush_receiver_; } - tbb::flow::sender& output_port() override { return tbb::flow::output_port<0>(fold_); } + tbb::flow::receiver& partition_port() override { return join_.partition_port(); } + tbb::flow::sender& output_port() override { return join_.output_port(); } product_specifications const& output() const override { return output_; } - template - void call(function_t const& ft, - messages_t const& messages, - std::index_sequence) - { - auto const parent_index = most_derived(messages).store->index()->parent(partition_); - - // FIXME: Not the safest approach! - auto it = results_.find(parent_index->hash()); - if (it == results_.end()) { - it = results_ - .insert({parent_index->hash(), - initialized_object( - initializer_, std::make_index_sequence>{})}) - .first; - } - - if constexpr (num_inputs == 1ull) { - std::invoke(ft, *it->second, std::get(input_).retrieve(messages)...); - } else { - std::invoke(ft, *it->second, std::get(input_).retrieve(std::get(messages))...); - } - } - named_index_ports index_ports() final { return join_.index_ports(); } std::size_t num_calls() const final { return calls_.load(); } - std::size_t product_count() const final { return product_count_.load(); } - - template - auto initialized_object(InitTuple const& init, std::index_sequence) const - { - return std::unique_ptr(new result_type{std::get(init)...}); - } + std::size_t product_count() const final { return join_.emitted_result_count(); } - auto commit(product_store_ptr& store) + std::size_t apply_fold( + function_t const& ft, + accumulator_with_messages const& accum_with_msgs) { - auto& result = results_.at(store->index()->hash()); - if constexpr (requires { send(*result); }) { - store->add_product(output()[0], send(*result)); - } else { - store->add_product(output()[0], std::move(*result)); - } - // Reclaim some memory; it would be better to erase the entire entry from the map, - // but that is not thread-safe. - result.reset(); + // We have to do awkward index management until we can use structured bindings with packs. + auto& accumulator = std::get<0>(accum_with_msgs); + [&](std::index_sequence) { + accumulator.partial_result->call( + ft, std::get(input_).retrieve(std::get(accum_with_msgs))...); + }(std::make_index_sequence{}); + return accumulator.index->hash(); } - InitTuple initializer_; input_retriever_types input_{input_arguments()}; product_specifications output_; - identifier partition_; - tbb::flow::function_node flush_receiver_; - join_or_none_t join_; - tbb::flow::multifunction_node, message_tuple<1>> fold_; - tbb::concurrent_unordered_map> - results_; + fold_join_node join_; + tbb::flow::multifunction_node, + message_tuple<1>> + fold_; std::atomic calls_; - std::atomic product_count_; }; } diff --git a/phlex/core/declared_observer.hpp b/phlex/core/declared_observer.hpp index befb2751c..203e3cf3d 100644 --- a/phlex/core/declared_observer.hpp +++ b/phlex/core/declared_observer.hpp @@ -10,7 +10,6 @@ #include "phlex/core/multilayer_join_node.hpp" #include "phlex/core/product_selector.hpp" #include "phlex/core/products_consumer.hpp" -#include "phlex/core/store_counters.hpp" #include "phlex/metaprogramming/type_deduction.hpp" #include "phlex/model/algorithm_name.hpp" #include "phlex/model/data_cell_index.hpp" diff --git a/phlex/core/declared_unfold.cpp b/phlex/core/declared_unfold.cpp index 159521c67..786bc00f5 100644 --- a/phlex/core/declared_unfold.cpp +++ b/phlex/core/declared_unfold.cpp @@ -20,7 +20,7 @@ namespace phlex::experimental { product_store_const_ptr generator::make_child(std::size_t const i, products new_products) { auto child_index = parent_->index()->make_child(child_layer_name_, i); - ++child_counts_; + ++child_count_; return std::make_shared(child_index, node_name_, std::move(new_products)); } diff --git a/phlex/core/declared_unfold.hpp b/phlex/core/declared_unfold.hpp index dbe7e5cab..bfb68561d 100644 --- a/phlex/core/declared_unfold.hpp +++ b/phlex/core/declared_unfold.hpp @@ -42,7 +42,7 @@ namespace phlex::experimental { std::string const& child_layer_name); std::size_t child_layer_hash() const { return child_layer_hash_; } - std::size_t child_count() const { return child_counts_; } + std::size_t child_count() const { return child_count_; } product_store_const_ptr make_child(std::size_t i, products new_products); private: @@ -52,7 +52,7 @@ namespace phlex::experimental { // NOLINTNEXTLINE(cppcoreguidelines-avoid-const-or-ref-data-members) std::string const& child_layer_name_; std::size_t child_layer_hash_; - std::size_t child_counts_ = 0; + std::size_t child_count_ = 0; }; class PHLEX_CORE_EXPORT declared_unfold : public products_consumer { diff --git a/phlex/core/detail/accumulator_node.hpp b/phlex/core/detail/accumulator_node.hpp new file mode 100644 index 000000000..b47d3cbae --- /dev/null +++ b/phlex/core/detail/accumulator_node.hpp @@ -0,0 +1,337 @@ +#ifndef PHLEX_CORE_DETAIL_ACCUMULATOR_NODE_HPP +#define PHLEX_CORE_DETAIL_ACCUMULATOR_NODE_HPP + +#include "phlex/core/fold/send.hpp" +#include "phlex/core/message.hpp" +#include "phlex/phlex_core_export.hpp" + +#include "oneapi/tbb/concurrent_hash_map.h" +#include "oneapi/tbb/concurrent_queue.h" +#include "oneapi/tbb/flow_graph.h" +#include "spdlog/spdlog.h" + +#include +#include +#include +#include +#include + +namespace phlex::experimental::detail { + + template + class accumulator { + public: + using sendable_t = sendable_type; + + explicit accumulator(std::unique_ptr initial_value) : accumulator_(std::move(initial_value)) + { + } + + template + void call(FT const& f, Args const&... args) + { + std::invoke(f, *accumulator_, args...); + } + + auto release_as_product() + { + auto result = std::move(accumulator_); + if constexpr (requires { send(*result); }) { + return std::make_unique>(send(*result)); + } else { + return std::make_unique>(std::move(*result)); + } + } + + private: + std::unique_ptr accumulator_; + }; + + template + struct accumulator_message { + phlex::data_cell_index_ptr index; + std::shared_ptr> partial_result; + std::size_t id{}; + + accumulator_message propagate_with_id(std::size_t new_id) const + { + return {.index = index, .partial_result = partial_result, .id = new_id}; + } + + message release_as_message(std::string const& node_name, + product_specifications const& output, + std::size_t original_id) + { + auto store = std::make_shared(index, node_name); + // FIXME: Only read the first output specification, which is a temporary limitation until + // we support multiple outputs from folds. + store->add_product(output.front(), partial_result->release_as_product()); + partial_result.reset(); + return {.store = std::move(store), .id = original_id}; + } + }; + + template + struct accumulator_message_matcher { + std::size_t operator()(accumulator_message const& msg) const { return msg.id; } + }; + + using accumulator_node_input = + std::tuple; + + template + class PHLEX_CORE_EXPORT accumulator_node : + public tbb::flow::composite_node>> { + + using result_initializer_t = std::function(data_cell_index const&)>; + + public: + accumulator_node(tbb::flow::graph& g, + std::string node_name, + identifier partition_layer_name, + product_specifications output, + result_initializer_t initializer); + + accumulator_node(accumulator_node const&) = delete; + accumulator_node(accumulator_node&&) = delete; + accumulator_node& operator=(accumulator_node const&) = delete; + accumulator_node& operator=(accumulator_node&&) = delete; + + tbb::flow::receiver& partition_port(); + tbb::flow::receiver& flush_port(); + tbb::flow::receiver& index_port(); + + bool cache_is_empty() const; + std::size_t cache_size() const; + std::size_t emitted_result_count() const; + + ~accumulator_node() override; + + private: + using accumulator_msg_t = accumulator_message; + using base_t = + tbb::flow::composite_node>; + using tagged_msg_t = tbb::flow:: + tagged_msg; + using multifunction_node_t = + tbb::flow::multifunction_node>; + + struct cached_accumulator { + std::shared_ptr accumulator_msg; + tbb::concurrent_queue msg_ids{}; + std::atomic counter; + std::atomic_flag flush_received{}; + std::size_t original_message_id{}; + }; + + using cache_t = + tbb::concurrent_hash_map; // Key is the index hash + using accessor = cache_t::accessor; + + void emit_pending_ids(cached_accumulator* entry); + void handle_partition_message(index_message const& msg); + void handle_flush_token(indexed_end_token const& token); + void handle_index_message(index_message const& msg); + void cleanup_cache_entry(accessor& a); + void increment_cache_entry_then_cleanup(std::size_t key); + + tbb::flow::indexer_node indexer_; + multifunction_node_t repeater_; + cache_t cached_results_; + std::string node_name_; + identifier partition_layer_; + result_initializer_t initializer_; + product_specifications output_; + std::atomic emitted_result_count_{0}; + }; + + // ============================================================================== + // Implementation + + template + accumulator_node::accumulator_node(tbb::flow::graph& g, + std::string node_name, + identifier partition_layer_name, + product_specifications output, + result_initializer_t initializer) : + base_t{g}, + indexer_{g}, + repeater_{g, + tbb::flow::unlimited, + [this](tagged_msg_t const& tagged, auto& /* outputs */) { + if (tagged.tag() == 0) { + handle_partition_message(tagged.cast_to()); + } else if (tagged.tag() == 1) { + handle_flush_token(tagged.cast_to()); + } else if (tagged.tag() == 2) { + handle_index_message(tagged.cast_to()); + } else { + assert(tagged.tag() == 3); + // This means that a fold operation has taken place, and an attempt should + // be made to emit the fold result and clean up the cache entry. + increment_cache_entry_then_cleanup(tagged.cast_to()); + } + }}, + node_name_{std::move(node_name)}, + partition_layer_{std::move(partition_layer_name)}, + initializer_{std::move(initializer)}, + output_{std::move(output)} + { + base_t::set_external_ports( + typename base_t::input_ports_type{input_port<0>(indexer_), + input_port<1>(indexer_), + input_port<2>(indexer_), + input_port<3>(indexer_)}, + typename base_t::output_ports_type{output_port<0>(repeater_), output_port<1>(repeater_)}); + make_edge(indexer_, repeater_); + } + + template + tbb::flow::receiver& accumulator_node::partition_port() + { + return input_port<0>(indexer_); + } + + template + tbb::flow::receiver& accumulator_node::flush_port() + { + return input_port<1>(indexer_); + } + + template + tbb::flow::receiver& accumulator_node::index_port() + { + return input_port<2>(indexer_); + } + + template + bool accumulator_node::cache_is_empty() const + { + return cached_results_.empty(); + } + + template + std::size_t accumulator_node::cache_size() const + { + return cached_results_.size(); + } + + template + std::size_t accumulator_node::emitted_result_count() const + { + return emitted_result_count_.load(); + } + + template + accumulator_node::~accumulator_node() + { + if (cached_results_.empty()) { + return; + } + + spdlog::warn( + "[{}/{}] Cached accumulators: {}", node_name_, partition_layer_, cached_results_.size()); + for (auto const& [_, cache] : cached_results_) { + if (cache.accumulator_msg) { + spdlog::warn("[{}/{}] Partition {}", + node_name_, + partition_layer_, + cache.accumulator_msg->index->to_string()); + } else { + spdlog::warn("[{}/{}] Partition index not yet received", node_name_, partition_layer_); + } + } + } + + template + void accumulator_node::emit_pending_ids(cached_accumulator* entry) + { + assert(entry->accumulator_msg); + std::size_t msg_id{}; + while (entry->msg_ids.try_pop(msg_id)) { + auto& accum_msg = entry->accumulator_msg; + output_port<1>(repeater_).try_put(accum_msg->propagate_with_id(msg_id)); + } + } + + template + void accumulator_node::handle_partition_message(index_message const& msg) + { + auto const key = msg.index->hash(); + + accessor a; + cached_results_.insert(a, key); + auto* entry = &a->second; + entry->accumulator_msg.reset(new accumulator_msg_t{ + .index = msg.index, + .partial_result = std::make_shared>(initializer_(*msg.index))}); + entry->original_message_id = msg.msg_id; + emit_pending_ids(entry); + // Handle the flush-before-partition case: if the flush token already arrived (and + // left counter == 0 because no fold inputs flowed under this partition), emit the + // initial accumulator value now that we finally have the index and original message + // ID needed to construct the output. + cleanup_cache_entry(a); + } + + template + void accumulator_node::handle_flush_token(indexed_end_token const& token) + { + auto const& [index, count] = token; + accessor a; + cached_results_.insert(a, index->hash()); + auto* entry = &a->second; + entry->counter -= count; + std::ignore = entry->flush_received.test_and_set(); + cleanup_cache_entry(a); + } + + template + void accumulator_node::handle_index_message(index_message const& msg) + { + auto const& [index, msg_id, cache] = msg; + assert(cache); + auto const key = index->hash(); + + accessor a; + cached_results_.insert(a, key); + auto* entry = &a->second; + if (auto& accum_msg = entry->accumulator_msg) { + output_port<1>(repeater_).try_put(accum_msg->propagate_with_id(msg_id)); + emit_pending_ids(entry); + } else { + entry->msg_ids.push(msg_id); + } + } + + template + void accumulator_node::cleanup_cache_entry(accessor& a) + { + auto* entry = &a->second; + // The `accumulator_msg` check guards the flush-before-partition case: a zero-count + // flush can establish the cache entry with `flush_received == true` and + // `counter == 0` before the partition message has supplied the index and initial + // value. In that case we defer emission until `handle_partition_message` re-enters + // cleanup with a populated `accumulator_msg`. + if (entry->flush_received.test() and entry->counter == 0 and entry->accumulator_msg) { + output_port<0>(repeater_).try_put(entry->accumulator_msg->release_as_message( + node_name_, output_, entry->original_message_id)); + ++emitted_result_count_; + cached_results_.erase(a); + } + } + + template + void accumulator_node::increment_cache_entry_then_cleanup(std::size_t key) + { + accessor a; + if (!cached_results_.find(a, key)) { + return; + } + ++a->second.counter; + cleanup_cache_entry(a); + } +} + +#endif // PHLEX_CORE_DETAIL_ACCUMULATOR_NODE_HPP diff --git a/phlex/core/detail/repeater_node.cpp b/phlex/core/detail/repeater_node.cpp index db7c811d2..d3ec1ffb5 100644 --- a/phlex/core/detail/repeater_node.cpp +++ b/phlex/core/detail/repeater_node.cpp @@ -159,8 +159,7 @@ namespace phlex::experimental::detail { auto* entry = &a->second; if (!cache_enabled_) { - if (entry->counter == 0) { - assert(entry->data_msg); + if (entry->counter == 0 and entry->data_msg) { output_port<0>(repeater_).try_put(*entry->data_msg); } cached_products_.erase(a); diff --git a/phlex/core/fold/send.hpp b/phlex/core/fold/send.hpp index 01a0a657d..b936e9ba0 100644 --- a/phlex/core/fold/send.hpp +++ b/phlex/core/fold/send.hpp @@ -9,6 +9,9 @@ // // For non-moveable types, an overload of 'T send(U const&)' is required, where U is the // unmoveable type, and T is the type that can be a represented as a data product. +// +// If a type both has a 'send(...)' overload and is move-constructible, the 'send(...)' +// overload takes precedence. // ======================================================================================= #include @@ -20,6 +23,32 @@ namespace phlex::experimental { { return a.load(); } + + template + concept has_send = requires(T const& t) { + { send(t) } -> std::move_constructible; + }; + + template + concept move_constructible_only = std::move_constructible && !has_send; + + namespace detail { + template + struct sendable_type_impl {}; + + template + struct sendable_type_impl { + using type = decltype(send(std::declval())); + }; + + template + struct sendable_type_impl { + using type = T; + }; + } + + template + using sendable_type = typename detail::sendable_type_impl::type; } #endif // PHLEX_CORE_FOLD_SEND_HPP diff --git a/phlex/core/fold_join_node.hpp b/phlex/core/fold_join_node.hpp new file mode 100644 index 000000000..007586425 --- /dev/null +++ b/phlex/core/fold_join_node.hpp @@ -0,0 +1,234 @@ +#ifndef PHLEX_CORE_FOLD_JOIN_NODE_HPP +#define PHLEX_CORE_FOLD_JOIN_NODE_HPP + +#include "phlex/core/detail/accumulator_node.hpp" +#include "phlex/core/detail/repeater_node.hpp" +#include "phlex/core/message.hpp" + +#include "oneapi/tbb/flow_graph.h" + +#include +#include +#include +#include +#include + +namespace phlex::experimental { + + // A fold_join_node is a TBB composite flow-graph node that synchronises a + // partition-level fold accumulator with N data-input streams. For each data + // unit it waits until the accumulator_node has emitted the current partial + // result *and* all N repeater_nodes have forwarded their data messages, then + // emits the complete tuple (accumulator, data[0..N-1]) to the fold executor. + // + // The accumulator_node manages fold results at the partition layer: it receives + // a partition-level index and, for every expected data cell, emits an + // accumulator_message carrying the shared partial result. + // + // A repeater_node is always present for each data input because the + // partition-layer accumulator necessarily lives at a higher hierarchy level than + // the data products being folded. Repeaters are therefore never collapsed away + // as they are in multilayer_join_node. + // + // Schematic: + // + // partition ──► accumulator_node ──► port<0>(join) ──┐ + // │ + // data[0] ──► repeater[0] ──────► port<1>(join) ──┤ + // ⋮ ├──► join_node ──► (accum, data[0..N-1]) + // data[N-1] ──► repeater[N-1] ────► port(join) ───┘ + + template + using partition_index_with_messages = + decltype(std::tuple_cat(std::tuple{}, message_tuple{})); + + template + using accumulator_with_messages = decltype(std::tuple_cat( + std::tuple>{}, message_tuple{})); + + template + using fold_join_node_base_t = + tbb::flow::composite_node, + std::tuple>>; + + template + class fold_join_node : public fold_join_node_base_t { + using args_t = partition_index_with_messages; + using base_t = fold_join_node_base_t; + using join_args_t = accumulator_with_messages; + using result_initializer_t = std::function(data_cell_index const&)>; + using input_t = typename base_t::input_ports_type; + using output_t = typename base_t::output_ports_type; + + template + static auto make_join(tbb::flow::graph& g, std::index_sequence) + { + // tag_matching causes TBB to group messages by the value returned by + // message_matcher, which extracts the message ID. Messages with the same ID across + // all ports are forwarded together as one tuple, ensuring that each output contains + // exactly the messages that belong to the same data unit. + return tbb::flow::join_node{ + g, + [](detail::accumulator_message const& msg) { return msg.id; }, + type_t{}...}; + } + + public: + fold_join_node(tbb::flow::graph& g, + std::string const& node_name, + identifier partition_layer_name, + std::vector layer_names, + product_specifications output, + result_initializer_t result_initializer) : + base_t{g}, + result_repeater_{ + g, node_name, partition_layer_name, std::move(output), std::move(result_initializer)}, + join_{make_join(g, std::make_index_sequence{})}, + name_{node_name}, + partition_layer_{partition_layer_name}, + layers_{std::move(layer_names)} + { + assert(NInputs == layers_.size()); + + // We do not worry about collapsing layers for the fold-join node as the accumulator + // repeater will always belong to a higher layer than (at least one of) the data products + // that serve as input to the fold operation. + repeaters_.reserve(NInputs); + for (auto const& layer : layers_) { + repeaters_.push_back(std::make_unique(g, name_, layer)); + } + + make_edge(tbb::flow::output_port<1>(result_repeater_), input_port<0>(join_)); + auto set_ports = [this](std::index_sequence) { + this->set_external_ports( + input_t{result_repeater_.partition_port(), repeaters_[Is]->data_port()...}, + output_t{join_}); + // Connect repeaters to join + (make_edge(*repeaters_[Is], input_port(join_)), ...); + }; + + set_ports(std::make_index_sequence{}); + } + + std::size_t emitted_result_count() const { return result_repeater_.emitted_result_count(); } + + tbb::flow::receiver& partition_port() + { + return result_repeater_.partition_port(); + } + tbb::flow::sender& output_port() + { + return tbb::flow::output_port<0>(result_repeater_); + } + tbb::flow::receiver& notify_result_repeater_port() + { + return input_port<3>(result_repeater_); + } + + // Returns one named_index_port per slot the index router must wire up. + // + // The first entry is the *partition slot* — it routes the partition index to the + // accumulator's index_port and delivers the partition-level flush token to the + // accumulator's flush_port. Its routing layer is the partition layer (e.g. "job"), + // but its counting layer is the fold's most-derived input data layer (e.g. "event"), + // because the accumulator is incremented once per fold call (= once per cell at the + // most-derived input layer) and the flush count must balance against that. + // + // The remaining entries are the data-input slots, one per repeater. For these the + // routing layer and counting layer are the same: the repeater both routes and counts + // at its own input layer. + // + // FIXME: For multi-input folds, the "most-derived" input layer is the deepest of + // `layers_` in the data hierarchy. Until the router exposes a depth comparison, we + // pick `layers_[0]`; all current tests have single-input folds so this is correct in + // practice. + std::vector index_ports() + { + std::vector result; + result.reserve(1 + repeaters_.size()); // +1 for the result repeater + identifier const counting_layer_for_partition = + layers_.empty() ? partition_layer_ : layers_[0]; + result.emplace_back(partition_layer_, + counting_layer_for_partition, + &result_repeater_.flush_port(), + &result_repeater_.index_port()); + for (std::size_t i = 0; i != repeaters_.size(); ++i) { + result.emplace_back( + layers_[i], layers_[i], &repeaters_[i]->flush_port(), &repeaters_[i]->index_port()); + } + return result; + } + + private: + detail::accumulator_node result_repeater_; + std::vector> repeaters_; + tbb::flow::join_node join_; + // Immutable after construction; tbb::flow::join_node is already non-movable. + // NOLINTBEGIN(cppcoreguidelines-avoid-const-or-ref-data-members) + std::string const name_; + identifier const partition_layer_; + std::vector const layers_; + // NOLINTEND(cppcoreguidelines-avoid-const-or-ref-data-members) + }; + + // Translates a runtime port index into a reference to the corresponding compile-time + // input port by recursively incrementing the compile-time parameter I until it matches + // the runtime index. + template + tbb::flow::receiver& receiver_for(fold_join_node& join, + std::size_t const index) + { + if constexpr (I < N + 1) { // +1 to account for the result repeater port at index 0 + if (I != index + 1ull) { // +1 to account for the result repeater port at index 0 + return receiver_for(join, index); + } + return input_port(join); + } + throw std::runtime_error("Should never get here"); + } + + namespace detail { + // Returns pointers to all N input ports of the join node. Only valid for N > 1; + // callers with a single input should use the node directly. + template + std::vector*> input_ports(fold_join_node& join) + { + return [&join]( + std::index_sequence) -> std::vector*> { + return {&input_port<1 + Is>(join)...}; // +1 to skip the result repeater port + }(std::make_index_sequence{}); + } + + // Looks up the port index for the given input product query, then returns a reference to + // the corresponding input port of the join node. Only valid for N > 1. + template + tbb::flow::receiver& receiver_for(fold_join_node& join, + product_selectors const& input_products, + product_selector const& input_product) + { + auto const index = port_index_for(input_products, input_product); + return receiver_for<1ull>(join, index); // Start at 1 to skip the result repeater port + } + } + + // Returns all input-port pointers for a node. For N == 1 the node itself is the sole + // receiver; for N > 1 each port of the join is returned. + template + std::vector*> input_ports(fold_join_node& join) + { + return detail::input_ports(join); + } + + // Returns the receiver for the input port that corresponds to the given input product query. + // For N == 1 there is only one port so the node itself is returned; for N > 1 the port is + // looked up by query within the join. + template + tbb::flow::receiver& receiver_for(fold_join_node& join, + product_selectors const& input_products, + product_selector const& input_product) + { + return detail::receiver_for(join, input_products, input_product); + } +} + +#endif // PHLEX_CORE_FOLD_JOIN_NODE_HPP diff --git a/phlex/core/framework_graph.cpp b/phlex/core/framework_graph.cpp index aa58af777..1251e612b 100644 --- a/phlex/core/framework_graph.cpp +++ b/phlex/core/framework_graph.cpp @@ -15,6 +15,67 @@ #include namespace phlex::experimental { + namespace { + template + auto internal_edges_for_predicates(oneapi::tbb::flow::graph& g, + declared_predicates& all_predicates, + T const& consumers) + { + std::map result; + for (auto const& [name, consumer] : consumers) { + auto const& predicates = consumer->when(); + if (empty(predicates)) { + continue; + } + + auto [it, success] = result.try_emplace(name, g, *consumer); + for (auto const& predicate_name : predicates) { + if (auto predicate = all_predicates.get(predicate_name)) { + make_edge(predicate->sender(), it->second.predicate_port()); + continue; + } + throw std::runtime_error(std::format( + "A non-existent filter with the name '{}' was specified for {}", predicate_name, name)); + } + } + return result; + } + + index_router::fold_partition_ports_t fold_partition_ports(declared_folds const& folds) + { + index_router::fold_partition_ports_t result; + for (auto& [fold_name, fold_node] : folds) { + result.try_emplace(fold_name, fold_node->partition_layer(), &fold_node->partition_port()); + } + 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(). + index_router::unfold_data unfold_layers(declared_unfolds const& unfolds) + { + index_router::unfold_data result; + for (auto const& n : unfolds | std::views::values) { + identifier const child_layer{n->child_layer()}; + for (auto const& input : n->input()) { + auto const& input_layer = static_cast(input.layer); + if (input_layer.empty()) { + continue; + } + ++result.count_per_input_layer[input_layer]; + result.layer_pairs.push_back({.input = input_layer, .output = child_layer}); + } + } + return result; + } + } + framework_graph::framework_graph(int const max_parallelism) : framework_graph{[](framework_driver& driver) { driver.yield(data_cell_index::job()); }, max_parallelism} @@ -60,7 +121,7 @@ namespace phlex::experimental { framework_graph::~framework_graph() { if (shutdown_on_error_) { - // When in an error state, we need to sanely pop the layer stack and wait for any tasks to finish. + // 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); index_router_.drain(std::move(remaining_flushes)); graph_.wait_for_all(); @@ -104,33 +165,6 @@ namespace phlex::experimental { graph_.wait_for_all(); } - namespace { - template - auto internal_edges_for_predicates(oneapi::tbb::flow::graph& g, - declared_predicates& all_predicates, - T const& consumers) - { - std::map result; - for (auto const& [name, consumer] : consumers) { - auto const& predicates = consumer->when(); - if (empty(predicates)) { - continue; - } - - auto [it, success] = result.try_emplace(name, g, *consumer); - for (auto const& predicate_name : predicates) { - if (auto predicate = all_predicates.get(predicate_name)) { - make_edge(predicate->sender(), it->second.predicate_port()); - continue; - } - throw std::runtime_error(std::format( - "A non-existent filter with the name '{}' was specified for {}", predicate_name, name)); - } - } - return result; - } - } - void framework_graph::throw_if_registration_errors() const { if (registration_errors_.empty()) { @@ -160,10 +194,6 @@ namespace phlex::experimental { make_edge(index_receiver_, hierarchy_node_); make_edge(index_router_.unfold_index_receiver(), hierarchy_node_); - for (auto& node : nodes_.folds | std::views::values) { - make_edge(index_router_.flusher(), node->flush_port()); - } - for (auto& node : nodes_.unfolds | std::views::values) { make_edge(node->output_index_port(), index_router_.unfold_index_receiver()); make_edge(node->flush_sender(), index_router_.unfold_flush_receiver()); @@ -186,41 +216,11 @@ namespace phlex::experimental { } // Index-router finalization makes edges between the index-set nodes and the provider nodes. - finalize_router(std::move(provider_input_ports), std::move(multilayer_join_index_ports)); - } - - // FIXME: Much, if not all, of this logic should be moved to the index_router. - void framework_graph::finalize_router( - index_router::provider_input_ports_t provider_input_ports, - std::map multilayer_join_index_ports) - { - std::set unfold_input_layer_names; - - // Count how many distinct unfold nodes consume each input layer. When that count is - // greater than one, the flush_gate for an index in that layer must collect a flush - // message from every unfold before it knows the total number of children it will see. - std::map unfold_count_per_input_layer; - for (auto const& n : nodes_.unfolds | std::views::values) { - for (auto const& input : n->input()) { - if (!static_cast(input.layer).empty()) { - unfold_input_layer_names.insert(input.layer); - ++unfold_count_per_input_layer[identifier{input.layer}]; - } - } - } - - std::vector unfold_output_layer_names; - for (auto const& n : nodes_.unfolds | std::views::values) { - unfold_output_layer_names.emplace_back(n->child_layer()); - } - - // FIXME: All of this should be collapsed into one call to index_router::finalize() - index_router_.establish_layers( - fixed_hierarchy_.layer_paths(), - std::vector(unfold_input_layer_names.begin(), unfold_input_layer_names.end()), - unfold_output_layer_names); - index_router_.register_unfold_count_per_input_layer(std::move(unfold_count_per_input_layer)); - index_router_.finalize( - graph_, std::move(provider_input_ports), std::move(multilayer_join_index_ports)); + index_router_.finalize(graph_, + fixed_hierarchy_.layer_paths(), + unfold_layers(nodes_.unfolds), + std::move(provider_input_ports), + fold_partition_ports(nodes_.folds), + std::move(multilayer_join_index_ports)); } } diff --git a/phlex/core/framework_graph.hpp b/phlex/core/framework_graph.hpp index 5a359ab16..67ab39e4b 100644 --- a/phlex/core/framework_graph.hpp +++ b/phlex/core/framework_graph.hpp @@ -167,8 +167,6 @@ namespace phlex::experimental { void throw_if_registration_errors() const; void make_filter_edges(); void make_bookkeeping_edges(); - void finalize_router(index_router::provider_input_ports_t provider_input_ports, - std::map multilayer_join_index_ports); resource_usage graph_resource_usage_{}; max_allowed_parallelism parallelism_limit_; diff --git a/phlex/core/fwd.hpp b/phlex/core/fwd.hpp index 076f4073f..d8db418b1 100644 --- a/phlex/core/fwd.hpp +++ b/phlex/core/fwd.hpp @@ -3,19 +3,14 @@ #include "phlex/model/fwd.hpp" -#include "oneapi/tbb/flow_graph.h" - namespace phlex::experimental { class consumer; class declared_output; class generator; - struct flush_message; class framework_graph; struct message; class index_router; class products_consumer; - - using flusher_t = tbb::flow::broadcast_node; } #endif // PHLEX_CORE_FWD_HPP diff --git a/phlex/core/index_router.cpp b/phlex/core/index_router.cpp index 2482e9446..1e9abe702 100644 --- a/phlex/core/index_router.cpp +++ b/phlex/core/index_router.cpp @@ -19,34 +19,42 @@ namespace phlex::experimental { //======================================================================================== // multilayer_slot implementation + // + // A multilayer_slot is a static, pre-registered routing component created once per + // `named_index_port` in `finalize()`. Each slot is responsible *only for message routing*: it + // decides whether a routed index covers this slot (via `matches_exactly` / `is_parent_of`) and + // forwards index messages to a single downstream `input_port` at the appropriate layer. + // + // The flush-side metadata for the same `named_index_port` — namely the `counting_layer` name + // used to look up `committed_counts_` entries at flush time, and the downstream `flush_port` + // that receives `indexed_end_token`s — lives in a paired `flush_spec` (see below). The two are + // stored side-by-side in `join_node_slots` so that `multilayer_slots_for` can pair them up + // while resolving end-token entries for a routed partition index. namespace detail { class multilayer_slot { public: multilayer_slot(tbb::flow::graph& g, identifier layer, - tbb::flow::receiver* flush_port, tbb::flow::receiver* input_port); void put_message(data_cell_index_ptr const& index, std::size_t message_id); - void put_end_token(data_cell_index_ptr const& index, flush_gate const& fc); bool matches_exactly(layer_path const& layer_path) const; bool is_parent_of(data_cell_index_ptr const& index) const; + identifier const& layer() const { return layer_; } + private: identifier layer_; index_set_node broadcaster_; - flush_node flusher_; }; multilayer_slot::multilayer_slot(tbb::flow::graph& g, identifier layer, - tbb::flow::receiver* flush_port, tbb::flow::receiver* input_port) : - layer_{std::move(layer)}, broadcaster_{g}, flusher_{g} + layer_{std::move(layer)}, broadcaster_{g} { make_edge(broadcaster_, *input_port); - make_edge(flusher_, *flush_port); } void multilayer_slot::put_message(data_cell_index_ptr const& index, std::size_t message_id) @@ -59,13 +67,6 @@ namespace phlex::experimental { broadcaster_.try_put({.index = index->parent(layer_), .msg_id = message_id}); } - void multilayer_slot::put_end_token(data_cell_index_ptr const& index, flush_gate const& fc) - { - // We're going to have to be a little more careful about this. The committed total count may - // not be enough granularity for some downstream nodes. - flusher_.try_put({.index = index, .count = static_cast(fc.committed_total_count())}); - } - bool multilayer_slot::matches_exactly(layer_path const& layer_path) const { return layer_path.ends_with(layer_); @@ -87,62 +88,142 @@ namespace phlex::experimental { assert(index); return route(index, index_is_lowest_layer(index), message_id); }}, - unfold_flush_receiver_{g, - tbb::flow::unlimited, - [this](unfold_flush input) -> tbb::flow::continue_msg { - auto&& [index, layer_hash, count] = input; - apply_expected_count(*gate_for(index), layer_hash, count); - flush_if_done(index); - return {}; - }}, - flusher_{g} + unfold_flush_receiver_{ + g, tbb::flow::unlimited, [this](unfold_flush input) -> tbb::flow::continue_msg { + auto&& [index, layer_hash, count] = input; + apply_expected_count(*gate_for(index), layer_hash, count); + flush_if_done(index); + return {}; + }} { } - void index_router::establish_layers(std::vector const& layer_paths_from_driver, - std::vector unfold_input_layer_names, - std::vector unfold_output_layer_names) + void index_router::finalize(tbb::flow::graph& g, + std::vector const& layer_paths_from_driver, + unfold_data unfolds, + provider_input_ports_t provider_input_ports, + fold_partition_ports_t fold_partition_ports, + std::map multilayer_join_ports) { - auto sorted_layer_paths = layer_paths_from_driver; - std::ranges::sort(sorted_layer_paths); + // We must have at least one provider port, or there can be no data to process. + assert(!provider_input_ports.empty()); - // In sorted order, a path can only be a prefix of paths that follow it. - for (std::size_t i = 0; i + 1 < sorted_layer_paths.size(); ++i) { - bool const is_not_lowest_layer = - sorted_layer_paths[i].is_strict_prefix_of(sorted_layer_paths[i + 1]); - if (is_not_lowest_layer) { - auto const layer_hash = sorted_layer_paths[i].hash(); - is_lowest_layer_hashes_.emplace(layer_hash, false); + establish_layer_hierarchy(std::move(layer_paths_from_driver), unfolds.layer_pairs); + unfold_count_per_input_layer_ = std::move(unfolds.count_per_input_layer); + 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, std::move(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. 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); + bool changed = true; + while (changed) { + changed = false; + 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)) { + continue; + } + layer_path candidate{parent_path.to_string() + "/" + std::string(output_name)}; + if (not std::ranges::contains(sorted_layer_paths_, candidate)) { + sorted_layer_paths_.push_back(std::move(candidate)); + changed = true; + } + } } } - unfold_input_layer_names_ = std::move(unfold_input_layer_names); - unfold_output_layer_names_ = std::move(unfold_output_layer_names); + std::ranges::sort(sorted_layer_paths_); + + // 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) { + auto const layer_hash = sorted_layer_paths_[i].hash(); + bool const is_lowest_layer = + 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. + is_lowest_layer_hashes_.emplace(layer_hash, is_lowest_layer); + } } - void index_router::finalize(tbb::flow::graph& g, - provider_input_ports_t provider_input_ports, - std::map multilayer_join_ports) + void index_router::wire_provider_index_sets(tbb::flow::graph& g, + provider_input_ports_t provider_input_ports) { - // We must have at least one provider port, or there can be no data to process. - assert(!provider_input_ports.empty()); - - // Create the index-set broadcast nodes for providers for (auto& [input_product, provider_port] : provider_input_ports | std::views::values) { auto [it, _] = index_set_nodes_.emplace(input_product.layer, std::make_shared(g)); make_edge(*it->second, *provider_port); } + } + void index_router::wire_fold_partition_index_sets(tbb::flow::graph& g, + fold_partition_ports_t fold_partition_ports) + { + for (auto& [fold_node_name, partition_port] : fold_partition_ports) { + auto const& [layer, port] = partition_port; + auto [it, _] = index_set_nodes_.emplace(static_cast(layer), + std::make_shared(g)); + make_edge(*it->second, *port); + } + } + + // -------------------------------------------------------------------------------------------- + // For each multi-layer join node: compute its deepest slot layer, normalize any slot whose + // counting_layer was left equal to its routing layer to the node's deepest layer instead, + // construct the multilayer_slot objects, and store everything in multilayer_join_slots_. + void index_router::build_multilayer_join_slots( + tbb::flow::graph& g, std::map multilayer_join_ports) + { for (auto const& [node_name, join_ports] : multilayer_join_ports) { spdlog::trace("Making multilayer slots for {}", node_name); - detail::multilayer_slots slots; - slots.reserve(join_ports.size()); - for (auto const& [layer, flush_port, input_port] : join_ports) { - auto slot = std::make_shared(g, layer, flush_port, input_port); - slots.push_back(slot); + + // 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 left `counting_layer == layer` + // (the "use my routing layer" default) gets the node's deepest layer instead. Slots that + // explicitly chose 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) { + slot_layers.push_back(port.layer); + } + identifier const node_deepest_layer = deepest_layer_name(slot_layers); + + detail::join_node_slots node_slots; + node_slots.slots.reserve(join_ports.size()); + node_slots.flush_specs.reserve(join_ports.size()); + for (auto const& [layer, counting_layer, flush_port, input_port] : join_ports) { + identifier const effective_counting_layer = + (counting_layer == layer) ? node_deepest_layer : counting_layer; + node_slots.slots.push_back(std::make_shared(g, layer, input_port)); + node_slots.flush_specs.push_back( + {.counting_layer = effective_counting_layer, .flush_port = flush_port}); } - multilayer_join_slots_.emplace(identifier{node_name}, std::move(slots)); + multilayer_join_slots_.emplace(identifier{node_name}, std::move(node_slots)); } } @@ -160,27 +241,23 @@ namespace phlex::experimental { index_set_node->try_put({.index = index, .msg_id = message_id}); } - auto [message_slots, end_token_slots] = multilayer_slots_for(index); + auto [message_slots, end_token_entries] = multilayer_slots_for(index); for (auto const& slot : *message_slots) { slot->put_message(index, message_id); } - // Lowest-layer indices have no flush gate and contribute to their parent's readiness - // solely through the expected-count message that announced them — there is nothing - // to do here for them. + // Lowest-layer indices have no flush gate and contribute to their parent's readiness solely + // through the expected-count message that announced them — nothing to do here for them. if (is_lowest_layer) { return index; } gate_for(index)->set_flush_callback( - [this, end_token_slots = std::move(end_token_slots), index, message_id]( - flush_gate const& fc) { - for (auto const& slot : *end_token_slots) { - slot->put_end_token(index, fc); + [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); + entry.flush_port->try_put({.index = fc.index(), .count = static_cast(count)}); } - - // Used only for folds, until folds use the slot infrastructure above. - flusher_.try_put({index, fc.committed_counts(), message_id}); }); flush_if_done(index); @@ -190,12 +267,6 @@ namespace phlex::experimental { void index_router::drain(index_flushes flushes) { update_flush_counts(std::move(flushes)); } - void index_router::register_unfold_count_per_input_layer(std::map counts) - { - // Called once during finalize(), before any indices are routed, so no concurrent access. - unfold_count_per_input_layer_ = std::move(counts); - } - bool index_router::index_is_lowest_layer(data_cell_index_ptr const& index) { auto it = is_lowest_layer_hashes_.find(index->layer_hash()); @@ -203,16 +274,11 @@ namespace phlex::experimental { return it->second; } - if (std::ranges::contains(unfold_input_layer_names_, index->layer_name())) { - // FIXME: Need to make sure that the index is a child of existing layers - return is_lowest_layer_hashes_.emplace(index->layer_hash(), false).first->second; - } - - if (std::ranges::contains(unfold_output_layer_names_, index->layer_name())) { - return is_lowest_layer_hashes_.emplace(index->layer_hash(), true).first->second; - } - - // If the index is neither and input or an output to an unfold, it is assumed to be a lowest layer. + // 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; } @@ -238,7 +304,7 @@ namespace phlex::experimental { } } - if (candidates.size() == 1ull) { + if (candidates.size() == 1uz) { return candidates[0]->second; } @@ -253,7 +319,7 @@ namespace phlex::experimental { throw std::runtime_error(msg); } - std::pair + std::pair index_router::multilayer_slots_for(data_cell_index_ptr const& index) { auto const layer_hash = index->layer_hash(); @@ -262,7 +328,7 @@ namespace phlex::experimental { { multilayer_slot_cache_const_accessor acc; if (multilayer_slot_cache_.find(acc, layer_hash)) { - return {acc->second.message_slots, acc->second.end_token_slots}; + return {acc->second.message_slots, acc->second.end_token_entries}; } } @@ -270,29 +336,55 @@ namespace phlex::experimental { multilayer_slot_cache_accessor acc; auto const inserted = multilayer_slot_cache_.insert(acc, layer_hash); if (not inserted) { - return {acc->second.message_slots, acc->second.end_token_slots}; + return {acc->second.message_slots, acc->second.end_token_entries}; } auto const layer_path = index->layer_path(); detail::multilayer_slots message_slots; - detail::multilayer_slots end_token_slots; + detail::end_token_entries end_token_entries; // For each multi-layer join node, determine which slots are relevant to this index. - // Message entries: All slots from a node are added if (1) at least one slot exactly - // matches the current layer, and (2) all slots either exactly match - // or are parent layers of the current index. - // End-token entries: Only slots that exactly match the current layer are added. - for (auto& [node_name, slots] : multilayer_join_slots_) { + // Message entries: All slots from a node are added if (1) at least one slot exactly matches + // the current layer, and (2) all slots either exactly match or are parent + // layers of the current index. + // End-token entries: For each slot that exactly matches the current layer, consult the slot's + // paired flush_spec to materialize one entry per path-aware counting-layer + // descendant hash. When the flush_spec's counting layer equals the slot's + // routing layer (the common case), this resolves to exactly the routed + // index's own layer_hash. When they differ (a fold's partition slot), it + // resolves to one entry per descendant of `layer_path` whose trailing layer + // name equals the counting layer. + for (auto& [node_name, node_slots] : multilayer_join_slots_) { + auto const& slots = node_slots.slots; + auto const& flush_specs = node_slots.flush_specs; + assert(slots.size() == flush_specs.size()); + detail::multilayer_slots matching_slots; matching_slots.reserve(slots.size()); bool has_exact_match = false; std::size_t matched_count = 0; - for (auto& slot : slots) { + for (std::size_t i = 0; i != slots.size(); ++i) { + auto& slot = slots[i]; + auto const& flush = flush_specs[i]; if (slot->matches_exactly(layer_path)) { has_exact_match = true; - end_token_slots.push_back(slot); + if (flush.counting_layer == slot->layer()) { + // Counting layer is the routing layer: the routed index's own layer_hash is the unique + // counting hash. + end_token_entries.push_back( + {.counting_layer_hash = layer_hash, .flush_port = flush.flush_port}); + } else { + // Counting layer differs (fold partition slot). Enumerate all descendant paths under + // the routed partition path whose trailing name equals the counting layer; emit one + // entry per descendant. + 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}); + } + } matching_slots.push_back(slot); ++matched_count; } else if (slot->is_parent_of(index)) { @@ -312,9 +404,60 @@ namespace phlex::experimental { acc->second.message_slots = std::make_shared(std::move(message_slots)); - acc->second.end_token_slots = - std::make_shared(std::move(end_token_slots)); - return {acc->second.message_slots, acc->second.end_token_slots}; + acc->second.end_token_entries = + std::make_shared(std::move(end_token_entries)); + return {acc->second.message_slots, acc->second.end_token_entries}; + } + + std::vector index_router::counting_layer_hashes_under( + layer_path const& partition_layer_path, identifier const& counting_layer_name) const + { + std::vector result; + for (auto const& candidate : sorted_layer_paths_) { + // candidate must be a strict descendant of the partition layer path + if (not partition_layer_path.is_strict_prefix_of(candidate)) { + continue; + } + // ...and its trailing segment must equal the counting layer name + if (not candidate.ends_with(counting_layer_name)) { + continue; + } + result.push_back(candidate.hash()); + } + return result; + } + + identifier index_router::deepest_layer_name(std::vector const& layer_names) const + { + // 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 `counting_layer == layer` is + // correct for every slot 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(); + if (depth > best) { + best = depth; + } + } + } + return best; + }; + + assert(not layer_names.empty()); + auto const* result = &layer_names.front(); + std::size_t best_depth = depth_of(*result); + for (auto const& name : layer_names | std::views::drop(1)) { + auto const d = depth_of(name); + if (d > best_depth or (d == best_depth and name < *result)) { + result = &name; + best_depth = d; + } + } + return *result; } void index_router::update_flush_counts(index_flushes flushes) @@ -332,16 +475,16 @@ namespace phlex::experimental { data_cell_index::hash_type const child_layer_hash, std::size_t const count) { - // Non-lowest children contribute to the parent's readiness via rollup - // (roll_up_child() called from flush_if_done()). Lowest-layer children need no - // further accounting: their full count is already reflected in the expected-count - // message and will be merged into committed_counts_ at commit time. + // Non-lowest children contribute to the parent's readiness via rollup (roll_up_child() called + // from flush_if_done()). Lowest-layer children need no further accounting: their full count is + // already reflected in the expected-count message and will be merged into committed_counts_ at + // commit time. // // The pending counter must be bumped BEFORE update_expected_count() increments - // received_flush_count_; otherwise a concurrent all_children_accounted() call could - // observe a positive received count while the pending counter is still at its - // pre-bump value (which may be at or below zero from earlier rollup notifications) - // and erroneously declare the tracker ready. + // received_flush_count_; otherwise a concurrent all_children_accounted() call could observe a + // positive received count while the pending counter is still at its pre-bump value (which may + // be at or below zero from earlier rollup notifications) and erroneously declare the tracker + // ready. if (not is_lowest_layer_hash(child_layer_hash)) { gate.expect_child_rollups(static_cast(count)); } @@ -367,9 +510,9 @@ namespace phlex::experimental { accessor a; if (flush_gates_.insert(a, index->hash())) { // Newly inserted — initialize the value. - // If multiple unfolds consume this layer, the gate must wait for a flush message - // from each 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. + // If multiple unfolds consume this layer, the gate must wait for a flush message from each + // 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; @@ -384,16 +527,16 @@ namespace phlex::experimental { assert(index); while (index) { - // Erase the entry while holding the exclusive accessor, then release the lock before - // calling send_flush(). The erase claims exclusive ownership of this gate — - // any concurrent flush_if_done call for the same index will fail to find the entry - // and return immediately, preventing double-flush. + // Erase the entry while holding the exclusive accessor, then release the lock before calling + // send_flush(). The erase claims exclusive ownership of this gate — any concurrent + // flush_if_done call for the same index will fail to find the entry and return immediately, + // preventing double-flush. flush_gate_ptr gate; { accessor a; if (not flush_gates_.find(a, index->hash())) { - // This can happen when two threads process the same parent index, - // and one of them releases it before the other completes. + // This can happen when two threads process the same parent index, and one of them + // releases it before the other completes. return; } diff --git a/phlex/core/index_router.hpp b/phlex/core/index_router.hpp index 8e0fb8bb3..f9c6618dc 100644 --- a/phlex/core/index_router.hpp +++ b/phlex/core/index_router.hpp @@ -23,17 +23,46 @@ namespace phlex::experimental { namespace detail { using index_set_node = tbb::flow::broadcast_node; using index_set_node_ptr = std::shared_ptr; - using flush_node = tbb::flow::broadcast_node; // ========================================================================================== - // A multilayer_slot manages routing and flushing for a single layer slot (a repeater) in - // multi-layer join nodes. Each slot corresponds to one input data layer of a multi-layer - // join operation. It: - // (a) routes index messages to either the matching layer or its data-layer parent, and - // (b) emits flush tokens to the repeater to evict a cached data product from memory. + // A multilayer_slot (one per registered named_index_port) captures the routing decision + // for one input slot of a multi-layer join node. See the implementation file for the full + // description. The slot is concerned purely with message routing — flush-side metadata is + // carried separately in a paired flush_spec (see below) so that the slot need not know + // about counting layers or flush ports. 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`. + struct flush_spec { + identifier counting_layer; + tbb::flow::receiver* flush_port; + }; + + // Per multi-layer join node: the slots, and the flush metadata paired positionally + // with each slot (slots[i] corresponds to flush_specs[i]). Stored together so + // that multilayer_slots_for can iterate the two in lockstep. + struct join_node_slots { + multilayer_slots slots; + 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. + struct end_token_entry { + std::size_t counting_layer_hash; + tbb::flow::receiver* flush_port; + }; + using end_token_entries = std::vector; + using end_token_entries_ptr = std::shared_ptr; } class PHLEX_CORE_EXPORT index_router { @@ -53,23 +82,48 @@ namespace phlex::experimental { }; using provider_input_ports_t = std::map; + struct fold_input_port_t { + identifier layer; + tbb::flow::receiver* partition_port; + }; + 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 { + identifier input; + identifier output; + }; + explicit index_router(tbb::flow::graph& g); data_cell_index_ptr route(data_cell_index_ptr const& index, index_flushes flushes); - void establish_layers(std::vector const& layer_paths_from_driver, - std::vector unfold_input_layer_names, - std::vector unfold_output_layer_names); - - // Registers how many unfolds produce children from each input layer. Must be called - // before execution so that flush_gates are initialized with the correct expected - // child count when they are first created. - void register_unfold_count_per_input_layer(std::map counts); + // 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; + }; void finalize(tbb::flow::graph& g, + std::vector const& layer_paths_from_driver, + unfold_data unfolds, provider_input_ports_t provider_input_ports, + fold_partition_ports_t fold_partition_ports, std::map multilayer_join_ports); void drain(index_flushes flushes); - flusher_t& flusher() { return flusher_; } tbb::flow::function_node& unfold_index_receiver() { @@ -91,9 +145,18 @@ namespace phlex::experimental { // correct for unfold outputs (the only source of unknown hashes) and consistent with // index_is_lowest_layer()'s fall-through default. 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); + 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); + void build_multilayer_join_slots( + tbb::flow::graph& g, std::map multilayer_join_ports); detail::index_set_node_ptr index_set_node_for(layer_path const& layer); detail::index_set_node_ptr index_set_node_for(data_cell_index_ptr const& index); - std::pair multilayer_slots_for( + std::pair multilayer_slots_for( data_cell_index_ptr const& index); void update_flush_counts(index_flushes flushes); void apply_expected_count(flush_gate& gate, @@ -102,13 +165,31 @@ namespace phlex::experimental { 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. + std::vector counting_layer_hashes_under( + layer_path const& partition_layer_path, 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. + identifier deepest_layer_name(std::vector const& layer_names) const; + tbb::flow::function_node unfold_index_receiver_; tbb::flow::function_node unfold_flush_receiver_; std::atomic received_indices_{}; - flusher_t flusher_; tbb::concurrent_unordered_map is_lowest_layer_hashes_; - std::vector unfold_input_layer_names_; - std::vector unfold_output_layer_names_; + // Layer paths from the driver, 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_; // ========================================================================================== // Routing to provider nodes @@ -122,13 +203,25 @@ namespace phlex::experimental { // ========================================================================================== // Routing to multi-layer join nodes // Maps from join-node name to the multilayer slots for that node. - tbb::concurrent_unordered_map multilayer_join_slots_; - - // This struct lets get_multilayer_slots return message and end-token slots together, - // instead of passing concurrent_hash_map accessors as output parameters. + tbb::concurrent_unordered_map 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. + // The two are populated independently: + // - message_slots: one entry per slot template whose routing layer covers this + // routed index (either an exact match or a parent layer). + // - end_token_entries:one entry per (slot template, counting-layer descendant path). + // For non-fold cases the counting layer equals the routing + // layer, so a slot at the routed index's layer contributes a + // single entry whose counting_layer_hash is the routed index's + // own layer_hash. For a fold's partition slot the routing layer + // is the partition (e.g. "job") while the counting layer is the + // fold's input data layer (e.g. "event"), so the slot may + // contribute multiple entries — one per "event"-named + // descendant of the partition path. struct multilayer_slot_cache_entry { detail::multilayer_slots_ptr message_slots; - detail::multilayer_slots_ptr end_token_slots; + detail::end_token_entries_ptr end_token_entries; }; // Cache from layer hash to matched message/end-token slots for that layer. using multilayer_slot_cache_t = diff --git a/phlex/core/message.hpp b/phlex/core/message.hpp index 0cf7e03d1..82846e4ef 100644 --- a/phlex/core/message.hpp +++ b/phlex/core/message.hpp @@ -28,18 +28,11 @@ namespace phlex::experimental { bool cache{true}; }; - // FIXME: Do we need both indexed_end_token and flush_message? struct indexed_end_token { data_cell_index_ptr index; int count; }; - struct flush_message { - data_cell_index_ptr index; - data_cell_counts_const_ptr counts; - std::size_t original_id{}; // FIXME: Used only by folds - }; - struct message { // FIXME: Maybe consider adding an 'index' data member? product_store_const_ptr store; @@ -56,8 +49,29 @@ namespace phlex::experimental { template using messages_t = std::conditional_t>; + // A named_index_port describes one input slot of a multi-layer join node from the + // perspective of the index router. Two distinct layer concepts are carried: + // + // - `layer` — the *routing* layer. The router decides whether a routed index + // feeds this slot's `index_port` (and whether the slot should + // receive a flush token) using this layer name via + // `matches_exactly` / `is_parent_of` checks. + // + // - `counting_layer` — the *counting* layer. When a flush gate fires at the routing + // layer, the count carried in the resulting `indexed_end_token` is + // the sum of `committed_counts_` entries whose path-aware layer + // hash corresponds to the `counting_layer` name *under* the routed + // partition path. For most slots `counting_layer == layer` (the + // slot's repeater both routes and counts at the same layer). For + // a fold's partition slot, however, the routing layer is the + // partition (e.g. "job") while the counting layer is the fold's + // most-derived input data layer (e.g. "event"): the accumulator + // is incremented once per fold-input-layer cell, so the flush + // token must balance against the count at that input layer rather + // than the partition layer. struct named_index_port { identifier layer; + identifier counting_layer; tbb::flow::receiver* token_port; tbb::flow::receiver* index_port; }; diff --git a/phlex/core/multilayer_join_node.hpp b/phlex/core/multilayer_join_node.hpp index b9e8d5486..b23df0bb1 100644 --- a/phlex/core/multilayer_join_node.hpp +++ b/phlex/core/multilayer_join_node.hpp @@ -111,7 +111,10 @@ namespace phlex::experimental { std::vector result; result.reserve(repeaters_.size()); for (auto const& [layer, repeater] : std::views::zip(layers_, repeaters_)) { - result.emplace_back(layer, &repeater->flush_port(), &repeater->index_port()); + // Counting layer is the same as the routing layer: this repeater both routes + // messages at `layer` and emits one message per cell at `layer`, so the flush + // token's count must balance against `layer`'s own committed-counts entries. + result.emplace_back(layer, layer, &repeater->flush_port(), &repeater->index_port()); } return result; } diff --git a/phlex/core/store_counters.cpp b/phlex/core/store_counters.cpp deleted file mode 100644 index eade0e867..000000000 --- a/phlex/core/store_counters.cpp +++ /dev/null @@ -1,86 +0,0 @@ -#include "phlex/core/store_counters.hpp" -#include "phlex/core/message.hpp" -#include "phlex/model/data_cell_counts.hpp" - -#include "fmt/std.h" -#include "spdlog/spdlog.h" - -#include - -namespace phlex::experimental { - - void store_counter::set_flush_value(data_cell_counts_const_ptr counts, - std::size_t const original_message_id) - { - if (not counts) { - return; - } - -#ifdef __cpp_lib_atomic_shared_ptr - flush_counts_ = counts; -#else - atomic_store(&flush_counts_, counts); -#endif - original_message_id_ = original_message_id; - } - - void store_counter::increment(data_cell_index::hash_type const layer_hash) - { - ++counts_[layer_hash]; - } - - bool store_counter::is_complete() - { - if (!ready_to_flush_) { - return false; - } - -#ifdef __cpp_lib_atomic_shared_ptr - auto flush_counts = flush_counts_.load(); -#else - auto flush_counts = atomic_load(&flush_counts_); -#endif - if (not flush_counts) { - return false; - } - - // The 'counts_' data member can be empty if the flush_counts member has been filled - // but none of the children stores have been processed. - if (counts_.empty() and flush_counts->size() > 0uz) { - return false; - } - - for (auto const& [layer_hash, count] : counts_) { - if (count != flush_counts->count(layer_hash)) { - return false; - } - } - - // Flush only once! - return ready_to_flush_.exchange(false); - } - - unsigned int store_counter::original_message_id() const noexcept { return original_message_id_; } - - store_counter& count_stores::counter_for(data_cell_index::hash_type const hash) - { - counter_accessor ca; - if (!counters_.find(ca, hash)) { - counters_.emplace(ca, hash, std::make_unique()); - } - return *ca->second; - } - - std::unique_ptr count_stores::done_with(data_cell_index::hash_type const hash) - { - // Must be called after an insertion has already been performed - counter_accessor ca; - bool const found = counters_.find(ca, hash); - if (found and ca->second->is_complete()) { - std::unique_ptr result{std::move(ca->second)}; - counters_.erase(ca); - return result; - } - return nullptr; - } -} diff --git a/phlex/core/store_counters.hpp b/phlex/core/store_counters.hpp deleted file mode 100644 index 81a2b1d1a..000000000 --- a/phlex/core/store_counters.hpp +++ /dev/null @@ -1,53 +0,0 @@ -#ifndef PHLEX_CORE_STORE_COUNTERS_HPP -#define PHLEX_CORE_STORE_COUNTERS_HPP - -#include "phlex/phlex_core_export.hpp" - -#include "phlex/core/fwd.hpp" -#include "phlex/model/data_cell_index.hpp" -#include "phlex/model/product_store.hpp" - -#include "oneapi/tbb/concurrent_hash_map.h" -#include "oneapi/tbb/concurrent_unordered_map.h" - -#include -#include -#include - -namespace phlex::experimental { - class PHLEX_CORE_EXPORT store_counter { - public: - void set_flush_value(data_cell_counts_const_ptr counts, std::size_t original_message_id); - void increment(data_cell_index::hash_type layer_hash); - bool is_complete(); - unsigned int original_message_id() const noexcept; - - private: - using counts_t = - tbb::concurrent_unordered_map>; - - counts_t counts_{}; -#ifdef __cpp_lib_atomic_shared_ptr - std::atomic flush_counts_{nullptr}; -#else - data_cell_counts_const_ptr flush_counts_{nullptr}; -#endif - unsigned int original_message_id_{}; // Necessary for matching inputs to downstream join nodes. - std::atomic ready_to_flush_{true}; - }; - - class PHLEX_CORE_EXPORT count_stores { - protected: - store_counter& counter_for(data_cell_index::hash_type hash); - std::unique_ptr done_with(data_cell_index::hash_type hash); - - private: - using counters_t = - tbb::concurrent_hash_map>; - using counter_accessor = counters_t::accessor; - - counters_t counters_; - }; -} - -#endif // PHLEX_CORE_STORE_COUNTERS_HPP diff --git a/phlex/model/flush_gate.cpp b/phlex/model/flush_gate.cpp index 828d804dc..8984cc5e7 100644 --- a/phlex/model/flush_gate.cpp +++ b/phlex/model/flush_gate.cpp @@ -1,19 +1,14 @@ #include "phlex/model/flush_gate.hpp" -#include "spdlog/spdlog.h" - #include #include -#include #include #include namespace phlex::experimental { flush_gate::flush_gate(data_cell_index_ptr index, std::size_t expected_flush_count) : - index_{std::move(index)}, - committed_counts_{std::make_shared()}, - expected_flush_count_{expected_flush_count} + index_{std::move(index)}, expected_flush_count_{expected_flush_count} { } @@ -24,13 +19,13 @@ namespace phlex::experimental { std::size_t flush_gate::committed_total_count() const { - return std::ranges::fold_left(*committed_counts_ | std::views::values, 0uz, std::plus{}); + return std::ranges::fold_left(committed_counts_ | std::views::values, 0uz, std::plus{}); } std::size_t flush_gate::committed_count_for_layer( data_cell_index::hash_type const layer_hash) const { - return committed_counts_->count(layer_hash); + return committed_counts_.count(layer_hash); } void flush_gate::update_expected_count(data_cell_index::hash_type const layer_hash, @@ -40,11 +35,10 @@ namespace phlex::experimental { ++received_flush_count_; } - void flush_gate::roll_up_child(data_cell_counts_const_ptr child_committed_counts) + void flush_gate::roll_up_child(data_cell_counts const& child_committed_counts) { - assert(child_committed_counts); - for (auto const& [layer_hash, count] : *child_committed_counts) { - committed_counts_->add_to(layer_hash, count); + for (auto const& [layer_hash, count] : child_committed_counts) { + committed_counts_.add_to(layer_hash, count); } --pending_child_rollups_; } @@ -89,7 +83,7 @@ namespace phlex::experimental { void flush_gate::commit() { for (auto const& [layer_hash, count] : expected_counts_) { - committed_counts_->add_to(layer_hash, count.load()); + committed_counts_.add_to(layer_hash, count.load()); } // At some point, we might consider clearing the expected_counts_ map to free memory, diff --git a/phlex/model/flush_gate.hpp b/phlex/model/flush_gate.hpp index 7cea5adc5..3d29d7536 100644 --- a/phlex/model/flush_gate.hpp +++ b/phlex/model/flush_gate.hpp @@ -40,6 +40,7 @@ #include #include #include +#include namespace phlex::experimental { @@ -57,7 +58,7 @@ namespace phlex::experimental { std::size_t expected_total_count() const; std::size_t committed_total_count() const; std::size_t committed_count_for_layer(data_cell_index::hash_type layer_hash) const; - data_cell_counts_const_ptr committed_counts() const { return committed_counts_; } + data_cell_counts const& committed_counts() const { return committed_counts_; } // Merges an expected child count into the accumulated expected counts. Each call // represents one flush message arriving (e.g. one unfold completing for this index). @@ -66,7 +67,7 @@ namespace phlex::experimental { // Records that a non-lowest direct child has rolled up: merges its committed_counts // into this gate's and decrements the pending-rollups balance. The two steps are // bundled because every rollup must do both, in the same call. - void roll_up_child(data_cell_counts_const_ptr child_committed_counts); + void roll_up_child(data_cell_counts const& child_committed_counts); // Announces that n additional non-lowest direct children are expected to roll up. // Lowest-layer children require no such bookkeeping: their counts are fully accounted @@ -84,11 +85,7 @@ namespace phlex::experimental { data_cell_index_ptr const index_; std::once_flag commit_once_; - // FIXME: We express committed_counts_ as a shared pointer so that we can copy the committed - // counts (this is done for determining the flush values for folds). Once the fold - // flushes are incorporated as part of the multi-layer join node infrastructure, it - // should be possible for committed_counts_ to no longer be a pointer, but a value. - std::shared_ptr committed_counts_; + data_cell_counts committed_counts_; // Accumulated expected child counts from all unfolds. data_cell_counts expected_counts_; std::atomic received_flush_count_{0}; diff --git a/phlex/model/layer_path.cpp b/phlex/model/layer_path.cpp index 3c15e506e..b4321b420 100644 --- a/phlex/model/layer_path.cpp +++ b/phlex/model/layer_path.cpp @@ -100,4 +100,6 @@ namespace phlex::experimental { } return hashes; } + + std::size_t layer_path::depth() const noexcept { return layer_path_.size(); } } diff --git a/phlex/model/layer_path.hpp b/phlex/model/layer_path.hpp index 112f268af..502e90ae5 100644 --- a/phlex/model/layer_path.hpp +++ b/phlex/model/layer_path.hpp @@ -44,6 +44,9 @@ namespace phlex::experimental { /// Return hash of this path and every parent path std::set hashes() const; + /// Return the number of layer names in the path + std::size_t depth() const noexcept; + private: std::vector layer_path_; void validate() const; diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 4c2d93575..230319a1a 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -11,6 +11,7 @@ include(CetTest) cet_test_env(SPDLOG_LEVEL=debug) +cet_test(accumulator USE_CATCH2_MAIN SOURCE accumulator_test.cpp LIBRARIES phlex::core_internal) cet_test(concepts SOURCE concepts.cpp LIBRARIES phlex::core_internal) cet_test( adjust_config diff --git a/test/accumulator_test.cpp b/test/accumulator_test.cpp new file mode 100644 index 000000000..f205d26f3 --- /dev/null +++ b/test/accumulator_test.cpp @@ -0,0 +1,293 @@ +#include "phlex/core/detail/accumulator_node.hpp" +#include "phlex/model/data_cell_index.hpp" +#include "phlex/model/product_specification.hpp" + +#include "catch2/catch_test_macros.hpp" +#include "catch2/matchers/catch_matchers_string.hpp" +#include "oneapi/tbb/flow_graph.h" +#include "spdlog/sinks/ostream_sink.h" +#include "spdlog/spdlog.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace Catch { + // Specialize Catch's StringMaker for the fold result type to get better test output formatting + template <> + struct StringMaker> { + static std::string convert(std::pair const& p) + { + return std::format("{{{}, {}}}", p.first, p.second); + } + }; +} + +using namespace oneapi; +using namespace phlex::experimental; + +namespace { + auto make_run_index(int run_number) + { + return phlex::data_cell_index::job()->make_child("run", run_number); + } + + class message_collector : public tbb::flow::function_node { + public: + explicit message_collector(tbb::flow::graph& g) : + tbb::flow::function_node{g, tbb::flow::serial, [this](message const& msg) { + messages_.emplace(msg.store->index()->hash(), msg); + }} + { + } + + std::size_t received_result_count() const { return messages_.size(); } + + std::pair fold_result(phlex::data_cell_index_ptr const& idx) const + { + auto it = messages_.find(idx->hash()); + assert(it != messages_.end()); + return {it->second.store->get_product(product_specification{}), it->second.id}; + } + + private: + std::map messages_; + }; + + void use_ostream_logger(std::ostringstream& oss) + { + auto ostream_sink = std::make_shared(oss); + auto ostream_logger = std::make_shared("my_logger", ostream_sink); + spdlog::set_default_logger(ostream_logger); + } + + // Fold function: increments the accumulator by 1 per call. The accumulated value thus equals + // the number of data cells processed. + auto count_fold(detail::accumulator_message const& msg) -> std::size_t + { + msg.partial_result->call([](int& acc) { + // The count_fold function may be called concurrently for the same accumulator, so we use + // atomic_ref to safely update the count. + std::atomic_ref(acc) += 1; + }); + return msg.index->hash(); + } + + class accumulator_test_fixture { + public: + explicit accumulator_test_fixture(std::string node_name) : + accumulator_{g_, + std::move(node_name), + "run"_id, + product_specifications(1), + [](phlex::data_cell_index const&) { return std::make_unique(0); }}, + fold_op_{g_, tbb::flow::unlimited, count_fold}, + result_consumer_{g_} + { + make_edge(output_port<0>(accumulator_), result_consumer_); + make_edge(output_port<1>(accumulator_), fold_op_); + make_edge(fold_op_, input_port<3>(accumulator_)); + } + + void put_partition(index_message const& msg) { accumulator_.partition_port().try_put(msg); } + void put_index_message(index_message const& msg) { accumulator_.index_port().try_put(msg); } + void put_flush_token(indexed_end_token const& t) { accumulator_.flush_port().try_put(t); } + + void wait_for_all() { g_.wait_for_all(); } + + bool cache_is_empty() const { return accumulator_.cache_is_empty(); } + std::size_t cache_size() const { return accumulator_.cache_size(); } + std::size_t emitted_result_count() const { return accumulator_.emitted_result_count(); } + std::size_t received_result_count() const { return result_consumer_.received_result_count(); } + + std::pair fold_result(phlex::data_cell_index_ptr const& idx) const + { + return result_consumer_.fold_result(idx); + } + + private: + tbb::flow::graph g_; + detail::accumulator_node accumulator_; + tbb::flow::function_node, std::size_t> fold_op_; + message_collector result_consumer_; + }; +} + +TEST_CASE("Test accumulator basic functionality", "[multithreading]") +{ + auto partition_index = make_run_index(1); + accumulator_test_fixture fixture{"test_accumulator_multiple_cells"}; + + fixture.put_partition({.index = partition_index, .msg_id = 1}); + auto const num_data_cells = 5; + for (std::size_t i = 0; i < num_data_cells; ++i) { + fixture.put_index_message({.index = partition_index, .msg_id = 2 + i, .cache = true}); + } + fixture.wait_for_all(); + + // Fold is complete but no flush received — result should not be emitted yet + CHECK(fixture.emitted_result_count() == 0); + CHECK(fixture.received_result_count() == 0); + CHECK(fixture.cache_size() == 1); + + fixture.put_flush_token({.index = partition_index, .count = num_data_cells}); + fixture.wait_for_all(); + + // Exactly one result carrying the accumulated count + CHECK(fixture.emitted_result_count() == 1); + REQUIRE(fixture.received_result_count() == 1); + CHECK(fixture.fold_result(partition_index) == + std::pair{num_data_cells, + 1uz}); // Fold value equals number of data cells, original message ID 1 + CHECK(fixture.cache_is_empty()); +} + +TEST_CASE("Test accumulator with multiple partitions", "[multithreading]") +{ + auto run1 = make_run_index(1); + auto run2 = make_run_index(2); + accumulator_test_fixture fixture{"test_accumulator_multiple_partitions"}; + + // Feed both partitions interleaved + fixture.put_partition({.index = run1, .msg_id = 1}); + fixture.put_partition({.index = run2, .msg_id = 2}); + + // 3 data cells for run1, 2 for run2, interleaved + fixture.put_index_message({.index = run1, .msg_id = 3, .cache = true}); + fixture.put_index_message({.index = run2, .msg_id = 4, .cache = true}); + fixture.put_index_message({.index = run1, .msg_id = 5, .cache = true}); + fixture.put_index_message({.index = run2, .msg_id = 6, .cache = true}); + fixture.put_index_message({.index = run1, .msg_id = 7, .cache = true}); + + fixture.put_flush_token({.index = run1, .count = 3}); + fixture.put_flush_token({.index = run2, .count = 2}); + fixture.wait_for_all(); + + // One result per partition: run1 accumulates 3 calls, run2 accumulates 2 calls + CHECK(fixture.emitted_result_count() == 2); + REQUIRE(fixture.received_result_count() == 2); + CHECK(fixture.fold_result(run1) == std::pair{3, 1uz}); // 3 fold operations, original message ID 1 + CHECK(fixture.fold_result(run2) == std::pair{2, 2uz}); // 2 fold operations, original message ID 2 + CHECK(fixture.cache_is_empty()); +} + +TEST_CASE("Test accumulator with index messages before partition", "[multithreading]") +{ + auto partition_index = make_run_index(1); + accumulator_test_fixture fixture{"test_accumulator_index_before_partition"}; + + // Send index messages before the partition message — they should be queued + fixture.put_index_message({.index = partition_index, .msg_id = 1, .cache = true}); + fixture.put_index_message({.index = partition_index, .msg_id = 2, .cache = true}); + fixture.wait_for_all(); + + // Cache entry exists but accumulator is not yet created, so no fold operations have run + CHECK(fixture.emitted_result_count() == 0); + CHECK(fixture.received_result_count() == 0); + CHECK(fixture.cache_size() == 1); + + // Partition arrives — pending index messages are dispatched and folds run + fixture.put_partition({.index = partition_index, .msg_id = 3}); + fixture.wait_for_all(); + + // Fold operations are done but no flush received yet + CHECK(fixture.emitted_result_count() == 0); + CHECK(fixture.received_result_count() == 0); + + fixture.put_flush_token({.index = partition_index, .count = 2}); + fixture.wait_for_all(); + + CHECK(fixture.emitted_result_count() == 1); + REQUIRE(fixture.received_result_count() == 1); + CHECK(fixture.fold_result(partition_index) == + std::pair{2, 3uz}); // 2 fold operations, original message ID 3 + CHECK(fixture.cache_is_empty()); +} + +TEST_CASE("Test accumulator warning message if cache is not flushed", "[multithreading]") +{ + std::ostringstream oss; + use_ostream_logger(oss); + + tbb::flow::graph g; + auto accumulator = std::make_unique>( + g, + "test_accumulator_warning", + "run"_id, + product_specifications(1), + [](phlex::data_cell_index const&) { return std::make_unique(0); }); + + SECTION("Partition received, no flush") + { + accumulator->partition_port().try_put({.index = make_run_index(1), .msg_id = 1}); + g.wait_for_all(); + + CHECK(accumulator->cache_size() == 1); + + accumulator.reset(); // Invoke destructor to trigger warning + + auto const warning = oss.str(); + CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Cached accumulators: 1")); + CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Partition [run:1]")); + } + + SECTION("Index message received, no partition") + { + accumulator->index_port().try_put({.index = make_run_index(1), .msg_id = 1, .cache = true}); + g.wait_for_all(); + + CHECK(accumulator->cache_size() == 1); + + accumulator.reset(); // Invoke destructor to trigger warning + + auto const warning = oss.str(); + CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Cached accumulators: 1")); + CHECK_THAT(warning, Catch::Matchers::ContainsSubstring("Partition index not yet received")); + } +} + +TEST_CASE("Test accumulator with zero data cells emits initial value", "[multithreading]") +{ + // This exercises the case where a partition exists but no data cells fall within it + // (e.g. all events were filtered out). A flush with count=0 should emit the initial + // accumulator value once both the partition and the flush have been received, without + // waiting for any fold operations. Both arrival orders must produce the same result. + auto partition_index = make_run_index(1); + accumulator_test_fixture fixture{"test_accumulator_zero_cells"}; + + SECTION("Put partition first") + { + fixture.put_partition({.index = partition_index, .msg_id = 1}); + fixture.wait_for_all(); + + CHECK(fixture.emitted_result_count() == 0); + CHECK(fixture.received_result_count() == 0); + CHECK(fixture.cache_size() == 1); + + fixture.put_flush_token({.index = partition_index, .count = 0}); + } + SECTION("Put flush token first") + { + fixture.put_flush_token({.index = partition_index, .count = 0}); + fixture.wait_for_all(); + + // Flush has been received but the partition has not — nothing to emit yet. + CHECK(fixture.emitted_result_count() == 0); + CHECK(fixture.received_result_count() == 0); + CHECK(fixture.cache_size() == 1); + + fixture.put_partition({.index = partition_index, .msg_id = 1}); + } + fixture.wait_for_all(); + + CHECK(fixture.emitted_result_count() == 1); + REQUIRE(fixture.received_result_count() == 1); + CHECK(fixture.fold_result(partition_index) == + std::pair{0, 1uz}); // Initial value, original message ID 1 + CHECK(fixture.cache_is_empty()); +} diff --git a/test/flush_gate_test.cpp b/test/flush_gate_test.cpp index 3412306c6..5b5a9392b 100644 --- a/test/flush_gate_test.cpp +++ b/test/flush_gate_test.cpp @@ -23,10 +23,10 @@ #include "oneapi/tbb/concurrent_hash_map.h" #include "oneapi/tbb/concurrent_vector.h" #include "oneapi/tbb/parallel_for.h" -#include "spdlog/spdlog.h" #include #include +#include using namespace phlex; using namespace phlex::experimental; @@ -222,13 +222,13 @@ TEST_CASE("flush_gate: roll_up_child accumulates across multiple children", "[fl job_gate->update_expected_count(run_layer_hash, 2); // Simulate run 0 rolling up with 3 spills. - auto run0_committed = std::make_shared(); - run0_committed->add_to(spill_layer_hash, 3); + data_cell_counts run0_committed; + run0_committed.add_to(spill_layer_hash, 3); job_gate->roll_up_child(run0_committed); // Simulate run 1 rolling up with 5 spills. - auto run1_committed = std::make_shared(); - run1_committed->add_to(spill_layer_hash, 5); + data_cell_counts run1_committed; + run1_committed.add_to(spill_layer_hash, 5); job_gate->roll_up_child(run1_committed); REQUIRE(job_gate->all_children_accounted());