refactor: clean up sendability naming and flush-gate count typing

phlex/core/fold/send.hpp

@@ -30,7 +30,7 @@ namespace phlex::experimental {
   };
 
   template <typename T>
-  concept move_constructible_only = std::move_constructible<T> && !has_send<T>;
+  concept sendable_by_move = std::move_constructible<T> && !has_send<T>;
 
   namespace detail {
     template <typename T>
@@ -41,7 +41,7 @@ namespace phlex::experimental {
       using type = decltype(send(std::declval<T const&>()));
     };
 
-    template <move_constructible_only T>
+    template <sendable_by_move T>
     struct sendable_type_impl<T> {
       using type = T;
     };

phlex/model/flush_gate.cpp

@@ -1,19 +1,14 @@
 #include "phlex/model/flush_gate.hpp"
 
-#include "spdlog/spdlog.h"
-
 #include <cassert>
 #include <functional>
-#include <mutex>
 #include <ranges>
 #include <utility>
 
 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<data_cell_counts>()},
-    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,

phlex/model/flush_gate.hpp

@@ -40,6 +40,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <utility>
 
 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<data_cell_counts> committed_counts_;
+    data_cell_counts committed_counts_;
     // Accumulated expected child counts from all unfolds.
     data_cell_counts expected_counts_;
     std::atomic<std::size_t> received_flush_count_{0};

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 <memory>
 #include <ranges>
+#include <vector>
 
 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<data_cell_counts>();
-  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<data_cell_counts>();
-  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());