diff --git a/phlex/core/fold/send.hpp b/phlex/core/fold/send.hpp index b936e9ba0..9d4a1a618 100644 --- a/phlex/core/fold/send.hpp +++ b/phlex/core/fold/send.hpp @@ -30,7 +30,7 @@ namespace phlex::experimental { }; template - concept move_constructible_only = std::move_constructible && !has_send; + concept sendable_by_move = std::move_constructible && !has_send; namespace detail { template @@ -41,7 +41,7 @@ namespace phlex::experimental { using type = decltype(send(std::declval())); }; - template + template struct sendable_type_impl { using type = T; }; diff --git a/phlex/model/flush_gate.cpp b/phlex/model/flush_gate.cpp index c44d457ce..8d1b6ae99 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::detail { 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::detail { 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::detail { ++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::detail { 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 4829a2feb..f4129abff 100644 --- a/phlex/model/flush_gate.hpp +++ b/phlex/model/flush_gate.hpp @@ -40,6 +40,7 @@ #include #include #include +#include namespace phlex::detail { @@ -57,7 +58,7 @@ namespace phlex::detail { 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::detail { // 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::detail { 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/test/flush_gate_test.cpp b/test/flush_gate_test.cpp index 08042ad85..ce8f81ffc 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::detail; @@ -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());