progress on decimation method

Author: frs69wq

Brainstorm_and_TODOs.md

@@ -89,6 +89,12 @@
 
    - [ ] new `ParameterizedDecimation` class
       - members:
+        - [x] `std::vector<size_t> stride_`
+        - [x] `std::string interpolation_method_` (default to empty string)
+        - [x] `double cost_per_element_`
+        - [x] `std::vector<size_t> reduced_shape_`
+        - [x] `std::unordered_map<sg4::ActorPtr, std::pair<std::vector<size_t>, std::vector<size_t>>> reduced_local_start_and_count_`
+        - [x] `size_t element_size_`
       - methods and behavior:
 
 

include/dtlmod/CompressionReductionMethod.hpp

@@ -66,6 +66,7 @@ class CompressionReductionMethod : public ReductionMethod {
     throw std::runtime_error("not implemented");
     // return;// std::make_pair(std::vector<size_t>(), std::vector<size_t>());
   }
+  [[nodiscard]] double get_flop_amount_to_reduce_variable(std::shared_ptr<Variable> var) const override { return 0.0; }
 };
 /// \endcond
 } // namespace dtlmod

include/dtlmod/DTLException.hpp

@@ -48,14 +48,16 @@ DECLARE_DTLMOD_EXCEPTION(UnknownOpenModeException, "Unknown open mode. Should be
 
 DECLARE_DTLMOD_EXCEPTION(UnknownVariableException, "Unknown Variable");
 DECLARE_DTLMOD_EXCEPTION(MultipleVariableDefinitionException, "Multiple Variable Definition");
-DECLARE_DTLMOD_EXCEPTION(InconsistentVariableDefinitionException,
-                         "Insconsistent Variable Definition. The 'shape', 'start', and 'count' vectors must "
-                         "have the same size");
+DECLARE_DTLMOD_EXCEPTION(InconsistentVariableDefinitionException, "Insconsistent Variable Definition");
 DECLARE_DTLMOD_EXCEPTION(IncorrectPathDefinitionException, "Fullpath must be structured as follows: "
                                                            "netzone_name:file_system_name:/path/to/file_name");
 
 DECLARE_DTLMOD_EXCEPTION(GetWhenNoTransactionException, "Impossible to get. No transaction exists for variable");
 
+DECLARE_DTLMOD_EXCEPTION(InconsistentDecimationStrideException, "Insconsistent Decimation Stride definition");
+DECLARE_DTLMOD_EXCEPTION(UnknownDecimationOptionException, "Unknown Decimation option");
+DECLARE_DTLMOD_EXCEPTION(UnknownDecimationInterpolationException, "Unknown Decimation interpolation method");
+
 } // namespace dtlmod
 
 #endif // DTLMOD_EXCEPTION_HPP

include/dtlmod/DecimationReductionMethod.hpp

@@ -9,7 +9,9 @@
 #include <boost/algorithm/string.hpp>
 #include <boost/algorithm/string/split.hpp>
 #include <cmath>
+#include <numeric> // for std::accumulate
 
+#include "dtlmod/DTLException.hpp"
 #include "dtlmod/ReductionMethod.hpp"
 #include "dtlmod/Variable.hpp"
 
@@ -21,17 +23,16 @@ namespace dtlmod {
 
 class ParameterizedDecimation {
   friend class DecimationReductionMethod;
+  std::shared_ptr<Variable> var_; // The variable to which this parameterized decimation is applied
+
   std::vector<size_t> stride_;
   std::string interpolation_method_ = "";
   double cost_per_element_;
 
   std::vector<size_t> reduced_shape_;
   std::unordered_map<sg4::ActorPtr, std::pair<std::vector<size_t>, std::vector<size_t>>> reduced_local_start_and_count_;
-  size_t element_size_;
 
 protected:
-  [[nodiscard]] const std::vector<size_t>& get_stride() const { return stride_; }
-
   void set_reduced_shape(const std::vector<size_t>& reduced_shape) { reduced_shape_ = reduced_shape; }
   void set_reduced_local_start_and_count(
       std::unordered_map<sg4::ActorPtr, std::pair<std::vector<size_t>, std::vector<size_t>>>
@@ -40,36 +41,48 @@ class ParameterizedDecimation {
     reduced_local_start_and_count_ = reduced_local_start_and_count;
   }
 
+  [[nodiscard]] const std::vector<size_t>& get_stride() const { return stride_; }
+
+  [[nodiscard]] const std::vector<size_t>& get_reduced_shape() const { return reduced_shape_; }
+
   [[nodiscard]] size_t get_global_reduced_size() const
   {
-    size_t total_size = element_size_;
-    for (const auto& s : reduced_shape_)
-      total_size *= s;
-    return total_size;
+    return std::accumulate(reduced_shape_.begin(), reduced_shape_.end(), var_->get_element_size(), std::multiplies<>{});
   }
 
   [[nodiscard]] size_t get_local_reduced_size() const
   {
-    size_t total_size = element_size_;
-    auto issuer       = sg4::Actor::self();
-    for (const auto& c : reduced_local_start_and_count_.at(issuer).second)
-      total_size *= c;
-    return total_size;
+    auto start_and_count = reduced_local_start_and_count_.at(sg4::Actor::self()).second;
+    return std::accumulate(start_and_count.begin(), start_and_count.end(), var_->get_element_size(),
+                           std::multiplies<>{});
   }
-  [[nodiscard]] const std::vector<size_t>& get_reduced_shape() const { return reduced_shape_; }
+
   [[nodiscard]] const std::pair<std::vector<size_t>, std::vector<size_t>>&
   get_reduced_start_and_count_for(sg4::ActorPtr publisher) const
   {
     return reduced_local_start_and_count_.at(publisher);
   }
 
+  [[nodiscard]] double get_flop_amount_to_decimate() const
+  {
+    double amount = cost_per_element_;
+    if (interpolation_method_.empty()) {
+      amount *= var_->get_local_size();
+    } else if (interpolation_method_ == "linear") {
+      amount = 2 * amount * var_->get_local_size();
+    } else if (interpolation_method_ == "quadratic") {
+      amount = 4 * amount * var_->get_local_size();
+    } else if (interpolation_method_ == "cubic") {
+      amount = 8 * amount * var_->get_local_size();
+    } else
+      throw UnknownDecimationInterpolationException(XBT_THROW_POINT, interpolation_method_.c_str());
+    return amount;
+  }
+
 public:
-  ParameterizedDecimation(const std::vector<size_t> stride, const std::string interpolation_method,
-                          double cost_per_element, size_t element_size)
-      : stride_(stride)
-      , interpolation_method_(interpolation_method)
-      , cost_per_element_(cost_per_element)
-      , element_size_(element_size)
+  ParameterizedDecimation(std::shared_ptr<Variable> var, const std::vector<size_t> stride,
+                          const std::string interpolation_method, double cost_per_element)
+      : var_(var), stride_(stride), interpolation_method_(interpolation_method), cost_per_element_(cost_per_element)
   {
   }
 };
@@ -89,31 +102,34 @@ class DecimationReductionMethod : public ReductionMethod {
       if (key == "stride") {
         std::vector<std::string> tokens;
         boost::split(tokens, value, boost::is_any_of(","), boost::token_compress_on);
-        // TODO Add Sanity check that the number of tokens equals the number of dimension of var
-        for (const auto t : tokens)
+        if (var->get_shape().size() != tokens.size())
+          throw InconsistentDecimationStrideException(
+              XBT_THROW_POINT, "Decimation Stride and Variable Shape vectors must have the same size. Stride: " +
+                                   std::to_string(tokens.size()) +
+                                   ", Shape: " + std::to_string(var->get_shape().size()));
+        for (const auto& t : tokens)
           stride.push_back(std::stoul(t));
       } else if (key == "interpolation") {
         interpolation_method = value;
       } else if (key == "cost_per_element") {
         cost_per_element = std::stod(value);
-      } // else
-        // TODO handle invalid key
+      } else
+        throw UnknownDecimationOptionException(XBT_THROW_POINT, key.c_str());
     }
 
     per_variable_parameterizations_.try_emplace(
-        var, std::make_shared<ParameterizedDecimation>(stride, interpolation_method, cost_per_element,
-                                                       var->get_element_size()));
+        var, std::make_shared<ParameterizedDecimation>(var, stride, interpolation_method, cost_per_element));
   }
 
   void reduce_variable(std::shared_ptr<Variable> var)
   {
     auto parameterization = per_variable_parameterizations_[var];
-    auto shape            = var->get_shape();
+    auto original_shape   = var->get_shape();
     auto stride           = parameterization->get_stride();
 
     std::vector<size_t> reduced_shape;
     size_t i = 0;
-    for (auto dim_size : shape)
+    for (auto dim_size : original_shape)
       reduced_shape.push_back(std::ceil(dim_size / (stride[i++] * 1.0)));
 
     std::unordered_map<sg4::ActorPtr, std::pair<std::vector<size_t>, std::vector<size_t>>>
@@ -123,11 +139,11 @@ class DecimationReductionMethod : public ReductionMethod {
       std::vector<size_t> reduced_start;
       std::vector<size_t> reduced_count;
 
-      for (size_t i = 0; i < shape.size(); i++) {
+      for (size_t i = 0; i < original_shape.size(); i++) {
         // Sanity checks that shape, start, and count have the same size have already been done
         size_t r_start = std::ceil(start[i] / (stride[i] * 1.0));
         size_t r_next_start =
-            std::min(shape[i], static_cast<size_t>(std::ceil((start[i] + count[i]) / (stride[i] * 1.0))));
+            std::min(original_shape[i], static_cast<size_t>(std::ceil((start[i] + count[i]) / (stride[i] * 1.0))));
         XBT_CDEBUG(dtlmod, "Dim %lu: stride = %lu, Start = %lu, r_start = %lu, Count = %lu, r_count = %lu", i,
                    stride[i], start[i], r_start, count[i], r_next_start - r_start);
         reduced_start.push_back(r_start);
@@ -149,10 +165,17 @@ class DecimationReductionMethod : public ReductionMethod {
   {
     return per_variable_parameterizations_.at(var)->get_local_reduced_size();
   }
+
+  [[nodiscard]] double get_flop_amount_to_reduce_variable(std::shared_ptr<Variable> var) const override
+  {
+    return per_variable_parameterizations_.at(var)->get_flop_amount_to_decimate();
+  }
+
   [[nodiscard]] const std::vector<size_t>& get_reduced_variable_shape(std::shared_ptr<Variable> var) const override
   {
     return per_variable_parameterizations_.at(var)->get_reduced_shape();
   }
+
   [[nodiscard]] const std::pair<std::vector<size_t>, std::vector<size_t>>&
   get_reduced_start_and_count_for(std::shared_ptr<Variable> var, sg4::ActorPtr publisher) const override
   {

include/dtlmod/ReductionMethod.hpp

@@ -35,6 +35,8 @@ class ReductionMethod {
   virtual const std::vector<size_t>& get_reduced_variable_shape(std::shared_ptr<Variable> var) const = 0;
   virtual const std::pair<std::vector<size_t>, std::vector<size_t>>&
   get_reduced_start_and_count_for(std::shared_ptr<Variable> var, simgrid::s4u::ActorPtr publisher) const = 0;
+  virtual double get_flop_amount_to_reduce_variable(std::shared_ptr<Variable> var) const                 = 0;
+
   /// @brief Helper function to print out the name of the ReductionMethod.
   /// @return The corresponding string
   [[nodiscard]] const std::string& get_name() const { return name_; }

src/Engine.cpp

@@ -38,8 +38,12 @@ void Engine::begin_transaction()
 void Engine::put(std::shared_ptr<Variable> var) const
 {
   if (var->is_reduced()) {
-    XBT_INFO("Put of a reduced version of %s (initial size = %lu, reduced size = %lu)", var->get_cname(),
-             var->get_local_size(), var->get_reduction_method()->get_reduced_variable_local_size(var));
+    // Perform a Exec activity before putting the variable into the DTL to account for the time needed to reduce it.
+    sg4::this_actor::execute(var->get_reduction_method()->get_flop_amount_to_reduce_variable(var));
+    XBT_DEBUG("Variable %s has been reduced!", var->get_cname());
+    // Now put the reduced version of the variable into the DTL, i.e., using its reduced local size.
+    XBT_DEBUG("Put this reduced version of %s (initial size = %lu, reduced size = %lu)", var->get_cname(),
+              var->get_local_size(), var->get_reduction_method()->get_reduced_variable_local_size(var));
     transport_->put(var, var->get_reduction_method()->get_reduced_variable_local_size(var));
   } else
     transport_->put(var, var->get_local_size());

src/FileEngine.cpp

@@ -115,7 +115,8 @@ void FileEngine::end_pub_transaction()
   XBT_DEBUG("Start the %d publish activities for the transaction", file_pub_transaction_[self].size());
   for (const auto& [file, size] : to_write) {
     auto write = file->write_async(size, true);
-    write->on_this_completion_cb([this, self, write](sg4::Io const&) {
+    write->on_this_completion_cb([this, self, write, size](sg4::Io const&) {
+      XBT_DEBUG("%llu bytes have been written for Actor %s", size, self->get_cname());
       pub_activities_completed_->notify_all();
       file_pub_transaction_[self].erase(write);
     });

test/dtl_reduction.cpp

@@ -32,8 +32,8 @@ class DTLReductionTest : public ::testing::Test {
   void setup_platform()
   {
     auto* zone = sg4::Engine::get_instance()->get_netzone_root()->add_netzone_empty("zone");
-    host_      = zone->add_host("host", "1Gf");
-    disk_      = host_->add_disk("disk", "5.5GBps", "2.1GBps");
+    host_      = zone->add_host("host", "6Gf");
+    disk_      = host_->add_disk("disk", "560MBps", "510MBps");
     zone->seal();
 
     auto my_fs = sgfs::FileSystem::create("my_fs");
@@ -49,6 +49,7 @@ class DTLReductionTest : public ::testing::Test {
 TEST_F(DTLReductionTest, SimpleDecimationFileEngine)
 {
   DO_TEST_WITH_FORK([this]() {
+    xbt_log_control_set("dtlmod_engine.thresh:debug");
     this->setup_platform();
     host_->add_actor("TestActor", [this]() {
       std::shared_ptr<dtlmod::ReductionMethod> decimator;      
@@ -64,10 +65,6 @@ TEST_F(DTLReductionTest, SimpleDecimationFileEngine)
           stream->define_variable("var3D", {640, 640, 640}, {0, 0, 0}, {640, 640, 640}, sizeof(double));
       XBT_INFO("Define a Decimation Reduction Method");
       ASSERT_NO_THROW(decimator = stream->define_reduction_method("decimation"));
-      XBT_INFO("Assign the decimation method to 'var3D'");
-      ASSERT_NO_THROW(var->set_reduction_operation(decimator, {{"stride", "1,2,4"}}));
-      XBT_INFO("Check that the variable is marked as 'reduced'");
-      ASSERT_TRUE(var->is_reduced());
       XBT_INFO("Open the stream in Pulish mode");
       auto engine = stream->open("zone:my_fs:/host/scratch/my-working-dir/my-output", dtlmod::Stream::Mode::Publish);
       ASSERT_NO_THROW(sg4::this_actor::sleep_for(1));
@@ -77,6 +74,18 @@ TEST_F(DTLReductionTest, SimpleDecimationFileEngine)
       ASSERT_NO_THROW(engine->put(var));
       XBT_INFO("End a Transaction");
       engine->end_transaction();
+      XBT_INFO("Sleep until t = 6s");
+      sg4::this_actor::sleep_until(6);
+      XBT_INFO("Assign the decimation method to 'var3D'");
+      ASSERT_NO_THROW(var->set_reduction_operation(decimator, {{"stride", "1,2,4"}}));
+      XBT_INFO("Check that the variable is marked as 'reduced'");
+      ASSERT_TRUE(var->is_reduced());
+      XBT_INFO("Start a Transaction");
+      engine->begin_transaction();
+      XBT_INFO("Put reduced Variable 'var' into the DTL");
+      ASSERT_NO_THROW(engine->put(var));
+      XBT_INFO("End a Transaction");
+      engine->end_transaction();
       XBT_INFO("Close the engine");
       engine->close();