emu: make the TFID dependent on system time to allow testing on cluster without external synchronization

Author: ironMann

src/ReadoutEmulator/CruEmulator.cxx

@@ -27,14 +27,13 @@ namespace DataDistribution
 
 void CruLinkEmulator::linkReadoutThread()
 {
-  static const size_t cHBFrameFreq = 11223;
-  static const size_t cStfPerS = 43; /* Parametrize this? */
+  static const std::uint64_t cHBFrameFreq = 11223;
 
   const auto cSuperpageSize = mMemHandler->getSuperpageSize();
   const auto cHBFrameSize = (mLinkBitsPerS / cHBFrameFreq) >> 3;
-  const auto cStfLinkSize = cHBFrameSize * cHBFrameFreq / cStfPerS;
+  const auto cStfLinkSize = cHBFrameSize * cHBFrameFreq * 256 / cHBFrameFreq;
   const auto cNumDmaChunkPerSuperpage = std::min(size_t(256), size_t(cSuperpageSize / mDmaChunkSize));
-  constexpr int64_t cStfTimeUs = std::chrono::microseconds(1000000 / cStfPerS).count();
+  constexpr int64_t cStfTimeUs = std::chrono::microseconds(std::uint64_t(1000000) * 256 / cHBFrameFreq).count();
 
   LOG(DEBUG) << "Superpage size: " << cSuperpageSize;
   LOG(DEBUG) << "mDmaChunkSize size: " << mDmaChunkSize;
@@ -57,9 +56,7 @@ void CruLinkEmulator::linkReadoutThread()
     const int64_t lStfToSend = lUsSinceStart / cStfTimeUs - lSentStf;
 
     if (lStfToSend <= 0) {
-      // std::this_thread::sleep_for(std::chrono::microseconds());
       usleep(500);
-      // std::this_thread::yield();
       continue;
     }
 

src/ReadoutEmulator/ReadoutDevice.cxx

@@ -89,6 +89,8 @@ void ReadoutDevice::PreRun()
     mGui = std::make_unique<RootGui>("Emulator", "Readout Emulator", 500, 500);
     mGuiThread = std::thread(&ReadoutDevice::GuiThread, this);
   }
+
+  mSendingThread = std::thread(&ReadoutDevice::SendingThread, this);
 }
 
 void ReadoutDevice::ResetTask()
@@ -101,65 +103,85 @@ void ReadoutDevice::ResetTask()
   if (mBuildHistograms && mGuiThread.joinable()) {
     mGuiThread.join();
   }
+
+  if (mSendingThread.joinable()) {
+    mSendingThread.join();
+  }
 }
 
 bool ReadoutDevice::ConditionalRun()
 {
+  using namespace std::chrono_literals;
+  std::this_thread::sleep_for(1s);
+  return true;
+}
+
+void ReadoutDevice::SendingThread()
+{
+
+  WaitForRunningState();
+
   auto& lOutputChan = GetChannel(mOutChannelName, 0);
 
   // finish an STF every ~1/45 seconds
   static const auto cDataTakingStart = std::chrono::high_resolution_clock::now();
-  static constexpr auto cStfInterval = std::chrono::microseconds(22222);
+  static constexpr auto cStfInterval = std::chrono::microseconds(22810);
   static uint64_t lNumberSentStfs = 0;
+  static uint64_t lCurrentTfId = 0;
 
-  auto isStfFinished =
-    (std::chrono::high_resolution_clock::now() - cDataTakingStart) - (lNumberSentStfs * cStfInterval) > cStfInterval;
-
-  if (isStfFinished)
-    lNumberSentStfs += 1;
-
-  ReadoutLinkO2Data lCruLinkData;
-  if (!mCruMemoryHandler->getLinkData(lCruLinkData)) {
-    LOG(INFO) << "GetLinkData failed. Stopping interface thread.";
-    return false;
-  }
-
-  if (mBuildHistograms)
-    mFreeSuperpagesSamples.Fill(mCruMemoryHandler->free_superpages());
-
-  // check no data signal
-  if (lCruLinkData.mLinkDataHeader.subSpecification ==
-    o2::header::DataHeader::SubSpecificationType(-1)) {
-    // LOG(WARN) << "No Superpages left! Losing data...";
-    return true;
-  }
-
-  ReadoutSubTimeframeHeader lHBFHeader;
-  lHBFHeader.mTimeFrameId = lNumberSentStfs;
-  lHBFHeader.mNumberHbf = lCruLinkData.mLinkRawData.size();
-  lHBFHeader.mLinkId = lCruLinkData.mLinkDataHeader.subSpecification;
-
-  assert(mDataBlockMsgs.empty());
-  mDataBlockMsgs.reserve(lCruLinkData.mLinkRawData.size());
-
-  // create messages for the header
-  mDataBlockMsgs.emplace_back(lOutputChan.NewMessage(sizeof(ReadoutSubTimeframeHeader)));
-  std::memcpy(mDataBlockMsgs.front()->GetData(), &lHBFHeader, sizeof(ReadoutSubTimeframeHeader));
-
-  // create messages for the data
-  for (const auto& lDmaChunk : lCruLinkData.mLinkRawData) {
-    // mark this as used in the memory handler
-    mCruMemoryHandler->get_data_buffer(lDmaChunk.mDataPtr, lDmaChunk.mDataSize);
+  while (IsRunningState()) {
 
-    // create a message out of unmanaged region
-    mDataBlockMsgs.emplace_back(lOutputChan.NewMessage(mDataRegion, lDmaChunk.mDataPtr, lDmaChunk.mDataSize));
+    auto isStfFinished =
+      (std::chrono::high_resolution_clock::now() - cDataTakingStart) - (lNumberSentStfs * cStfInterval) > cStfInterval;
+
+    if (isStfFinished) {
+      lNumberSentStfs += 1;
+      lCurrentTfId = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::high_resolution_clock::now().time_since_epoch()).count() / cStfInterval.count();
+    }
+
+    ReadoutLinkO2Data lCruLinkData;
+    if (!mCruMemoryHandler->getLinkData(lCruLinkData)) {
+      LOG(INFO) << "GetLinkData failed. Stopping interface thread.";
+      return;
+    }
+
+    if (mBuildHistograms) {
+      mFreeSuperpagesSamples.Fill(mCruMemoryHandler->free_superpages());
+    }
+
+    // check no data signal
+    if (lCruLinkData.mLinkDataHeader.subSpecification ==
+      o2::header::DataHeader::SubSpecificationType(-1)) {
+      // LOG(WARN) << "No Superpages left! Losing data...";
+    }
+
+    ReadoutSubTimeframeHeader lHBFHeader;
+    lHBFHeader.mTimeFrameId = lCurrentTfId;
+    lHBFHeader.mNumberHbf = lCruLinkData.mLinkRawData.size();
+    lHBFHeader.mLinkId = lCruLinkData.mLinkDataHeader.subSpecification;
+
+    assert(mDataBlockMsgs.empty());
+    mDataBlockMsgs.reserve(lCruLinkData.mLinkRawData.size());
+
+    // create messages for the header
+    mDataBlockMsgs.emplace_back(lOutputChan.NewMessage(sizeof(ReadoutSubTimeframeHeader)));
+    std::memcpy(mDataBlockMsgs.front()->GetData(), &lHBFHeader, sizeof(ReadoutSubTimeframeHeader));
+
+    // create messages for the data
+    for (const auto& lDmaChunk : lCruLinkData.mLinkRawData) {
+      // mark this as used in the memory handler
+      mCruMemoryHandler->get_data_buffer(lDmaChunk.mDataPtr, lDmaChunk.mDataSize);
+
+      // create a message out of unmanaged region
+      mDataBlockMsgs.emplace_back(lOutputChan.NewMessage(mDataRegion, lDmaChunk.mDataPtr, lDmaChunk.mDataSize));
+    }
+
+    lOutputChan.Send(mDataBlockMsgs);
+    mDataBlockMsgs.clear();
   }
+}
 
-  lOutputChan.Send(mDataBlockMsgs);
-  mDataBlockMsgs.clear();
 
-  return true;
-}
 
 void ReadoutDevice::GuiThread()
 {

src/ReadoutEmulator/ReadoutDevice.h

@@ -58,6 +58,7 @@ class ReadoutDevice : public DataDistDevice
 
 
   void GuiThread();
+  void SendingThread();
 
   // data and Descriptor regions
   // must be here because NewUnmanagedRegionFor() is a method of FairMQDevice...
@@ -79,6 +80,7 @@ class ReadoutDevice : public DataDistDevice
 
   // messages to send
   std::vector<FairMQMessagePtr> mDataBlockMsgs;
+  std::thread mSendingThread;
 
   /// Observables
   bool mBuildHistograms = true;