tfbuilder: reset run counters

Author: ironMann

src/TfBuilder/TfBuilderDevice.cxx

@@ -285,6 +285,12 @@ void TfBuilderDevice::PreRun()
   // enable monitoring
   DataDistMonitor::enable_datadist(DataDistLogger::sRunNumber, mPartitionId);
 
+  // reset tf counters and start accepting tfs
+  // NOTE: TfBuilder accepts TFs in both READY and RUNNING state.
+  mFlpInputHandler->reset();
+  mRpc->reset_run_counters();
+  mRpc->startAcceptingTfs();
+
   IDDLOG("Entering running state. RunNumber: {}", DataDistLogger::sRunNumberStr);
 
   mTfFwdTotalDataSize = 0;
@@ -298,6 +304,12 @@ void TfBuilderDevice::PreRun()
 // Get here when ConditionalRun returns false
 void TfBuilderDevice::PostRun()
 {
+  // stop accepting tfs until the state is clear in-between runs
+  mRpc->stopAcceptingTfs();
+
+  // send EOS
+  mShouldSendEos = true;
+
   // Not in Running state
   mInRunningState = false;
   DDMON("tfbuilder", "running", 0);
@@ -310,6 +322,9 @@ void TfBuilderDevice::PostRun()
   // disable monitoring
   DataDistMonitor::disable_datadist();
 
+  // start accepting tfs for the next run
+  mRpc->startAcceptingTfs();
+
   IDDLOG("Exiting running state. RunNumber: {}", DataDistLogger::sRunNumberStr);
 }
 
@@ -334,8 +349,6 @@ void TfBuilderDevice::TfForwardThread()
 {
   using hres_clock = std::chrono::high_resolution_clock;
   auto lRateStartTime = hres_clock::now();
-  std::uint64_t lTfOutCnt = 0;
-  bool lShouldSendEos = false;
 
   while (mRunning) {
     std::optional<std::unique_ptr<SubTimeFrame>> lTfOpt = dequeue_for(eTfFwdIn, 100ms);
@@ -350,16 +363,14 @@ void TfBuilderDevice::TfForwardThread()
         WDDLOG_RL(1000, "Dropping a raw TimeFrame because stop of the run is requested.");
       }
 
-      if (dplEnabled() && lShouldSendEos) {
-        lShouldSendEos = false;
+      // send EOS if exiting the running state
+      if (dplEnabled() && mShouldSendEos) {
         mTfDplAdapter->sendEosToDpl();
+        mShouldSendEos = false;
       }
       continue;
     }
 
-    // Make sure to send EoS if in running state
-    lShouldSendEos = true;
-
     if (lTfOpt == std::nullopt) {
       DDMON("tfbuilder", "data_output.rate", 0);
       DDMON("tfbuilder", "tf_output.sent_size", mTfFwdTotalDataSize);
@@ -386,9 +397,8 @@ void TfBuilderDevice::TfForwardThread()
 
     if (!mStandalone) {
       try {
-        lTfOutCnt++;
         IDDLOG_RL(5000, "Forwarding a new TF to DPL. tf_id={} stf_size={:d} unique_equipments={:d} total={:d}",
-          lTfId, lTf->getDataSize(), lTf->getEquipmentIdentifiers().size(), lTfOutCnt);
+          lTfId, lTf->getDataSize(), lTf->getEquipmentIdentifiers().size(), mTfFwdTotalTfCount);
 
         if (dplEnabled()) {
           // adapt headers to include DPL processing header on the stack
@@ -413,8 +423,9 @@ void TfBuilderDevice::TfForwardThread()
   }
 
   // leaving the output thread, send end of the stream info
-  if (dplEnabled() && lShouldSendEos) {
+  if (dplEnabled() && mShouldSendEos) {
     mTfDplAdapter->sendEosToDpl();
+    mShouldSendEos = false;
   }
 
   DDDLOG("Exiting TF forwarding thread.");

src/TfBuilder/TfBuilderDevice.h

@@ -159,6 +159,7 @@ class TfBuilderDevice : public DataDistDevice,
   std::atomic_bool mRunning = false;         // Task initialized
   std::atomic_bool mInRunningState = false;  // FMQ in running state
   std::atomic_bool mShouldExit = false;
+  std::atomic_bool mShouldSendEos = false; // toggle in post run
 };
 
 } /* namespace o2::DataDistribution */

src/TfBuilder/TfBuilderInput.h

@@ -46,6 +46,13 @@ class TfBuilderInput
 
   bool start(std::shared_ptr<ConsulTfBuilder> pConfig);
   void stop(std::shared_ptr<ConsulTfBuilder> pConfig);
+  void reset() {
+    mReceivedData.flush();
+    mStfsForMerging.flush();
+    std::unique_lock<std::mutex> lQueueLock(mStfMergerQueueLock);
+    mStfMergeMap.clear();
+    mMaxMergedTfId = 0;
+  }
 
   void DataHandlerThread(const std::uint32_t pFlpIndex, const std::string pStfSenderId);
   void StfPacingThread();

src/TfBuilder/TfBuilderRpc.cxx

@@ -99,9 +99,7 @@ void TfBuilderRpcImpl::stop()
   mStfSenderRpcClients.stop();
   mTfSchedulerRpcClient.stop();
 
-  mCurrentTfBufferSize = 0;
-  mNumBufferedTfs = 0;
-  mLastBuiltTfId = 0;
+  reset_run_counters();
 }
 
 // make sure these are sent immediately
@@ -223,7 +221,7 @@ bool TfBuilderRpcImpl::recordTfBuilt(const SubTimeFrame &pTf)
   }
   mUpdateCondition.notify_one();
 
-  { // record the current TP
+  { // record the current TF
     std::unique_lock lLock(mStfDurationMapLock);
     mStfReqDuration.erase(lTfId);
   }

src/TfBuilder/TfBuilderRpc.h

@@ -109,6 +109,16 @@ class TfBuilderRpcImpl final : public TfBuilderRpc::Service
 
   ::grpc::Status TerminatePartition(::grpc::ServerContext* context, const ::o2::DataDistribution::PartitionInfo* request, ::o2::DataDistribution::PartitionResponse* response) override;
 
+  /// reset counters on each new run
+  void reset_run_counters() {
+    mTfIdSizes.clear();
+    mCurrentTfBufferSize = mBufferSize;
+    mLastBuiltTfId = 0;
+    mNumBufferedTfs = 0;
+    mNumTfsInBuilding = 0;
+    mTfBuildRequests->flush();
+  }
+
 private:
   std::atomic_bool mRunning = false;
   std::atomic_bool mTerminateRequested = false;