register subscriber variable in CompressionReductionMethod during inquire

Author: frs69wq

include/dtlmod/CompressionReductionMethod.hpp

@@ -79,6 +79,10 @@ class CompressionReductionMethod : public ReductionMethod {
 
   [[nodiscard]] double get_flop_amount_to_reduce_variable(const Variable& var) const override;
   [[nodiscard]] double get_flop_amount_to_decompress_variable(const Variable& var) const override;
+
+  /// @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);
 };
 /// \endcond
 } // namespace dtlmod

src/CompressionReductionMethod.cpp

@@ -139,4 +139,13 @@ void CompressionReductionMethod::parameterize_for_variable(
   // Always (re)create the parameterization — avoids field-by-field update complexity.
   per_variable_parameterizations_[&var] = std::make_shared<ParameterizedCompression>(var, std::move(cfg));
 }
+
+void CompressionReductionMethod::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())
+    per_variable_parameterizations_[&subscriber_var] =
+        std::make_shared<ParameterizedCompression>(subscriber_var, it->second->get_config());
+}
+
 } // namespace dtlmod

src/Stream.cpp

@@ -410,6 +410,10 @@ 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);
     }
 
     return new_var;

test/dtl_reduction.cpp

@@ -571,8 +571,14 @@ TEST_F(DTLReductionTest, CompressionStagingEngine)
       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 compression reduction operation");
+      sg4::this_actor::sleep_for(1);
       auto var    = stream->inquire_variable("var");
 
+      XBT_INFO("Verify that the subscriber variable carries the publisher compression state");
+      ASSERT_TRUE(var->is_reduced());
+      ASSERT_TRUE(var->is_reduced_by_publisher());
+
       XBT_INFO("Get the compressed variable (decompression cost should be applied on subscriber)");
       engine->begin_transaction();
       ASSERT_NO_THROW(engine->get(var));