Propagate reduction method to subscribers for all reduction methods + test

Author: frs69wq

include/dtlmod/CompressionReductionMethod.hpp

@@ -82,7 +82,7 @@ class CompressionReductionMethod : public ReductionMethod {
 
   /// @brief Copy a publisher variable's parameterization to a subscriber variable.
   ///        Called by Stream::inquire_variable so that Engine::get() can compute decompression costs.
-  void propagate_for_subscriber(const Variable& publisher_var, const Variable& subscriber_var);
+  void propagate_for_subscriber(const Variable& publisher_var, const Variable& subscriber_var) override;
 };
 /// \endcond
 } // namespace dtlmod

include/dtlmod/DecimationReductionMethod.hpp

@@ -98,6 +98,10 @@ class DecimationReductionMethod : public ReductionMethod {
 
 public:
   using ReductionMethod::ReductionMethod;
+
+  /// @brief Copy a publisher variable's parameterization to a subscriber variable.
+  ///        Called by Stream::inquire_variable so that subscribers can query reduced sizes.
+  void propagate_for_subscriber(const Variable& publisher_var, const Variable& subscriber_var) override;
 };
 ///\endcond
 } // namespace dtlmod

include/dtlmod/ReductionMethod.hpp

@@ -38,6 +38,7 @@ class ReductionMethod {
   get_reduced_start_and_count_for(const Variable& var, simgrid::s4u::ActorPtr publisher) const = 0;
   virtual double get_flop_amount_to_reduce_variable(const Variable& var) const                 = 0;
   virtual double get_flop_amount_to_decompress_variable(const Variable& /*var*/) const { return 0.0; }
+  virtual void propagate_for_subscriber(const Variable& /*publisher_var*/, const Variable& /*subscriber_var*/) {}
 
   /// @brief Helper function to print out the name of the ReductionMethod.
   /// @return The corresponding string

src/DecimationReductionMethod.cpp

@@ -123,6 +123,26 @@ void DecimationReductionMethod::parameterize_for_variable(
   }
 }
 
+void DecimationReductionMethod::propagate_for_subscriber(const Variable& publisher_var, const Variable& subscriber_var)
+{
+  auto it = per_variable_parameterizations_.find(&publisher_var);
+  if (it == per_variable_parameterizations_.end())
+    return;
+
+  auto pub_param = it->second;
+  auto sub_param = std::make_shared<ParameterizedDecimation>(subscriber_var, pub_param->get_stride(),
+                                                             pub_param->get_interpolation_method(),
+                                                             pub_param->get_cost_per_element());
+  sub_param->set_reduced_shape(pub_param->get_reduced_shape());
+
+  // The subscriber receives the full reduced variable, so its local region is the entire reduced shape.
+  const auto& reduced_shape = pub_param->get_reduced_shape();
+  std::vector<size_t> reduced_start(reduced_shape.size(), 0);
+  sub_param->set_reduced_local_start_and_count(sg4::Actor::self(), reduced_start, reduced_shape);
+
+  per_variable_parameterizations_[&subscriber_var] = std::move(sub_param);
+}
+
 void DecimationReductionMethod::reduce_variable(const Variable& var)
 {
   auto parameterization = per_variable_parameterizations_[&var];

src/Stream.cpp

@@ -410,10 +410,9 @@ std::shared_ptr<Variable> Stream::inquire_variable(std::string_view name) const
     if (var->second->is_reduced()) {
       new_var->is_reduced_with_  = var->second->get_reduction_method();
       new_var->reduction_origin_ = var->second->reduction_origin_;
-      // Register the subscriber's variable in the compression method's map so that
-      // Engine::get() can compute decompression costs via get_flop_amount_to_decompress_variable.
-      if (auto compressor = std::dynamic_pointer_cast<CompressionReductionMethod>(new_var->is_reduced_with_))
-        compressor->propagate_for_subscriber(*var->second, *new_var);
+      // Register the subscriber's variable in the reduction method's map so that
+      // Engine::get() can compute costs, and subscribers can query reduced sizes.
+      new_var->is_reduced_with_->propagate_for_subscriber(*var->second, *new_var);
     }
 
     return new_var;

test/dtl_reduction.cpp

@@ -501,9 +501,21 @@ TEST_F(DTLReductionTest, DecimationStagingEngine)
       auto dtl    = dtlmod::DTL::connect();
       auto stream = dtl->add_stream("my-output");
       auto engine = stream->open("my-output", dtlmod::Stream::Mode::Subscribe);
+      XBT_INFO("Wait for the publisher to have set the decimation reduction operation");
+      sg4::this_actor::sleep_for(0.5);
       auto var    = stream->inquire_variable("var");
 
+      ASSERT_TRUE(var->is_reduced());
+      ASSERT_TRUE(var->is_reduced_by_publisher());
+
+      XBT_INFO("Verify that the subscriber can access the reduction method set by the publisher");
+      auto reduction = var->get_reduction_method();
+      ASSERT_TRUE(reduction != nullptr);
+      XBT_INFO("Verify that the subscriber can get the reduced local size");
+      auto reduced_size = reduction->get_reduced_variable_local_size(*var);
+      ASSERT_DOUBLE_EQ(reduced_size, 5000 * 5000 * 8.0);
       XBT_INFO("Get the decimated variable");
+
       engine->begin_transaction();
       ASSERT_NO_THROW(engine->get(var));
       engine->end_transaction();