refactor Engine.hpp by adding ActorRegistry class

Author: frs69wq

CMakeLists.txt

@@ -152,6 +152,7 @@ set(SOURCE_FILES
 )
 
 set(HEADER_FILES
+  include/dtlmod/ActorRegistry.hpp
   include/dtlmod/DTL.hpp
   include/dtlmod/DTLException.hpp
   include/dtlmod/Engine.hpp

include/dtlmod/ActorRegistry.hpp

@@ -0,0 +1,53 @@
+/* Copyright (c) 2026. The SWAT Team. All rights reserved.          */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef __DTLMOD_ACTOR_REGISTRY_HPP__
+#define __DTLMOD_ACTOR_REGISTRY_HPP__
+
+#include <simgrid/s4u/Actor.hpp>
+#include <simgrid/s4u/Barrier.hpp>
+#include <simgrid/s4u/Mutex.hpp>
+
+#include <set>
+
+namespace sg4 = simgrid::s4u;
+
+namespace dtlmod {
+
+/// \cond EXCLUDE_FROM_DOCUMENTATION
+/// @brief A class that manages a registry of actors (publishers or subscribers) for an Engine.
+/// This class encapsulates actor management logic including addition, removal, lookup,
+/// and synchronization via barriers.
+
+class ActorRegistry {
+  friend class Engine;
+  sg4::MutexPtr mutex_ = sg4::Mutex::create();
+  std::set<sg4::ActorPtr> actors_;
+  sg4::BarrierPtr barrier_ = nullptr;
+
+public:
+  ActorRegistry() = default;
+
+  void add(sg4::ActorPtr actor) { actors_.insert(actor); }
+
+  void remove(sg4::ActorPtr actor) noexcept { actors_.erase(actor); }
+
+  [[nodiscard]] bool contains(sg4::ActorPtr actor) const noexcept { return actors_.find(actor) != actors_.end(); }
+
+  [[nodiscard]] size_t count() const noexcept { return actors_.size(); }
+  [[nodiscard]] const std::set<sg4::ActorPtr>& get_all() const noexcept { return actors_; }
+  [[nodiscard]] bool is_empty() const noexcept { return actors_.empty(); }
+  [[nodiscard]] sg4::BarrierPtr get_or_create_barrier()
+  {
+    if (!barrier_)
+      barrier_ = sg4::Barrier::create(actors_.size());
+    return barrier_;
+  }
+  [[nodiscard]] bool is_last_at_barrier() { return barrier_ && barrier_->wait(); }
+  [[nodiscard]] sg4::MutexPtr get_mutex() noexcept { return mutex_; };
+};
+
+} // namespace dtlmod
+#endif

include/dtlmod/Engine.hpp

@@ -18,6 +18,7 @@
 
 #include <string>
 
+#include "dtlmod/ActorRegistry.hpp"
 #include "dtlmod/Transport.hpp"
 #include "dtlmod/Variable.hpp"
 
@@ -45,66 +46,52 @@ class Engine {
 
   friend class Stream;
 
-protected:
   std::string name_;
   Type type_                            = Type::Undefined;
   std::shared_ptr<Transport> transport_ = nullptr;
   std::weak_ptr<Stream> stream_;
 
-  sg4::MutexPtr pub_mutex_ = sg4::Mutex::create();
-  std::set<sg4::ActorPtr> publishers_;
+  ActorRegistry publishers_;
+  ActorRegistry subscribers_;
+  std::string metadata_file_;
+
+protected:
   sg4::ActivitySet pub_transaction_;
   sg4::ConditionVariablePtr pub_transaction_completed_ = sg4::ConditionVariable::create();
-  sg4::BarrierPtr pub_barrier_                         = nullptr;
   unsigned int current_pub_transaction_id_             = 0;
   unsigned int completed_pub_transaction_id_           = 0;
   bool pub_transaction_in_progress_                    = false;
-  virtual void begin_pub_transaction()                 = 0;
-  virtual void end_pub_transaction()                   = 0;
-  virtual void pub_close()                             = 0;
 
-  sg4::MutexPtr sub_mutex_ = sg4::Mutex::create();
-  std::set<sg4::ActorPtr> subscribers_;
   sg4::ActivitySet sub_transaction_;
-
-  sg4::BarrierPtr sub_barrier_             = nullptr;
   unsigned int current_sub_transaction_id_ = 0;
   bool sub_transaction_in_progress_        = false;
-  virtual void begin_sub_transaction()     = 0;
-  virtual void end_sub_transaction()       = 0;
-  virtual void sub_close()                 = 0;
-
-  std::string metadata_file_;
-  void export_metadata_to_file() const;
 
-  /// \cond EXCLUDE_FROM_DOCUMENTATION
   void close_stream() const;
 
   virtual void create_transport(const Transport::Method& transport_method) = 0;
-
   void set_transport(std::shared_ptr<Transport> transport) noexcept { transport_ = transport; }
   [[nodiscard]] std::shared_ptr<Transport> get_transport() const noexcept { return transport_; }
 
   void add_publisher(sg4::ActorPtr actor);
-  void rm_publisher(sg4::ActorPtr actor) noexcept { publishers_.erase(actor); }
-  [[nodiscard]] bool is_publisher(sg4::ActorPtr actor) const noexcept
-  {
-    return publishers_.find(actor) != publishers_.end();
-  }
+  void rm_publisher(sg4::ActorPtr actor) noexcept { publishers_.remove(actor); }
+  [[nodiscard]] bool is_publisher(sg4::ActorPtr actor) const noexcept { return publishers_.contains(actor); }
   // Synchronize publishers on engine closing
-  [[nodiscard]] int is_last_publisher() const { return (pub_barrier_ && pub_barrier_->wait()); }
+  [[nodiscard]] bool is_last_publisher() { return publishers_.is_last_at_barrier(); }
+  virtual void begin_pub_transaction() = 0;
+  virtual void end_pub_transaction()   = 0;
+  virtual void pub_close()             = 0;
 
   void add_subscriber(sg4::ActorPtr actor);
-  void rm_subscriber(sg4::ActorPtr actor) noexcept { subscribers_.erase(actor); }
-  [[nodiscard]] bool is_subscriber(sg4::ActorPtr actor) const noexcept
-  {
-    return subscribers_.find(actor) != subscribers_.end();
-  }
+  void rm_subscriber(sg4::ActorPtr actor) noexcept { subscribers_.remove(actor); }
+  [[nodiscard]] bool is_subscriber(sg4::ActorPtr actor) const noexcept { return subscribers_.contains(actor); }
   // Synchronize subscribers on engine closing
-  [[nodiscard]] int is_last_subscriber() const noexcept { return subscribers_.empty(); }
+  [[nodiscard]] bool is_last_subscriber() const noexcept { return subscribers_.is_empty(); }
+  virtual void begin_sub_transaction() = 0;
+  virtual void end_sub_transaction()   = 0;
+  virtual void sub_close()             = 0;
 
   void set_metadata_file_name();
-  /// \endcond
+  void export_metadata_to_file() const;
 
 public:
   /// \cond EXCLUDE_FROM_DOCUMENTATION
@@ -118,9 +105,9 @@ class Engine {
   [[nodiscard]] sg4::ActivitySet& get_pub_transaction() noexcept { return pub_transaction_; }
   [[nodiscard]] const sg4::ActivitySet& get_sub_transaction() const noexcept { return sub_transaction_; }
   [[nodiscard]] sg4::ActivitySet& get_sub_transaction() noexcept { return sub_transaction_; }
-  [[nodiscard]] const std::set<sg4::ActorPtr>& get_publishers() const noexcept { return publishers_; }
-  [[nodiscard]] size_t get_num_publishers() const noexcept { return publishers_.size(); }
-  [[nodiscard]] size_t get_num_subscribers() const noexcept { return subscribers_.size(); }
+  [[nodiscard]] const std::set<sg4::ActorPtr>& get_publishers() const noexcept { return publishers_.get_all(); }
+  [[nodiscard]] size_t get_num_publishers() const noexcept { return publishers_.count(); }
+  [[nodiscard]] size_t get_num_subscribers() const noexcept { return subscribers_.count(); }
   /// \endcond
 
   /// @brief Helper function to print out the name of the Engine.

src/Engine.cpp

@@ -75,14 +75,14 @@ void Engine::close()
 /// \cond EXCLUDE_FROM_DOCUMENTATION
 void Engine::add_publisher(sg4::ActorPtr actor)
 {
-  transport_->add_publisher(publishers_.size());
-  publishers_.insert(actor);
+  transport_->add_publisher(publishers_.count());
+  publishers_.add(actor);
 }
 
 void Engine::add_subscriber(sg4::ActorPtr actor)
 {
-  transport_->add_subscriber(subscribers_.size());
-  subscribers_.insert(actor);
+  transport_->add_subscriber(subscribers_.count());
+  subscribers_.add(actor);
 }
 
 void Engine::export_metadata_to_file() const

src/FileEngine.cpp

@@ -92,7 +92,7 @@ void FileEngine::begin_pub_transaction()
     XBT_DEBUG("Wait for the completion of %u publish activities from the previous transaction",
               file_pub_transaction_[self].size());
     while (file_pub_transaction_[self].size() > 0)
-      pub_activities_completed_->wait(std::unique_lock(*pub_mutex_));
+      pub_activities_completed_->wait(std::unique_lock(*publishers_.get_mutex()));
     XBT_DEBUG("All on-flight publish activities are completed. Proceed with the current transaction.");
     transport->clear_to_write_in_transaction(self);
   }
@@ -104,9 +104,9 @@ void FileEngine::end_pub_transaction()
   auto transport = std::static_pointer_cast<FileTransport>(transport_);
 
   // This is the end of the first transaction, create a barrier
-  if (!pub_barrier_) {
-    XBT_DEBUG("Create a barrier for %zu publishers", publishers_.size());
-    pub_barrier_ = sg4::Barrier::create(publishers_.size());
+  auto pub_barrier = publishers_.get_or_create_barrier();
+  if (pub_barrier) {
+    XBT_DEBUG("Barrier created for %zu publishers", publishers_.count());
   }
 
   // Publisher gets the list of files and size to write that has been build during the put() operations
@@ -143,7 +143,7 @@ void FileEngine::pub_close()
   XBT_DEBUG("[%s] Wait for the completion of %u publish activities from the previous transaction", get_cname(),
             file_pub_transaction_[self].size());
   while (file_pub_transaction_[self].size() > 0)
-    pub_activities_completed_->wait(std::unique_lock(*pub_mutex_));
+    pub_activities_completed_->wait(std::unique_lock(*publishers_.get_mutex()));
   transport->clear_to_write_in_transaction(self);
 
   rm_publisher(self);
@@ -172,7 +172,7 @@ void FileEngine::begin_sub_transaction()
 
   // We have publishers on that stream, wait for them to complete a transaction first
   if (get_num_publishers() > 0) {
-    std::unique_lock lock(*sub_mutex_);
+    std::unique_lock lock(*subscribers_.get_mutex());
     while (completed_pub_transaction_id_ < current_sub_transaction_id_) {
       XBT_DEBUG("Wait for publishers to end the transaction I need");
       pub_transaction_completed_->wait(lock);
@@ -190,7 +190,7 @@ void FileEngine::end_sub_transaction()
   // activities
   if (current_sub_transaction_id_ == current_pub_transaction_id_ && get_num_publishers() > 0) {
     XBT_DEBUG("Wait for the completion of publish activities from the current transaction");
-    pub_activities_completed_->wait(std::unique_lock(*sub_mutex_));
+    pub_activities_completed_->wait(std::unique_lock(*subscribers_.get_mutex()));
     XBT_DEBUG("All on-flight publish activities are completed. Proceed with the subscribe activities.");
   }
 

src/StagingEngine.cpp

@@ -47,7 +47,7 @@ void StagingEngine::begin_pub_transaction()
   }
 
   // Only one publisher has to do this
-  std::unique_lock lock(*pub_mutex_);
+  std::unique_lock lock(*publishers_.get_mutex());
   if (current_pub_transaction_id_ > 1) { // This is not the first transaction.
     // Wait for the completion of the Publish activities from the previous transaction
     XBT_DEBUG("[T %d] (%d) Wait for the completion of %u publish activities from the previous transaction",
@@ -69,9 +69,9 @@ void StagingEngine::begin_pub_transaction()
 void StagingEngine::end_pub_transaction()
 {
   // This is the end of the first transaction, create a barrier
-  if (!pub_barrier_) {
-    XBT_DEBUG("Create a barrier for %zu publishers", publishers_.size());
-    pub_barrier_ = sg4::Barrier::create(publishers_.size());
+  auto pub_barrier = publishers_.get_or_create_barrier();
+  if (pub_barrier) {
+    XBT_DEBUG("Barrier created for %zu publishers", publishers_.count());
   }
 
   // A new pub transaction has been completed, notify subscribers that they can starting getting variables
@@ -117,7 +117,7 @@ void StagingEngine::begin_sub_transaction()
 {
   if (current_sub_transaction_id_ == 0) { // This is the first transaction
     // Wait for at least one publisher to start a tran
-    std::unique_lock lock(*sub_mutex_);
+    std::unique_lock lock(*subscribers_.get_mutex());
     while (current_pub_transaction_id_ == 0)
       first_pub_transaction_started_->wait(lock);
     XBT_DEBUG("Publishers have started a transaction, create rendez-vous points");
@@ -141,28 +141,28 @@ void StagingEngine::begin_sub_transaction()
     sub_transaction_started_->notify_all();
   }
 
-  std::unique_lock lock(*sub_mutex_);
+  std::unique_lock lock(*subscribers_.get_mutex());
   while (completed_pub_transaction_id_ < current_sub_transaction_id_)
     pub_transaction_completed_->wait(lock);
 }
 
 void StagingEngine::end_sub_transaction()
 {
   // This is the end of the first transaction, create a barrier
-  if (!sub_barrier_) {
-    XBT_DEBUG("Create a barrier for %zu subscribers", subscribers_.size());
-    sub_barrier_ = sg4::Barrier::create(subscribers_.size());
+  auto sub_barrier = subscribers_.get_or_create_barrier();
+  if (sub_barrier) {
+    XBT_DEBUG("Barrier created for %zu subscribers", subscribers_.count());
   }
 
-  if (sub_barrier_->wait()) {
+  if (subscribers_.is_last_at_barrier()) {
     XBT_DEBUG("Wait for the %d subscribe activities for the transaction", sub_transaction_.size());
     sub_transaction_.wait_all();
     XBT_DEBUG("All on-flight subscribe activities are completed. Proceed with the current transaction.");
     sub_transaction_.clear();
   }
 
   // Prevent subscribers to start a new transaction before this one is really over
-  if (sub_barrier_->wait())
+  if (subscribers_.is_last_at_barrier())
     // Mark this transaction as over
     sub_transaction_in_progress_ = false;
   // Decrease counter for next iteration