revise how cancel_transaction works

Author: frs69wq

ChangeLog

@@ -20,12 +20,12 @@ Improvements:
 
 API Changes:
   - New Engine method:
-    - Engine::cancel_transaction() cancel the currently ongoing transaction
-      performed by this engine. This method has to be called by an external
-      actor not involved in the transaction. It will make the transaction raise
-      a DTLMod::TransactionCancelledException that must be caught if you want
-      the publishers and subscribers to survive to the cancellation of this
-      transaction.
+    - Engine::cancel_transaction(unsigned int transaction_id) cancels a
+      specific transaction performed by this engine. This method has to be
+      called by an external actor not involved in the transaction. It will make
+      the transaction raise a DTLMod::TransactionCanceledException that must be
+      caught if you want the publishers and subscribers to survive to the
+      cancelation of this transaction.
   - New Stream helper methods:
     - Stream::get_engine_type() and Stream::get_transport_method() respectively
       return the enum value of the engine type and transport method for a 

include/dtlmod/DTLException.hpp

@@ -70,7 +70,7 @@ DECLARE_DTLMOD_EXCEPTION(UnknownCompressionOptionException, "Unknown Compression
 DECLARE_DTLMOD_EXCEPTION(InconsistentCompressionRatioException, "Inconsistent Compression ratio");
 DECLARE_DTLMOD_EXCEPTION(SubscriberSideCompressionException, "Compression can only be applied on the publisher side");
 
-DECLARE_DTLMOD_EXCEPTION(TransactionCancelledException, "Transaction cancelled");
+DECLARE_DTLMOD_EXCEPTION(TransactioncanceledException, "Transaction canceled");
 
 } // namespace dtlmod
 

include/dtlmod/Engine.hpp

@@ -57,7 +57,7 @@ class Engine {
   std::weak_ptr<Stream> stream_;
 
   bool pub_ever_present_ = false;
-  std::atomic<bool> cancelled_{false};
+  std::atomic<unsigned int> canceled_transaction_id_{0};
 
   ActorRegistry publishers_;
 
@@ -77,8 +77,9 @@ class Engine {
   [[nodiscard]] const sg4::ActivitySet& get_sub_transaction() const noexcept { return sub_transaction_; }
   [[nodiscard]] sg4::ActivitySet& get_sub_transaction() noexcept { return sub_transaction_; }
 
-  // Protected virtual method for derived classes to implement
-  [[nodiscard]] virtual unsigned int get_current_transaction_impl() const noexcept = 0;
+  // Protected virtual methods for derived classes to implement
+  [[nodiscard]] virtual unsigned int get_current_transaction_impl() const noexcept     = 0;
+  [[nodiscard]] virtual unsigned int get_current_sub_transaction_impl() const noexcept = 0;
 
   // Protected methods for derived classes only
   void close_stream() const;
@@ -93,7 +94,11 @@ class Engine {
 
   [[nodiscard]] bool pub_ever_present() const noexcept { return pub_ever_present_; }
 
-  [[nodiscard]] bool is_cancelled() const noexcept { return cancelled_; }
+  [[nodiscard]] bool is_canceled() const noexcept { return canceled_transaction_id_ != 0; }
+  [[nodiscard]] bool is_transaction_canceled(unsigned int tx_id) const noexcept
+  {
+    return canceled_transaction_id_ == tx_id;
+  }
 
   // Pure virtual methods for derived classes to implement
   virtual void create_transport(const Transport::Method& transport_method) = 0;
@@ -144,9 +149,11 @@ class Engine {
   /// @return The id of the ongoing transaction.
   [[nodiscard]] unsigned int get_current_transaction() const noexcept { return get_current_transaction_impl(); }
 
-  /// @brief Cancel all in-flight activities of the current transaction, unblocking publishers and subscribers.
+  /// @brief Cancel all in-flight activities of a specific transaction, unblocking publishers and subscribers.
+  /// @param transaction_id The id of the transaction to cancel. If both sides have already moved past this
+  ///        transaction, the call is a no-op to avoid accidentally cancelling a subsequent transaction.
   /// @note Must be called from an external actor not participating in the transaction.
-  void cancel_transaction();
+  void cancel_transaction(unsigned int transaction_id);
 
   /// @brief Close the Engine associated to a Stream.
   void close();

include/dtlmod/FileEngine.hpp

@@ -55,6 +55,10 @@ class FileEngine : public Engine {
   {
     return current_pub_transaction_id_;
   }
+  [[nodiscard]] unsigned int get_current_sub_transaction_impl() const noexcept override
+  {
+    return current_sub_transaction_id_;
+  }
 
 protected:
   [[nodiscard]] std::shared_ptr<FileTransport> get_file_transport() const;

include/dtlmod/StagingEngine.hpp

@@ -44,6 +44,10 @@ class StagingEngine : public Engine {
   {
     return current_pub_transaction_id_;
   }
+  [[nodiscard]] unsigned int get_current_sub_transaction_impl() const noexcept override
+  {
+    return current_sub_transaction_id_;
+  }
 
 protected:
   [[nodiscard]] std::shared_ptr<StagingTransport> get_staging_transport() const;

src/Engine.cpp

@@ -86,9 +86,15 @@ void Engine::close()
   publishers_.contains(sg4::Actor::self()) ? pub_close() : sub_close();
 }
 
-void Engine::cancel_transaction()
+void Engine::cancel_transaction(unsigned int transaction_id)
 {
-  cancelled_ = true;
+  // No-op if both sides have already moved past the target transaction: cancelling now would
+  // affect the wrong (subsequent) transaction.
+  if (get_current_transaction_impl() > transaction_id && get_current_sub_transaction_impl() > transaction_id)
+    return;
+  // transaction_id == 0 means no transaction has started yet; treat as T1 so all checks against
+  // (current + 1 >= 1) still fire correctly.
+  canceled_transaction_id_.store(transaction_id == 0 ? 1 : transaction_id);
   cancel_activities();
 }
 

src/FileEngine.cpp

@@ -100,8 +100,8 @@ std::string FileEngine::get_path_to_dataset() const
 
 void FileEngine::begin_pub_transaction()
 {
-  if (is_cancelled())
-    throw TransactionCancelledException(XBT_THROW_POINT);
+  if (is_transaction_canceled(current_pub_transaction_id_ + 1))
+    throw TransactioncanceledException(XBT_THROW_POINT);
 
   auto self = sg4::Actor::self();
 
@@ -115,12 +115,12 @@ void FileEngine::begin_pub_transaction()
     // Wait for the completion of the Publish activities from the previous transaction
     XBT_DEBUG("Wait for the completion of %u publish activities from the previous transaction",
               file_pub_transaction_[self].size());
-    while (!is_cancelled() && file_pub_transaction_[self].size() > 0) {
+    while (!is_canceled() && file_pub_transaction_[self].size() > 0) {
       std::unique_lock lock(*(get_publishers().get_mutex()));
       pub_activities_completed_->wait(lock);
     }
-    if (is_cancelled())
-      throw TransactionCancelledException(XBT_THROW_POINT);
+    if (is_transaction_canceled(current_pub_transaction_id_))
+      throw TransactioncanceledException(XBT_THROW_POINT);
     XBT_DEBUG("All on-flight publish activities are completed. Proceed with the current transaction.");
     get_file_transport()->clear_to_write_in_transaction(self);
   }
@@ -169,7 +169,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 (!is_cancelled() && file_pub_transaction_[self].size() > 0) {
+  while (!is_transaction_canceled(current_pub_transaction_id_) && file_pub_transaction_[self].size() > 0) {
     std::unique_lock lock(*(get_publishers().get_mutex()));
     pub_activities_completed_->wait(lock);
   }
@@ -191,8 +191,8 @@ void FileEngine::pub_close()
 
 void FileEngine::begin_sub_transaction()
 {
-  if (is_cancelled())
-    throw TransactionCancelledException(XBT_THROW_POINT);
+  if (is_transaction_canceled(current_sub_transaction_id_ + 1))
+    throw TransactioncanceledException(XBT_THROW_POINT);
 
   // Only one subscriber has to do this
   if (!sub_transaction_in_progress_) {
@@ -204,12 +204,15 @@ void FileEngine::begin_sub_transaction()
   // We have publishers on that stream, wait for them to complete a transaction first
   if (not get_publishers().is_empty()) {
     std::unique_lock lock(*get_subscribers().get_mutex());
-    while (!is_cancelled() && completed_pub_transaction_id_ < current_sub_transaction_id_) {
+    while (!is_transaction_canceled(current_sub_transaction_id_) &&
+           completed_pub_transaction_id_ < current_sub_transaction_id_) {
       XBT_DEBUG("Wait for publishers to end the transaction I need");
       pub_transaction_completed_->wait(lock);
     }
-    if (is_cancelled())
-      throw TransactionCancelledException(XBT_THROW_POINT);
+    if (is_transaction_canceled(current_sub_transaction_id_)) {
+      sub_transaction_in_progress_ = false;
+      throw TransactioncanceledException(XBT_THROW_POINT);
+    }
     XBT_DEBUG("Publishers stored metadata for that transaction, proceed");
   }
 }
@@ -221,17 +224,17 @@ void FileEngine::end_sub_transaction()
 
   // The files subscribers need to read may not have been fully written. Wait to be notified completion of the publish
   // activities
-  if (!is_cancelled() && current_sub_transaction_id_ == current_pub_transaction_id_ &&
-      not get_publishers().is_empty()) {
+  if (!is_transaction_canceled(current_sub_transaction_id_) &&
+      current_sub_transaction_id_ == current_pub_transaction_id_ && not get_publishers().is_empty()) {
     XBT_DEBUG("Wait for the completion of publish activities from the current transaction");
     pub_activities_completed_->wait(std::unique_lock(*get_subscribers().get_mutex()));
     XBT_DEBUG("All on-flight publish activities are completed. Proceed with the subscribe activities.");
   }
-  if (is_cancelled()) {
+  if (is_transaction_canceled(current_sub_transaction_id_)) {
     transport->close_sub_files(self);
     transport->clear_to_read_in_transaction(self);
     sub_transaction_in_progress_ = false;
-    throw TransactionCancelledException(XBT_THROW_POINT);
+    throw TransactioncanceledException(XBT_THROW_POINT);
   }
 
   // Subscriber get the list of files and size to read that has been build during the get() operations
@@ -243,12 +246,12 @@ void FileEngine::end_sub_transaction()
 
   XBT_DEBUG("Wait for the %d subscribe activities for the transaction", file_sub_transaction_[self].size());
   file_sub_transaction_[self].wait_all();
-  if (is_cancelled()) {
+  if (is_transaction_canceled(current_sub_transaction_id_)) {
     file_sub_transaction_[self].clear();
     transport->close_sub_files(self);
     transport->clear_to_read_in_transaction(self);
     sub_transaction_in_progress_ = false;
-    throw TransactionCancelledException(XBT_THROW_POINT);
+    throw TransactioncanceledException(XBT_THROW_POINT);
   }
   file_sub_transaction_[self].clear();
   // Close files opened in this transaction

src/StagingEngine.cpp

@@ -55,8 +55,8 @@ std::shared_ptr<StagingTransport> StagingEngine::get_staging_transport() const
 
 void StagingEngine::begin_pub_transaction()
 {
-  if (is_cancelled())
-    throw TransactionCancelledException(XBT_THROW_POINT);
+  if (is_transaction_canceled(current_pub_transaction_id_ + 1))
+    throw TransactioncanceledException(XBT_THROW_POINT);
 
   if (!pub_transaction_in_progress_) {
     pub_transaction_in_progress_ = true;
@@ -78,24 +78,24 @@ void StagingEngine::begin_pub_transaction()
     try {
       get_pub_transaction().wait_all();
     } catch (const simgrid::NetworkFailureException&) {
-      if (!is_cancelled())
+      if (!is_canceled())
         throw;
     }
     XBT_DEBUG("All on-flight publish activities are completed. Proceed with the current transaction.");
     XBT_DEBUG("%u sub activities pending", get_sub_transaction().size());
     get_pub_transaction().clear();
-    if (is_cancelled())
-      throw TransactionCancelledException(XBT_THROW_POINT);
+    if (is_transaction_canceled(current_pub_transaction_id_))
+      throw TransactioncanceledException(XBT_THROW_POINT);
   }
 
   // Then we wait for all subscribers to be at the same transaction
-  while (!is_cancelled() &&
+  while (!is_transaction_canceled(current_pub_transaction_id_) &&
          (get_subscribers().is_empty() || current_pub_transaction_id_ > current_sub_transaction_id_)) {
     XBT_DEBUG("Wait for subscribers");
     sub_transaction_started_->wait(lock);
   }
-  if (is_cancelled())
-    throw TransactionCancelledException(XBT_THROW_POINT);
+  if (is_transaction_canceled(current_pub_transaction_id_))
+    throw TransactioncanceledException(XBT_THROW_POINT);
   // Publisher has been notified by subscribers, it can proceed with the transaction
 }
 
@@ -146,16 +146,16 @@ void StagingEngine::pub_close()
 
 void StagingEngine::begin_sub_transaction()
 {
-  if (is_cancelled())
-    throw TransactionCancelledException(XBT_THROW_POINT);
+  if (is_transaction_canceled(current_sub_transaction_id_ + 1))
+    throw TransactioncanceledException(XBT_THROW_POINT);
 
   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(*get_subscribers().get_mutex());
-    while (!is_cancelled() && current_pub_transaction_id_ == 0)
+    while (!is_transaction_canceled(current_sub_transaction_id_ + 1) && current_pub_transaction_id_ == 0)
       first_pub_transaction_started_->wait(lock);
-    if (is_cancelled())
-      throw TransactionCancelledException(XBT_THROW_POINT);
+    if (is_transaction_canceled(current_sub_transaction_id_ + 1))
+      throw TransactioncanceledException(XBT_THROW_POINT);
     XBT_DEBUG("Publishers have started a transaction, create rendez-vous points");
     // We now know the number of publishers, subscriber can create mailboxes/mqs with publishers
     get_staging_transport()->create_rendez_vous_points();
@@ -178,10 +178,13 @@ void StagingEngine::begin_sub_transaction()
   }
 
   std::unique_lock lock(*get_subscribers().get_mutex());
-  while (!is_cancelled() && completed_pub_transaction_id_ < current_sub_transaction_id_)
+  while (!is_transaction_canceled(current_sub_transaction_id_) &&
+         completed_pub_transaction_id_ < current_sub_transaction_id_)
     pub_transaction_completed_->wait(lock);
-  if (is_cancelled())
-    throw TransactionCancelledException(XBT_THROW_POINT);
+  if (is_transaction_canceled(current_sub_transaction_id_)) {
+    sub_transaction_in_progress_ = false;
+    throw TransactioncanceledException(XBT_THROW_POINT);
+  }
 }
 
 void StagingEngine::end_sub_transaction()
@@ -195,19 +198,19 @@ void StagingEngine::end_sub_transaction()
     try {
       get_sub_transaction().wait_all();
     } catch (const simgrid::CancelException&) {
-      if (!is_cancelled())
+      if (!is_canceled())
         throw;
       get_sub_transaction().clear();
       sub_transaction_in_progress_ = false;
       num_subscribers_starting_--;
-      throw TransactionCancelledException(XBT_THROW_POINT);
+      throw TransactioncanceledException(XBT_THROW_POINT);
     } catch (const simgrid::NetworkFailureException&) {
-      if (!is_cancelled())
+      if (!is_canceled())
         throw;
       get_sub_transaction().clear();
       sub_transaction_in_progress_ = false;
       num_subscribers_starting_--;
-      throw TransactionCancelledException(XBT_THROW_POINT);
+      throw TransactioncanceledException(XBT_THROW_POINT);
     }
     XBT_DEBUG("All on-flight subscribe activities are completed. Proceed with the current transaction.");
     get_sub_transaction().clear();

src/bindings/python/dtlmod_python.cpp

@@ -88,7 +88,7 @@ PYBIND11_MODULE(dtlmod, m)
   py::register_exception<dtlmod::InconsistentCompressionRatioException>(m, "InconsistentCompressionRatioException");
   py::register_exception<dtlmod::SubscriberSideCompressionException>(m, "SubscriberSideCompressionException");
 
-  py::register_exception<dtlmod::TransactionCancelledException>(m, "TransactionCancelledException");
+  py::register_exception<dtlmod::TransactioncanceledException>(m, "TransactioncanceledException");
 
   /* Class Engine */
   py::class_<Engine, std::shared_ptr<Engine>> engine(
@@ -108,7 +108,8 @@ PYBIND11_MODULE(dtlmod, m)
       .def_property_readonly("current_transaction", &Engine::get_current_transaction,
                              "The id of the current transaction on this Engine (read-only)")
       .def("cancel_transaction", &Engine::cancel_transaction, py::call_guard<py::gil_scoped_release>(),
-           "Cancel all in-flight activities of the current transaction (must be called from an external actor)")
+           py::arg("transaction_id"),
+           "Cancel all in-flight activities of a specific transaction (must be called from an external actor)")
       .def("close", &Engine::close, py::call_guard<py::gil_scoped_release>(), "Close this Engine");
 
   py::enum_<Engine::Type>(engine, "Type", "The type of Engine")