stfb: add --max-built-stfs paramter; send DPL SourceInfo::Completed on exit

Author: ironMann

src/StfBuilder/StfBuilderDevice.cxx

@@ -13,6 +13,7 @@
 
 #include "StfBuilderDevice.h"
 
+#include <DataDistLogger.h>
 #include <SubTimeFrameUtils.h>
 #include <SubTimeFrameVisitors.h>
 #include <ReadoutDataModel.h>
@@ -21,14 +22,13 @@
 #include <Utilities.h>
 
 #include <options/FairMQProgOptions.h>
+#include <Framework/SourceInfoHeader.h>
 
 #include <chrono>
 #include <thread>
 #include <exception>
 #include <boost/algorithm/string.hpp>
 
-#include <DataDistLogger.h>
-
 namespace o2
 {
 namespace DataDistribution
@@ -94,6 +94,7 @@ void StfBuilderDevice::InitTask()
   I().mDplChannelName = GetConfig()->GetValue<std::string>(OptionKeyDplChannelName);
   I().mStandalone = GetConfig()->GetValue<bool>(OptionKeyStandalone);
   I().mMaxStfsInPipeline = GetConfig()->GetValue<std::int64_t>(OptionKeyMaxBufferedStfs);
+  I().mMaxBuiltStfs = GetConfig()->GetValue<std::uint64_t>(OptionKeyMaxBuiltStfs);
 
   // input data handling
   ReadoutDataUtils::sSpecifiedDataOrigin = getDataOriginFromOption(
@@ -127,6 +128,10 @@ void StfBuilderDevice::InitTask()
       "Possibility of creating back-pressure.");
   }
 
+  // Limited number of STF?
+  DDLOGF(fair::Severity::INFO, "Configuration: Number of built SubTimeFrames is {}",
+    I().mMaxBuiltStfs == 0 ? "not limited" : ("limited to " + std::to_string(I().mMaxBuiltStfs)));
+
   // File sink
   if (!I().mFileSink->loadVerifyConfig(*(this->GetConfig()))) {
     exit(-1);
@@ -254,6 +259,9 @@ void StfBuilderDevice::ResetTask()
 {
   DDLOGF(fair::Severity::DEBUG, "StfBuilderDevice::ResetTask()");
 
+  // Signal ConditionalRun() and other threads to stop
+  I().mRunning = false;
+
   // stop the memory resources
   MemI().stop();
 
@@ -364,11 +372,41 @@ void StfBuilderDevice::StfOutputThread()
       const double lTimeMs = std::max(1e-6, std::chrono::duration<double, std::milli>(lNow - lSendStartTime).count());
       I().mSentOutRate = double(I().mSentOutStfs) / std::chrono::duration<double>(lNow - sStartOfStfSending).count();
       I().mStfDataTimeSamples.Fill(lTimeMs);
-    } else {
-      // DDLOGF(fair::Severity::ERROR, "Dropping stf size={}", lStf->getDataSize());
+    }
+
+    // check if we should exit:
+    // 1. max number of stf set, or
+    // 2. file reply used without loop parameter
+    if ( (I().mMaxBuiltStfs > 0) && (I().mSentOutStfsTotal == I().mMaxBuiltStfs) ) {
+      DDLOGF(fair::Severity::INFO, "Maximum number of sent SubTimeFrames reached. Exiting.");
+      break;
     }
   }
 
+  // leaving the output thread, send end of the stream info
+  if (dplEnabled()) {
+    o2::framework::SourceInfoHeader lDplExitHdr;
+    lDplExitHdr.state = o2::framework::InputChannelState::Completed;
+    auto lDoneStack = Stack(
+      DataHeader(gDataDescriptionInfo, gDataOriginAny, 0, 0),
+      o2::framework::DataProcessingHeader(),
+      lDplExitHdr
+    );
+
+    // Send a multipart
+    FairMQParts lCompletedMsg;
+    auto lNoFree = [](void*, void*) { /* stack */ };
+    lCompletedMsg.AddPart(lOutputChan.NewMessage(lDoneStack.data(), lDoneStack.size(), lNoFree));
+    lCompletedMsg.AddPart(lOutputChan.NewMessage());
+    lOutputChan.Send(lCompletedMsg);
+
+    DDLOGF(fair::Severity::INFO, "Sent Source Completed message to DPL.");
+    // NOTE: no guarantees this will be sent out
+    std::this_thread::sleep_for(2s);
+  }
+
+  I().mRunning = false; // trigger stop via CondRun()
+
   DDLOGF(fair::Severity::info, "Exiting StfOutputThread...");
 }
 
@@ -397,7 +435,12 @@ bool StfBuilderDevice::ConditionalRun()
 {
   // nothing to do here sleep for awhile
   std::this_thread::sleep_for(500ms);
-  return true;
+
+  if (!I().mRunning) {
+    DDLOGF(fair::Severity::DEBUG, "ConditionalRun() returning false.");
+  }
+
+  return I().mRunning;
 }
 
 bpo::options_description StfBuilderDevice::getDetectorProgramOptions() {

src/StfBuilder/StfBuilderDevice.h

@@ -65,6 +65,7 @@ class StfBuilderDevice : public DataDistDevice,
   static constexpr const char* OptionKeyDplChannelName = "dpl-channel-name";
   static constexpr const char* OptionKeyStandalone = "stand-alone";
   static constexpr const char* OptionKeyMaxBufferedStfs = "max-buffered-stfs";
+  static constexpr const char* OptionKeyMaxBuiltStfs = "max-built-stfs";
 
   static constexpr const char* OptionKeyStfDetector = "detector";
   static constexpr const char* OptionKeyRhdVer = "detector-rdh";
@@ -201,6 +202,7 @@ class StfBuilderDevice : public DataDistDevice,
     bool mStandalone;
     bool mDplEnabled;
     std::int64_t mMaxStfsInPipeline;
+    std::uint64_t mMaxBuiltStfs;
     bool mPipelineLimit;
 
     /// Input Interface handler
@@ -209,6 +211,7 @@ class StfBuilderDevice : public DataDistDevice,
 
     /// Internal threads
     std::thread mOutputThread;
+    std::atomic_bool mRunning = true;
     std::atomic_bool mPaused = false;
 
     /// File sink

src/StfBuilder/runStfBuilderDevice.cxx

@@ -45,6 +45,11 @@ void addCustomOptions(bpo::options_description& options)
       bpo::value<std::int64_t>()->default_value(-1),
       "Maximum number of buffered SubTimeFrames before starting to drop data (unlimited: -1)."
     )
+    (
+      o2::DataDistribution::StfBuilderDevice::OptionKeyMaxBuiltStfs,
+      bpo::value<std::uint64_t>()->default_value(0),
+      "Maximum number of built and forwarded (Sub)TimeFrames before closing (unlimited: 0, default)."
+    )
     (
       o2::DataDistribution::StfBuilderDevice::OptionKeyOutputChannelName,
       bpo::value<std::string>()->default_value("builder-stf-channel"),

src/common/SubTimeFrameFileSource.cxx

@@ -244,6 +244,8 @@ void SubTimeFrameFileSource::DataInjectThread()
       mReadStfQueue.size(), sNumSentStfs / getElapsedTime());
   }
 
+  mPipelineI.close(mPipelineStageOut);
+
   DDLOGF(fair::Severity::INFO, "Exiting file source inject thread...");
 }
 
@@ -257,13 +259,13 @@ void SubTimeFrameFileSource::DataHandlerThread()
     mFilesVector = getDataFileList();
   }
 
-  if (mFilesVector.empty()) {
-    DDLOGF(fair::Severity::ERROR, "(Sub)TimeFrame directory contains no data files.");
-    return;
-  }
-
   while (mRunning) {
 
+    if (mFilesVector.empty()) {
+      DDLOGF(fair::Severity::ERROR, "(Sub)TimeFrame directory contains no data files.");
+      break;
+    }
+
     for (const auto &lFileName : mFilesVector) {
       if (!mRunning) {
         break; // stop looping over files
@@ -300,6 +302,10 @@ void SubTimeFrameFileSource::DataHandlerThread()
       break;
     }
   }
+
+  // notify the injection thread to stop
+  mReadStfQueue.stop();
+
   DDLOGF(fair::Severity::INFO, "Exiting file source data load thread...");
 }
 

src/common/base/ConcurrentQueue.h

@@ -68,11 +68,14 @@ class ConcurrentContainerImpl
     mImpl->mCond.notify_all();
   }
 
+  // push a new element to the queue, while in the running state
+  // return false (fail) if not running
   template <typename... Args>
   bool push(Args&&... args)
   {
     std::unique_lock<std::mutex> lLock(mImpl->mLock);
     if (!mImpl->mRunning) {
+      mImpl->mCond.notify_all(); // just in case someone is waiting
       return false;
     }
 
@@ -89,6 +92,8 @@ class ConcurrentContainerImpl
     return true;
   }
 
+  // pop an element from the queue. Caller will block while the queue is running
+  // returns true on success
   bool pop(T& d)
   {
     std::unique_lock<std::mutex> lLock(mImpl->mLock);
@@ -173,6 +178,8 @@ class ConcurrentContainerImpl
     return mImpl->mContainer.size();
   }
 
+  bool empty() const { return size() == 0; }
+
   bool is_running() const { return mImpl->mRunning; }
 
   private:
@@ -260,6 +267,18 @@ class IFifoPipeline
     return false;
   }
 
+  // notify the receiver the queue is closed
+  void close(unsigned pStage)
+  {
+    assert(pStage < mPipelineQueues.size());
+    auto lNextStage = getNextPipelineStage(pStage);
+    assert((lNextStage <= mPipelineQueues.size()) && "next stage larger than expected");
+
+    if (lNextStage < mPipelineQueues.size()) {
+      mPipelineQueues[lNextStage].stop();
+    }
+  }
+
   T dequeue(unsigned pStage)
   {
     T t;