fix

Author: frs69wq

src/StagingEngine.cpp

@@ -127,17 +127,18 @@ void StagingEngine::begin_sub_transaction()
   if (not sub_transaction_in_progress_) {
     current_sub_transaction_id_++;
     sub_transaction_in_progress_ = true;
-    XBT_DEBUG("Subscribe Transaction %u started by %s", current_sub_transaction_id_, sg4::Actor::self()->get_cname());
   }
 
   num_subscribers_starting_++;
+  XBT_DEBUG("Subscribe Transaction %u started by %s (%d/%lu)", current_sub_transaction_id_, sg4::Actor::self()->get_cname(),
+            num_subscribers_starting_,get_num_subscribers());
+
 
   // The last subscriber to start a transaction notifies the publishers
   if (num_subscribers_starting_ == get_num_subscribers() &&
       current_pub_transaction_id_ == current_sub_transaction_id_) {
     XBT_DEBUG("Notify Publishers that they can start their transaction");
     sub_transaction_started_->notify_all();
-    num_subscribers_starting_ = 0;
   }
 
   std::unique_lock<sg4::Mutex> lock(*sub_mutex_);
@@ -159,13 +160,17 @@ void StagingEngine::end_sub_transaction()
     sub_transaction_.wait_all();
     XBT_DEBUG("All on-flight subscribe activities are completed. Proceed with the current transaction.");
     sub_transaction_.clear();
-    // Mark this transaction as over
-    sub_transaction_in_progress_ = false;
   }
 
   // Prevent subscribers to start a new transaction before this one is really over
-  if (sub_barrier_)
-    sub_barrier_->wait();
+  if (sub_barrier_->wait())
+  // Mark this transaction as over
+  sub_transaction_in_progress_ = false;
+  // Decrease counter for next iteration
+  num_subscribers_starting_--;
+   XBT_DEBUG("Subscribe Transaction %u end by %s (%d/%lu)", current_sub_transaction_id_, sg4::Actor::self()->get_cname(),
+            num_subscribers_starting_,get_num_subscribers());
+
 }
 
 void StagingEngine::sub_close()