datadist: monitoring improvements; more accurate rates

	 + fix for env id

Author: ironMann

src/StfBuilder/StfBuilderDevice.cxx

@@ -98,10 +98,11 @@ void StfBuilderDevice::InitTask()
   I().mMaxBuiltStfs = GetConfig()->GetValue<std::uint64_t>(OptionKeyMaxBuiltStfs);
 
   // partition id
-  I().mPartitionId = Config::getPartitionOption(*GetConfig()).value_or(DataDistLogger::sPartitionIdStr);
-  if (DataDistLogger::sPartitionIdStr.empty() && !I().mPartitionId.empty()) {
-    DataDistLogger::sPartitionIdStr = I().mPartitionId;
-    impl::DataDistLoggerCtx::InitInfoLogger();
+  I().mPartitionId = Config::getPartitionOption(*GetConfig()).value_or("");
+  if (I().mPartitionId.empty()) {
+    EDDLOG("Partition id is not provided during InitTask(). Check command line or ECS parameters. Exiting.");
+    ChangeState(fair::mq::Transition::ErrorFound);
+    return;
   }
 
   // start monitoring
@@ -130,17 +131,18 @@ void StfBuilderDevice::InitTask()
   // check run type
   if (ReadoutDataUtils::sRunType == ReadoutDataUtils::RunType::eInvalid) {
     EDDLOG("Run type paramter must be correctly set.");
-    throw std::logic_error("Run type paramter must be correctly set.");
+    ChangeState(fair::mq::Transition::ErrorFound);
+    return;
   }
 
   // check run type
   if (ReadoutDataUtils::sRunType == ReadoutDataUtils::RunType::eTopology) {
     if (! ((ReadoutDataUtils::sSpecifiedDataOrigin == o2::header::gDataOriginITS) ||
       (ReadoutDataUtils::sSpecifiedDataOrigin == o2::header::gDataOriginMFT))) {
-
       EDDLOG("Run type paramter 'topology' is supported only for ITS and MFT. Please specify the detector option. detector={}",
         ReadoutDataUtils::sSpecifiedDataOrigin.as<std::string>());
-      throw std::logic_error("Run type paramter 'topology' is supported only for ITS and MFT. Please specify the detector option.");
+      ChangeState(fair::mq::Transition::ErrorFound);
+      return;
     }
   }
 
@@ -166,12 +168,14 @@ void StfBuilderDevice::InitTask()
 
   // File sink
   if (!I().mFileSink->loadVerifyConfig(*(this->GetConfig()))) {
-    std::this_thread::sleep_for(1s); exit(-1);
+    ChangeState(fair::mq::Transition::ErrorFound);
+    return;
   }
 
   // File source
   if (!I().mFileSource->loadVerifyConfig(*(this->GetConfig()))) {
-    std::this_thread::sleep_for(1s); exit(-1);
+    ChangeState(fair::mq::Transition::ErrorFound);
+    return;
   }
 
   // Discovery. Verify other parameters when running online and !standalone
@@ -185,17 +189,6 @@ void StfBuilderDevice::InitTask()
       lStatus.mutable_info()->set_type(StfBuilder);
       lStatus.mutable_info()->set_process_state(BasicInfo::NOT_RUNNING);
       lStatus.mutable_info()->set_process_id(Config::getIdOption(StfBuilder, *GetConfig(), lConsulRequired));
-
-      // wait for "partition-id"
-      while (!Config::getPartitionOption(*GetConfig())) {
-        WDDLOG_RL(1000, "StfBuilder waiting on 'discovery-partition' config parameter.");
-        std::this_thread::sleep_for(250ms);
-      }
-      if (I().mPartitionId.empty()) {
-        EDDLOG("StfBuilder 'discovery-partition' parameter not set.");
-        std::this_thread::sleep_for(1s); exit(-1);
-      }
-      I().mPartitionId = Config::getPartitionOption(*GetConfig()).value();
       lStatus.mutable_partition()->set_partition_id(I().mPartitionId);
       I().mDiscoveryConfig->write();
     }
@@ -209,14 +202,16 @@ void StfBuilderDevice::InitTask()
       (ReadoutDataUtils::sSpecifiedDataOrigin == o2::header::gDataOriginAny)) {
       EDDLOG("Detector string parameter must be specified when receiving the data from the "
         "readout and not using RDHv6 or greater.");
-      std::this_thread::sleep_for(1s); exit(-1);
+      ChangeState(fair::mq::Transition::ErrorFound);
+      return;
     } else {
       IDDLOG("READOUT INTERFACE: Configured detector: {}", ReadoutDataUtils::sSpecifiedDataOrigin.as<std::string>());
     }
 
     if (ReadoutDataUtils::sRdhVersion == ReadoutDataUtils::RdhVersion::eRdhInvalid) {
       EDDLOG("The RDH version must be specified when receiving data from readout.");
-      std::this_thread::sleep_for(1s); exit(-1);
+      ChangeState(fair::mq::Transition::ErrorFound);
+      return;
     } else {
       IDDLOG("READOUT INTERFACE: Configured RDHv{}", ReadoutDataUtils::sRdhVersion);
       RDHReader::Initialize(unsigned(ReadoutDataUtils::sRdhVersion));
@@ -254,7 +249,8 @@ void StfBuilderDevice::InitTask()
         GetChannel(I().mInputChannelName);
       } catch(std::exception &) {
         EDDLOG("Input channel not configured (from o2-readout-exe) and not running with file source enabled.");
-        std::this_thread::sleep_for(1s); exit(-1);
+        ChangeState(fair::mq::Transition::ErrorFound);
+        return;
       }
     }
 
@@ -264,7 +260,8 @@ void StfBuilderDevice::InitTask()
       }
     } catch(std::exception &e) {
       EDDLOG("Output channel (to DPL or StfSender) must be configured if not running in stand-alone mode.");
-      std::this_thread::sleep_for(1s); exit(-1);
+      ChangeState(fair::mq::Transition::ErrorFound);
+      return;
     }
   }
 
@@ -337,6 +334,8 @@ void StfBuilderDevice::StfOutputThread()
 
   decltype(hres_clock::now()) lStfStartTime = hres_clock::now();
 
+  DDMON_RATE("stfbuilder", "stf_output", 0.0);
+
   while (I().mState.mRunning) {
     // Get a STF ready for sending, or nullopt
     auto lStfOpt = I().dequeue_for(eStfSendIn, 100ms);
@@ -358,13 +357,12 @@ void StfBuilderDevice::StfOutputThread()
 
     lShouldSendEos = true;
     if (lStfOpt == std::nullopt) {
-      DDMON("stfbuilder", "stf_output.rate", 0);
-      DDMON("stfbuilder", "stf_output.size", 0);
       DDMON("stfbuilder", "data_output.rate", 0);
       continue;
     }
 
     auto &lStf = lStfOpt.value();
+    DDMON_RATE("stfbuilder", "stf_output", lStf->getDataSize());
 
     // decrement the stf counter
     I().mCounters.mNumStfs--;
@@ -379,8 +377,6 @@ void StfBuilderDevice::StfOutputThread()
       lStfStartTime = lNow;
 
       const auto lRate = 1.0 / std::max(lStfDur.count(), 0.00001);
-      DDMON("stfbuilder", "stf_output.rate", lRate);
-      DDMON("stfbuilder", "stf_output.size", lStf->getDataSize());
       DDMON("stfbuilder", "stf_output.id", lStf->id());
       DDMON("stfbuilder", "data_output.rate", (lRate * lStf->getDataSize()));
     }

src/StfBuilder/StfBuilderInput.cxx

@@ -231,10 +231,7 @@ void StfInputInterface::StfBuilderThread()
   };
 
   const auto cStfDataWaitFor = 2s;
-
-  using hres_clock = std::chrono::high_resolution_clock;
-
-  std::chrono::time_point<hres_clock, std::chrono::duration<double>> lStartSec = hres_clock::now();
+  DDMON_RATE("stfbuilder", "stf_input", 0.0);
 
   while (mRunning) {
 
@@ -251,13 +248,9 @@ void StfInputInterface::StfBuilderThread()
         }
 
         (*lStf)->setOrigin(SubTimeFrame::Header::Origin::eReadout);
+
+        DDMON_RATE("stfbuilder", "stf_input", (*lStf)->getDataSize());
         mSeqStfQueue.push(std::move(*lStf));
-        {
-          auto lNow = hres_clock::now();
-          std::chrono::duration<double> lTimeDiff = lNow - lStartSec;
-          lStartSec = lNow;
-          DDMON("stfbuilder", "stf_input.rate", (1.0 / lTimeDiff.count()));
-        }
       }
     };
 
@@ -407,7 +400,6 @@ void StfInputInterface::StfBuilderThread()
 void StfInputInterface::TopologicalStfBuilderThread()
 {
   using namespace std::chrono_literals;
-  using hres_clock = std::chrono::high_resolution_clock;
 
   bool lStarted = false;
   std::vector<FairMQMessagePtr> lReadoutMsgs(1U << 20);
@@ -443,43 +435,46 @@ void StfInputInterface::TopologicalStfBuilderThread()
     std::uint64_t(50000)
   );
 
-  IDDLOG("Aggregating {} pages for each link in topological STF building.", lNumPagesInThresholdScanStf);
+  // get a flag for cutting topo TFs on a new orbit
+  const bool lCutOnNewOrbit = mDiscoveryConfig->getBoolParam(StartTopologicalStfOnNewOrbitfKey, StartTopologicalStfOnNewOrbitDefault);
+
+  IDDLOG("Aggregating pages for each link in topological STF building. num_pages={} only_new_orbit={}",
+    lNumPagesInThresholdScanStf, lCutOnNewOrbit);
 
   // insert and mask the feeid. Return the Stf if aggregation is reached
-  auto lInsertTopoWithFeeIdMasking = [&lStfBuilder, lFeeIdMask, &lNumPagesInThresholdScanStf] (
+  // Note call again if lCutOnNewOrbit is not 0
+  auto lInsertTopoWithFeeIdMasking = [&lStfBuilder, lFeeIdMask, lNumPagesInThresholdScanStf, lCutOnNewOrbit] (
     const header::DataOrigin &pDataOrigin,
     const header::DataHeader::SubSpecificationType &pSubSpec,
     const ReadoutSubTimeframeHeader &pRdoHeader,
-    const FairMQParts::iterator pStartHbf,
-    const std::size_t pInsertCnt) {
+    FairMQParts::iterator &pStartHbf /* in/out */,
+    std::size_t &pInsertCnt /* in/out */) -> std::optional<std::unique_ptr<SubTimeFrame> > {
 
     // mask the subspecification if the fee mode is used
     auto lMaskedSubspec = pSubSpec;
     if (ReadoutDataUtils::SubSpecMode::eFeeId == ReadoutDataUtils::sRawDataSubspectype) {
       lMaskedSubspec &= lFeeIdMask;
     }
 
-    return lStfBuilder.addTopoStfData(pDataOrigin, lMaskedSubspec, pRdoHeader, pStartHbf, pInsertCnt, lNumPagesInThresholdScanStf);
+    return lStfBuilder.addTopoStfData(pDataOrigin, lMaskedSubspec, pRdoHeader, pStartHbf, pInsertCnt,
+      lNumPagesInThresholdScanStf, lCutOnNewOrbit);
   };
 
   const auto cStfDataWaitFor = 500ms;
-  std::chrono::time_point<hres_clock, std::chrono::duration<double>> lStartSec = hres_clock::now();
+  DDMON_RATE("stfbuilder", "stf_input", 0.0);
 
   auto queueStf = [&] (std::unique_ptr<SubTimeFrame> pStf) {
+    // static auto lStartSec = clock::now();
+
     static std::uint64_t sStfId = 0;
     if (pStf) {
       // make sure we queue STFs in ascending order
       sStfId += 1;
       pStf->updateId(sStfId);
       pStf->setOrigin(SubTimeFrame::Header::Origin::eReadoutTopology);
 
+      DDMON_RATE("stfbuilder", "stf_input", pStf->getDataSize());
       mDevice.I().queue(eStfBuilderOut, std::move(pStf));
-      {
-        auto lNow = hres_clock::now();
-        std::chrono::duration<double> lTimeDiff = lNow - lStartSec;
-        lStartSec = lNow;
-        DDMON("stfbuilder", "stf_input.rate", (1.0 / lTimeDiff.count()));
-      }
     }
   };
 
@@ -497,7 +492,6 @@ void StfInputInterface::TopologicalStfBuilderThread()
   };
 
   while (mRunning) {
-
     // Equipment ID for the HBFrames (from the header)
     lReadoutMsgs.clear();
 
@@ -558,9 +552,13 @@ void StfInputInterface::TopologicalStfBuilderThread()
     std::size_t lInsertCnt = lReadoutMsgs.size() - 1;
 
     // insert all HBFs
-    auto lStfOpt = lInsertTopoWithFeeIdMasking(lDataOrigin, lSubSpecification, lReadoutHdr, lStartHbf, lInsertCnt);
-    if (lStfOpt) {
-      queueStf(std::move(*lStfOpt));
+    while (lInsertCnt > 0) {
+      auto lStfOpt = lInsertTopoWithFeeIdMasking(lDataOrigin, lSubSpecification, lReadoutHdr, lStartHbf, lInsertCnt);
+      if (lStfOpt) {
+        queueStf(std::move(*lStfOpt));
+      } else {
+        assume(lInsertCnt == 0);
+      }
     }
   }