use weak_ptr and store stream in variables

Author: frs69wq

include/dtlmod/Engine.hpp

@@ -52,7 +52,7 @@ class Engine {
   std::string name_;
   Type type_                            = Type::Undefined;
   std::shared_ptr<Transport> transport_ = nullptr;
-  Stream* stream_;
+  std::weak_ptr<Stream> stream_;
 
   sg4::MutexPtr pub_mutex_ = sg4::Mutex::create();
   std::set<sg4::ActorPtr> publishers_;
@@ -109,7 +109,10 @@ class Engine {
 
 public:
   /// \cond EXCLUDE_FROM_DOCUMENTATION
-  explicit Engine(const std::string& name, Stream* stream, Type type) : name_(name), type_(type), stream_(stream) {}
+  explicit Engine(const std::string& name, std::shared_ptr<Stream> stream, Type type)
+      : name_(name), type_(type), stream_(stream)
+  {
+  }
   virtual ~Engine() = default;
   /// \endcond
 

include/dtlmod/FileEngine.hpp

@@ -47,7 +47,7 @@ class FileEngine : public Engine {
   void sub_close() override;
 
 public:
-  explicit FileEngine(const std::string& fullpath, Stream* stream);
+  explicit FileEngine(const std::string& fullpath, const std::shared_ptr<Stream>& stream);
 };
 /// \endcond
 

include/dtlmod/StagingEngine.hpp

@@ -29,7 +29,10 @@ class StagingEngine : public Engine {
   void sub_close() override;
 
 public:
-  explicit StagingEngine(const std::string& name, Stream* stream) : Engine(name, stream, Engine::Type::Staging) {}
+  explicit StagingEngine(const std::string& name, const std::shared_ptr<Stream>& stream)
+      : Engine(name, stream, Engine::Type::Staging)
+  {
+  }
 
   void create_transport(const Transport::Method& transport_method) override;
 };

include/dtlmod/Stream.hpp

@@ -24,7 +24,7 @@ class DTL;
  *         objects to be injected into or retrieve from this Stream.
  */
 
-class Stream {
+class Stream : public std::enable_shared_from_this<Stream> {
   friend class Engine;
   friend class FileEngine;
   friend class StagingEngine;

include/dtlmod/Variable.hpp

@@ -14,6 +14,8 @@
 
 namespace dtlmod {
 
+class Stream;
+
 /// @brief A class to translate a piece of data from an application into an object handled by the DTL and its metadata.
 class Variable {
   friend class Engine;
@@ -29,6 +31,8 @@ class Variable {
   int transaction_start_          = -1;
   unsigned int transaction_count_ = 0;
 
+  std::weak_ptr<const Stream> defined_in_stream_;
+
   std::shared_ptr<Metadata> metadata_;
 
   std::map<sg4::ActorPtr, std::pair<std::vector<size_t>, std::vector<size_t>>, std::less<>> subscriber_selections_;
@@ -65,9 +69,11 @@ class Variable {
 
 public:
   /// \cond EXCLUDE_FROM_DOCUMENTATION
-  Variable(const std::string& name, size_t element_size, const std::vector<size_t>& shape)
+  Variable(const std::string& name, size_t element_size, const std::vector<size_t>& shape,
+           std::shared_ptr<const Stream> stream)
       : name_(name), element_size_(element_size), shape_(shape), metadata_(std::make_shared<Metadata>(this))
   {
+    defined_in_stream_ = stream;
   }
   /// \endcond
 

src/Engine.cpp

@@ -88,7 +88,7 @@ void Engine::add_subscriber(sg4::ActorPtr actor)
 void Engine::export_metadata_to_file() const
 {
   std::ofstream metadata_export(metadata_file_, std::ofstream::out);
-  for (const auto& [name, v] : stream_->get_all_variables_internal())
+  for (const auto& [name, v] : stream_.lock()->get_all_variables_internal())
     v->get_metadata()->export_to_file(metadata_export);
   metadata_export.close();
 }
@@ -100,7 +100,7 @@ void Engine::set_metadata_file_name()
 
 void Engine::close_stream()
 {
-  stream_->close();
+  stream_.lock()->close();
 }
 /// \endcond
 

src/FileEngine.cpp

@@ -24,7 +24,8 @@ namespace dtlmod {
 
 // FileEngines require to know where to (virtually) write file. This information is given by fullpath which has the
 // following format: NetZone:FileSystem:PathToDirectory
-FileEngine::FileEngine(const std::string& fullpath, Stream* stream) : Engine(fullpath, stream, Engine::Type::File)
+FileEngine::FileEngine(const std::string& fullpath, const std::shared_ptr<Stream>& stream)
+    : Engine(fullpath, stream, Engine::Type::File)
 {
   XBT_DEBUG("Create a new FileEngine writing in %s", fullpath.c_str());
 
@@ -155,7 +156,7 @@ void FileEngine::pub_close()
     XBT_DEBUG("Closing opened files");
     transport->close_pub_files();
     XBT_DEBUG("Engine '%s' is now closed for all publishers ", get_cname());
-    if (stream_->does_export_metadata())
+    if (stream_.lock()->does_export_metadata())
       export_metadata_to_file();
   }
 }

src/StagingEngine.cpp

@@ -108,7 +108,7 @@ void StagingEngine::pub_close()
     XBT_DEBUG("All publishers have called the Engine::close() function");
     close_stream();
     XBT_DEBUG("Engine '%s' is now closed for all publishers ", get_cname());
-    if (stream_->does_export_metadata())
+    if (stream_.lock()->does_export_metadata())
       export_metadata_to_file();
   }
 }

src/Stream.cpp

@@ -140,11 +140,11 @@ std::shared_ptr<Engine> Stream::open(const std::string& name, Mode mode)
   dtl_->lock();
   if (not engine_) {
     if (engine_type_ == Engine::Type::Staging) {
-      engine_ = std::make_shared<StagingEngine>(name, this);
+      engine_ = std::make_shared<StagingEngine>(name, shared_from_this());
       engine_->create_transport(transport_method_);
     } else if (engine_type_ == Engine::Type::File) {
       try {
-        engine_ = std::make_shared<FileEngine>(name, this);
+        engine_ = std::make_shared<FileEngine>(name, shared_from_this());
         engine_->create_transport(transport_method_);
       } catch (const IncorrectPathDefinitionException& e) {
         got_exception = true;
@@ -222,7 +222,7 @@ std::shared_ptr<Variable> Stream::define_variable(const std::string& name, const
       return var->second;
     }
   } else {
-    auto new_var = std::make_shared<Variable>(name, element_size, shape);
+    auto new_var = std::make_shared<Variable>(name, element_size, shape, shared_from_this());
     new_var->set_local_start_and_count(publisher, std::make_pair(start, count));
     variables_.try_emplace(name, new_var);
     return new_var;
@@ -248,7 +248,8 @@ std::shared_ptr<Variable> Stream::inquire_variable(const std::string& name) cons
   if (not engine_ || engine_->is_publisher(actor))
     return var->second;
   else {
-    auto new_var = std::make_shared<Variable>(name, var->second->get_element_size(), var->second->get_shape());
+    auto new_var =
+        std::make_shared<Variable>(name, var->second->get_element_size(), var->second->get_shape(), shared_from_this());
     new_var->set_local_start_and_count(actor, std::make_pair(std::vector<size_t>(var->second->get_shape().size(), 0),
                                                              std::vector<size_t>(var->second->get_shape().size(), 0)));
     new_var->set_metadata(var->second->get_metadata());