Memory efficiency: progressive eviction of metadata transaction entries

Author: frs69wq

ChangeLog

@@ -1,5 +1,24 @@
 ----------------------------------------------------------------------------
 
+DTLMod (0.5) not released yet (target: May 2026)
+
+Improvements:
+  - Memory efficiency: progressive eviction of metadata transaction entries
+    - Once all subscribers have consumed a transaction, its entries are evicted
+      from the in-memory Metadata::transaction_infos_ map
+    - When metadata export is enabled and publishers and subscribers coexist
+      (file streaming), evicted entries are progressively flushed to 
+      per-variable temporary files; the final metadata file is assembled at 
+      pub_close() from those files and any remaining in-memory entries, 
+      preserving the existing format and transaction count
+    - When the stream is opened by subscribers only after all publishers have
+      closed (sequential scenario), memory-only eviction is performed (the
+      metadata file has already been written by pub_close())
+    - Memory footprint of the File engine now grows as O(N_pub) instead of
+      O(N_pub × N_transactions) for long-running concurrent streaming workloads
+
+----------------------------------------------------------------------------
+
 DTLMod (0.4) February 16, 2026
 
 Major improvements:

include/dtlmod/Engine.hpp

@@ -55,6 +55,8 @@ class Engine {
   std::shared_ptr<Transport> transport_ = nullptr;
   std::weak_ptr<Stream> stream_;
 
+  bool pub_ever_present_ = false;
+
   ActorRegistry publishers_;
   ActorRegistry subscribers_;
 
@@ -86,6 +88,8 @@ class Engine {
   [[nodiscard]] ActorRegistry& get_subscribers() noexcept { return subscribers_; }
   [[nodiscard]] const ActorRegistry& get_subscribers() const noexcept { return subscribers_; }
 
+  [[nodiscard]] bool pub_ever_present() const noexcept { return pub_ever_present_; }
+
   // Pure virtual methods for derived classes to implement
   virtual void create_transport(const Transport::Method& transport_method) = 0;
   virtual void begin_pub_transaction() = 0;

include/dtlmod/FileEngine.hpp

@@ -39,6 +39,7 @@ class FileEngine : public Engine {
 
   unsigned int current_sub_transaction_id_ = 0;
   bool sub_transaction_in_progress_        = false;
+  unsigned int subs_completed_current_tx_  = 0;
 
   void create_transport(const Transport::Method& transport_method) override;
   [[nodiscard]] const std::shared_ptr<sgfs::FileSystem>& get_file_system() const noexcept { return file_system_; }

include/dtlmod/Metadata.hpp

@@ -6,6 +6,8 @@
 #ifndef __DTLMOD_METADATA_HPP__
 #define __DTLMOD_METADATA_HPP__
 
+#include <fstream>
+
 #include <fsmod/File.hpp>
 #include <simgrid/s4u/Actor.hpp>
 
@@ -29,6 +31,8 @@ class Metadata {
            std::less<>>
       transaction_infos_;
 
+  unsigned int flushed_count_ = 0; // number of transactions already flushed to the prog file
+
 protected:
   const std::map<std::pair<std::vector<size_t>, std::vector<size_t>>, std::pair<std::string, sg4::ActorPtr>,
                  std::less<>>&
@@ -45,7 +49,12 @@ class Metadata {
   {
     return transaction_infos_.empty() ? 0 : (transaction_infos_.rbegin())->first;
   }
-  void export_to_file(std::ofstream& ostream) const;
+  // Write entries for tx_id to out, increment flushed_count_, erase from transaction_infos_
+  void write_transaction_to_stream(unsigned int tx_id, std::ofstream& out);
+  // Remove tx_id from transaction_infos_ without writing to file
+  void evict_transaction(unsigned int tx_id);
+  // Write all remaining transactions; prog_file_path contains already-flushed entries to prepend
+  void export_to_file(std::ofstream& ostream, const std::string& prog_file_path = "") const;
 };
 /// \endcond
 

include/dtlmod/Stream.hpp

@@ -47,6 +47,8 @@ class Stream : public std::enable_shared_from_this<Stream> {
   Transport::Method transport_method_ = Transport::Method::Undefined;
   bool metadata_export_               = false;
   std::string metadata_file_;
+  std::unordered_map<std::string, std::string> var_prog_file_paths_; // variable name -> prog file path
+  bool metadata_exported_ = false; // true once export_metadata_to_file() has been called
   sg4::MutexPtr mutex_ = sg4::Mutex::create();
   Mode access_mode_    = Mode::Publish;
 
@@ -63,7 +65,8 @@ class Stream : public std::enable_shared_from_this<Stream> {
   }
   void close() noexcept { engine_ = nullptr; }
 
-  void export_metadata_to_file() const;
+  void export_metadata_to_file();
+  void flush_and_evict_transaction(unsigned int tx_id);
 
   // Helper methods for Stream::open
   void validate_open_parameters(std::string_view name, Mode mode) const;

src/Engine.cpp

@@ -93,6 +93,7 @@ void Engine::close()
 /// \cond EXCLUDE_FROM_DOCUMENTATION
 void Engine::add_publisher(sg4::ActorPtr actor)
 {
+  pub_ever_present_ = true;
   transport_->add_publisher(publishers_.count());
   publishers_.add(actor);
 }

src/FileEngine.cpp

@@ -224,6 +224,22 @@ void FileEngine::end_sub_transaction()
   if (auto sub_barrier = get_subscribers().get_or_create_barrier())
     XBT_DEBUG("Barrier created for %zu subscribers", get_subscribers().count());
 
+  // Evict this transaction's metadata once all subscribers have completed their reads.
+  // Only applies in the concurrent streaming scenario (pub was registered on this same engine).
+  if (pub_ever_present()) {
+    unsigned int tx_to_evict = 0;
+    {
+      std::unique_lock lock(*get_subscribers().get_mutex());
+      if (++subs_completed_current_tx_ == get_subscribers().count()) {
+        subs_completed_current_tx_ = 0;
+        tx_to_evict                = current_sub_transaction_id_;
+      }
+    }
+    if (tx_to_evict > 0)
+      if (auto s = get_stream())
+        s->flush_and_evict_transaction(tx_to_evict);
+  }
+
   // Mark this transaction as over
   sub_transaction_in_progress_ = false;
 }

src/Metadata.cpp

@@ -18,38 +18,69 @@ void Metadata::add_transaction(unsigned int id,
   transaction_infos_[id][start_and_count] = std::make_pair(location, publisher);
 }
 
-void Metadata::export_to_file(std::ofstream& ostream) const
+static void write_block_entries(std::ofstream& ostream,
+                                const std::map<std::pair<std::vector<size_t>, std::vector<size_t>>,
+                                               std::pair<std::string, sg4::ActorPtr>, std::less<>>& transaction)
+{
+  for (const auto& [block_info, location] : transaction) {
+    const auto& [block_start, block_count] = block_info;
+    const auto& [where, actor]             = location;
+
+    ostream << "    " << where.c_str() << ": [";
+    XBT_DEBUG("    Actor %s wrote:", actor->get_cname());
+    unsigned long last = block_start.size() - 1;
+    for (unsigned long i = 0; i < last; i++) {
+      ostream << block_start[i] << ":" << block_start[i] + block_count[i] << ", ";
+      XBT_DEBUG("      Dimension %lu : [%zu..%zu]", i + 1, block_start[i], block_start[i] + block_count[i]);
+    }
+    ostream << block_start[last] << ":" << block_start[last] + block_count[last] << "]" << std::endl;
+    XBT_DEBUG("      Dimension %lu : [%zu..%zu]", last + 1, block_start[last], block_start[last] + block_count[last]);
+    XBT_DEBUG("    in: %s", where.c_str());
+  }
+}
+
+void Metadata::write_transaction_to_stream(unsigned int tx_id, std::ofstream& out)
+{
+  auto it = transaction_infos_.find(tx_id);
+  if (it == transaction_infos_.end())
+    return;
+  XBT_DEBUG("  Transaction %u:", tx_id);
+  out << "  Transaction " << tx_id << ":" << std::endl;
+  write_block_entries(out, it->second);
+  flushed_count_++;
+  transaction_infos_.erase(it);
+}
+
+void Metadata::evict_transaction(unsigned int tx_id)
+{
+  transaction_infos_.erase(tx_id);
+}
+
+void Metadata::export_to_file(std::ofstream& ostream, const std::string& prog_file_path) const
 {
   auto var = variable_.lock();
   xbt_assert(var, "Metadata::export_to_file called after its Variable has been destroyed");
   XBT_DEBUG("Variable %s:", var->get_cname());
-  ostream << var->get_element_size() << "\t" << var->get_cname() << "\t" << transaction_infos_.size();
+  unsigned int total = flushed_count_ + static_cast<unsigned int>(transaction_infos_.size());
+  ostream << var->get_element_size() << "\t" << var->get_cname() << "\t" << total;
   ostream << "*{";
   auto shape            = var->get_shape();
   const auto last_index = shape.size() - 1;
   for (unsigned int i = 0; i < last_index; i++)
     ostream << shape[i] << ",";
   ostream << shape[last_index] << "}" << std::endl;
 
+  // Copy already-flushed entries from the per-variable prog file (if any)
+  if (!prog_file_path.empty()) {
+    std::ifstream prog(prog_file_path, std::ios::binary);
+    ostream << prog.rdbuf();
+  }
+
+  // Write remaining in-memory entries
   for (const auto& [id, transaction] : transaction_infos_) {
     XBT_DEBUG("  Transaction %u:", id);
     ostream << "  Transaction " << id << ":" << std::endl;
-    for (const auto& [block_info, location] : transaction) {
-      const auto& [block_start, block_count] = block_info;
-      const auto& [where, actor]             = location;
-
-      ostream << "    " << where.c_str() << ": [";
-      XBT_DEBUG("    Actor %s wrote:", actor->get_cname());
-      unsigned long last = block_start.size() - 1;
-      for (unsigned long i = 0; i < last; i++) {
-        ostream << block_start[i] << ":" << block_start[i] + block_count[i] << ", ";
-        XBT_DEBUG("      Dimension %lu : [%zu..%zu]", i + 1, block_start[i], block_start[i] + block_count[i]);
-      }
-      ostream << block_start[last] << ":" << block_start[last] + block_count[last] << "]" << std::endl;
-      XBT_DEBUG("      Dimension %lu : [%zu..%zu]", last + 1, block_start[last], block_start[last] + block_count[last]);
-
-      XBT_DEBUG("    in: %s", where.c_str());
-    }
+    write_block_entries(ostream, transaction);
   }
 }
 /// \endcond

src/Stream.cpp

@@ -146,13 +146,43 @@ Stream& Stream::unset_metadata_export() noexcept
   return *this;
 }
 
-void Stream::export_metadata_to_file() const
+void Stream::export_metadata_to_file()
 {
+  metadata_exported_ = true;
   if (metadata_export_) {
     std::ofstream export_stream(metadata_file_, std::ofstream::out);
-    for (const auto& [name, v] : variables_)
-      v->get_metadata()->export_to_file(export_stream);
+    for (const auto& [name, v] : variables_) {
+      auto it          = var_prog_file_paths_.find(name);
+      std::string prog = (it != var_prog_file_paths_.end()) ? it->second : "";
+      v->get_metadata()->export_to_file(export_stream, prog);
+    }
     export_stream.close();
+    // Remove per-variable prog files now that the final file is written
+    for (const auto& [name, path] : var_prog_file_paths_)
+      std::remove(path.c_str());
+    var_prog_file_paths_.clear();
+  }
+}
+
+void Stream::flush_and_evict_transaction(unsigned int tx_id)
+{
+  if (metadata_exported_) {
+    // Sequential scenario or post-export: the metadata file is already complete; just free memory
+    for (const auto& [name, v] : variables_)
+      v->get_metadata()->evict_transaction(tx_id);
+    return;
+  }
+  if (metadata_export_) {
+    for (const auto& [name, v] : variables_) {
+      if (!var_prog_file_paths_.count(name))
+        var_prog_file_paths_[name] = metadata_file_ + "." + name + ".prog";
+      std::ofstream out(var_prog_file_paths_.at(name), std::ofstream::app);
+      v->get_metadata()->write_transaction_to_stream(tx_id, out);
+      // write_transaction_to_stream also increments flushed_count_ and evicts from transaction_infos_
+    }
+  } else {
+    for (const auto& [name, v] : variables_)
+      v->get_metadata()->evict_transaction(tx_id);
   }
 }