tfbuilder: make sure tfs are built and forwarded in-order

Author: ironMann

src/TfBuilder/TfBuilderDevice.cxx

@@ -171,7 +171,7 @@ void TfBuilderDevice::InitTask()
 
 bool TfBuilderDevice::start()
 {
-  while (!mRpc->start(mTfDataRegionSize)) {
+  while (!mRpc->start(mTfDataRegionSize, mFlpInputHandler->getDataQueue())) {
     // check if should stop looking for TfScheduler
     if (mRpc->isTerminateRequested()) {
       mShouldExit = true;

src/TfBuilder/TfBuilderInput.cxx

@@ -37,7 +37,8 @@ TfBuilderInput::TfBuilderInput(TfBuilderDevice& pStfBuilderDev, std::shared_ptr<
       mRpc(pRpc),
       mOutStage(pOutStage)
   {
-    mInputFairMQ = std::make_unique<TfBuilderInputFairMQ>(pRpc, pStfBuilderDev.TfBuilderI(), mReceivedDataQueue);
+    mReceivedDataQueue = std::make_shared<ConcurrentQueue<ReceivedStfMeta>>();
+    mInputFairMQ = std::make_unique<TfBuilderInputFairMQ>(pRpc, pStfBuilderDev.TfBuilderI(), *mReceivedDataQueue);
   }
 
 bool TfBuilderInput::start(std::shared_ptr<ConsulTfBuilder> pConfig)
@@ -53,7 +54,7 @@ bool TfBuilderInput::start(std::shared_ptr<ConsulTfBuilder> pConfig)
   mState = RUNNING;
 
   // Start the pacer thread
-  mReceivedDataQueue.start();
+  mReceivedDataQueue->start();
   mStfPacingThread = create_thread_member("tfb_pace", &TfBuilderInput::StfPacingThread, this);
 
   // Start the deserialize thread
@@ -84,7 +85,7 @@ void TfBuilderInput::stop(std::shared_ptr<ConsulTfBuilder> pConfig)
   mInputFairMQ->stop(pConfig);
 
   //Wait for pacer thread
-  mReceivedDataQueue.stop();
+  mReceivedDataQueue->stop();
   if (mStfPacingThread.joinable()) {
     mStfPacingThread.join();
   }
@@ -119,30 +120,50 @@ void TfBuilderInput::StfPacingThread()
 {
   // Deserialization object (stf ID)
   IovDeserializer lStfReceiver(mDevice.TfBuilderI());
-  std::uint64_t lTopoStfId = 0;
 
   while (mState == RUNNING) {
 
-    auto lData = mReceivedDataQueue.pop();
+    auto lData = mReceivedDataQueue->pop();
     if (lData == std::nullopt) {
       continue;
     }
     auto &lStfInfo = lData.value();
 
-    assert (lStfInfo.mRecvStfdata);
     std::uint64_t lTfId = lStfInfo.mStfId;
 
+    if (lStfInfo.mType == ReceivedStfMeta::MetaType::ADD) {
+      // only record the intent to build a TF
+      std::unique_lock<std::mutex> lQueueLock(mStfMergerQueueLock);
+      assert (mStfMergeMap.count(lTfId) == 0);
+      assert (mStfMergeMap.empty() || (mStfMergeMap.rbegin()->first < lTfId));
+
+      mStfMergeMap[lTfId].reserve(mNumStfSenders);
+      continue;
+    } else if (lStfInfo.mType == ReceivedStfMeta::MetaType::DELETE) {
+      // remove tf merge intent if no StfSenders were contacted
+      std::unique_lock<std::mutex> lQueueLock(mStfMergerQueueLock);
+      assert (mStfMergeMap.count(lTfId) == 1);
+      assert (mStfMergeMap[lTfId].empty());
+
+      mStfMergeMap.erase(lTfId);
+      continue;
+    }
+
+    assert (lStfInfo.mType == ReceivedStfMeta::MetaType::INFO);
+    assert (lStfInfo.mRecvStfdata);
+
     // Rename STF id if this is a Topological TF
     if (lStfInfo.mStfOrigin == SubTimeFrame::Header::Origin::eReadoutTopology) {
       // deserialize here to be able to rename the stf
       lStfInfo.mStf = std::move(lStfReceiver.deserialize(lStfInfo.mRecvStfHeaderMeta, *lStfInfo.mRecvStfdata));
       lStfInfo.mRecvStfdata = nullptr;
 
-      const std::uint64_t lNewTfId = ++lTopoStfId;
+      const std::uint64_t lNewTfId = mRpc->getIdForTopoTf(lStfInfo.mStfSenderId, lStfInfo.mStfId);
+
       DDDLOG_RL(5000, "Deserialized STF. stf_id={} new_id={}", lStfInfo.mStf->id(), lNewTfId);
       lStfInfo.mStf->updateId(lNewTfId);
       lTfId = lNewTfId;
-      lStfInfo.mStfId = lTfId;
+      lStfInfo.mStfId = lNewTfId;
     }
 
     /// TODO: STF receive pacing
@@ -185,29 +206,6 @@ void TfBuilderInput::deserialize_headers(std::vector<ReceivedStfMeta> &pStfs)
   }
 }
 
-// check if topological (S)TF
-bool TfBuilderInput::is_topo_stf(const std::vector<ReceivedStfMeta> &pStfs) const
-{
-  if (pStfs.empty()) {
-    return false;
-  }
-
-  if (pStfs.size() == 1 && pStfs.begin()->mStf) {
-    if (pStfs.begin()->mStf->header().mOrigin == SubTimeFrame::Header::Origin::eReadoutTopology) {
-      return true;
-    }
-  }
-
-  if (pStfs.size() > 1 && pStfs.begin()->mStf) {
-    if (pStfs.begin()->mStf->header().mOrigin == SubTimeFrame::Header::Origin::eReadoutTopology) {
-      EDDLOG_RL(1000, "TfBuilderInput::is_topo_stf: multiple topological STFs with a same ID. stf_id={}",
-        pStfs.begin()->mStf->id());
-    }
-  }
-
-  return false;
-}
-
 /// FMQ->STF thread
 /// This thread can block waiting on free O2 Header memory
 void TfBuilderInput::StfDeserializingThread()

src/TfBuilder/TfBuilderInput.h

@@ -45,13 +45,15 @@ class TfBuilderInput
   bool start(std::shared_ptr<ConsulTfBuilder> pConfig);
   void stop(std::shared_ptr<ConsulTfBuilder> pConfig);
   void reset() {
-    mReceivedDataQueue.flush();
+    mReceivedDataQueue->flush();
     mStfsForMerging.flush();
     std::unique_lock<std::mutex> lQueueLock(mStfMergerQueueLock);
     mStfMergeMap.clear();
     mMaxMergedTfId = 0;
   }
 
+  auto getDataQueue() const { return mReceivedDataQueue; }
+
   void StfPacingThread();
   void StfDeserializingThread();
   void StfMergerThread();
@@ -72,7 +74,7 @@ class TfBuilderInput
   std::unique_ptr<TfBuilderInputFairMQ> mInputFairMQ;
 
   /// Received Stfs from input stage
-  ConcurrentQueue<ReceivedStfMeta> mReceivedDataQueue;
+  std::shared_ptr<ConcurrentQueue<ReceivedStfMeta> > mReceivedDataQueue;
   std::thread mStfPacingThread;
 
   /// Stf Deserializer (add O2 headers etc)

src/TfBuilder/TfBuilderInputDefs.h

@@ -14,21 +14,20 @@
 #ifndef TF_BUILDER_INPUT_DEFS_H_
 #define TF_BUILDER_INPUT_DEFS_H_
 
-
 #include <SubTimeFrameDataModel.h>
 #include <ConcurrentQueue.h>
 
 #include <chrono>
 #include <vector>
 #include <memory>
 
-
 namespace o2::DataDistribution
 {
 
 enum InputRunState { CONFIGURING, RUNNING, TERMINATED };
 
 struct ReceivedStfMeta {
+    enum MetaType { ADD, DELETE, INFO } mType;
     TimeFrameIdType mStfId;
     SubTimeFrame::Header::Origin mStfOrigin;
     std::chrono::time_point<std::chrono::steady_clock> mTimeReceived;
@@ -38,9 +37,18 @@ struct ReceivedStfMeta {
     std::unique_ptr<SubTimeFrame> mStf;
     std::string mStfSenderId;
 
-    ReceivedStfMeta(const TimeFrameIdType pStfId, const SubTimeFrame::Header::Origin pStfOrigin,
-      const std::string &pStfSenderId, FairMQMessagePtr &&pRcvHdrMsg, std::unique_ptr<std::vector<FairMQMessagePtr>> &&pRecvStfdata)
-    : mStfId(pStfId),
+    ReceivedStfMeta(MetaType pType, const TimeFrameIdType pStfId)
+    : mType(pType),
+      mStfId(pStfId)
+    { }
+
+    ReceivedStfMeta(const TimeFrameIdType pStfId,
+                    const SubTimeFrame::Header::Origin pStfOrigin,
+                    const std::string &pStfSenderId,
+                    FairMQMessagePtr &&pRcvHdrMsg,
+                    std::unique_ptr<std::vector<FairMQMessagePtr>> &&pRecvStfdata)
+    : mType(INFO),
+      mStfId(pStfId),
       mStfOrigin(pStfOrigin),
       mTimeReceived(std::chrono::steady_clock::now()),
       mRecvStfHeaderMeta(std::move(pRcvHdrMsg)),

src/TfBuilder/TfBuilderRpc.cxx

@@ -12,6 +12,7 @@
 // along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 #include "TfBuilderRpc.h"
+#include "TfBuilderInputDefs.h"
 #include <grpcpp/grpcpp.h>
 
 #include <MemoryUtils.h>
@@ -39,10 +40,11 @@ void TfBuilderRpcImpl::initDiscovery(const std::string pRpcSrvBindIp, int &lReal
   IDDLOG("gRPC server is started. server_ep={}:{}", pRpcSrvBindIp, lRealPort);
 }
 
-bool TfBuilderRpcImpl::start(const std::uint64_t pBufferSize)
+bool TfBuilderRpcImpl::start(const std::uint64_t pBufferSize, std::shared_ptr<ConcurrentQueue<ReceivedStfMeta> > pRecvQueue)
 {
   mBufferSize = pBufferSize;
   mCurrentTfBufferSize = pBufferSize;
+  mReceivedDataQueue = pRecvQueue;
 
   // Interact with the scheduler
   if (!mTfSchedulerRpcClient.should_retry_start()) {
@@ -313,8 +315,29 @@ ::grpc::Status TfBuilderRpcImpl::BuildTfRequest(::grpc::ServerContext* /*context
     return a.mStfDataSize < b.mStfDataSize;
   });
 
+  // setup renaming of topological Stfs
+  auto lTopoStfId = mTopoStfId;
+  if (request->tf_source() == TOPOLOGICAL) {
+    auto &lStfSenderId = request->stf_size_map().begin()->first;
+
+    std::scoped_lock lLock(mTopoTfIdLock);
+
+    assert (mTopoTfIdRenameMap[lStfSenderId].count(lTfId) == 0);
+
+    mTopoTfIdRenameMap[lStfSenderId][lTfId] = lTopoStfId;
+
+    // notify Input stage about new Stf (renamed)
+    mReceivedDataQueue->push(ReceivedStfMeta(ReceivedStfMeta::MetaType::ADD, lTopoStfId));
+
+    mTopoStfId += 1;
+  } else {
+    lTopoStfId = lTfId; // set to the actual tf id if not topo
+    // notify Input stage about new Stf (regular)
+    mReceivedDataQueue->push(ReceivedStfMeta(ReceivedStfMeta::MetaType::ADD, lTfId));
+  }
+
   // add the vector to the stf request map
-  mStfRequestDeque.push(lTfId, std::move(lStfRequestVector));
+  mStfRequestQueue.push(lTfId, (request->tf_source() == TOPOLOGICAL), lTopoStfId, std::move(lStfRequestVector));
 
   response->set_status(BuildTfResponse::OK);
   return ::grpc::Status::OK;
@@ -335,14 +358,22 @@ void TfBuilderRpcImpl::StfRequestThread()
   while (mRunning) {
     StfRequests lStfRequest;
     {
-      std::optional<std::pair<std::uint64_t, std::vector<StfRequests>> > lReqOpt;
-      if ((lReqOpt = mStfRequestDeque.pop_wait_for(100ms)) == std::nullopt) {
+      std::optional<std::tuple<std::uint64_t, bool, std::uint64_t, std::vector<StfRequests>> > lReqOpt;
+      if ((lReqOpt = mStfRequestQueue.pop_wait_for(100ms)) == std::nullopt) {
         continue;
       }
 
       assert (lReqOpt);
-      const auto lTfId = lReqOpt.value().first;
-      auto &lReqVector = lReqOpt.value().second;
+      const auto lTfId = std::get<0>(lReqOpt.value());
+      const auto lIsTopo = std::get<1>(lReqOpt.value());
+      const auto lTfRenamedId = std::get<2>(lReqOpt.value());
+      auto &lReqVector = std::get<3>(lReqOpt.value());
+
+      std::string lStfSenderIdTopo;
+      if (lIsTopo) {
+        assert (lReqVector.size() == 1);
+        lStfSenderIdTopo = lReqVector.front().mStfSenderId;
+      }
 
       mMaxNumReqInFlight = std::clamp(mDiscoveryConfig->getUInt64Param(MaxNumStfTransfersKey, MaxNumStfTransferDefault),
         std::uint64_t(10), std::uint64_t(200));
@@ -416,7 +447,29 @@ void TfBuilderRpcImpl::StfRequestThread()
       }
 
       // set the number of STFs for merging thread
-      setNumberOfStfs(lTfId, lNumExpectedStfs);
+      if (!lIsTopo) {
+        setNumberOfStfs(lTfId, lNumExpectedStfs);
+      } else {
+        setNumberOfStfs(lTfRenamedId, lNumExpectedStfs);
+      }
+
+      // cleanup if we reached no StfSenders
+      if (lNumExpectedStfs == 0) {
+        if (lIsTopo) {
+          // Topological: indicate that we're deleting topological (renamed) Id
+          std::scoped_lock lLock(mTopoTfIdLock);
+          assert (mTopoTfIdRenameMap[lStfSenderIdTopo].count(lTfId) == 1);
+          assert (lTfRenamedId == mTopoTfIdRenameMap[lStfSenderIdTopo][lTfId]);
+
+          mTopoTfIdRenameMap[lStfSenderIdTopo].erase(lTfId);
+
+          // notify Input stage about new Stf (renamed)
+          mReceivedDataQueue->push(ReceivedStfMeta(ReceivedStfMeta::MetaType::DELETE, lTfRenamedId));
+        } else {
+          // notify Input stage not to wait for STFs if we reached none of StfSender
+          mReceivedDataQueue->push(ReceivedStfMeta(ReceivedStfMeta::MetaType::DELETE, lTfId));
+        }
+      }
     }
   }
   // send disconnect update

src/TfBuilder/TfBuilderRpc.h

@@ -14,6 +14,8 @@
 #ifndef ALICEO2_TF_BUILDER_RPC_H_
 #define ALICEO2_TF_BUILDER_RPC_H_
 
+#include "TfBuilderInputDefs.h"
+
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wunused-parameter"
 #include <grpcpp/grpcpp.h>
@@ -57,7 +59,7 @@ class TfBuilderRpcImpl final : public TfBuilderRpc::Service
   TfSchedulerRpcClient& TfSchedRpcCli() { return mTfSchedulerRpcClient; }
 
   void initDiscovery(const std::string pRpcSrvBindIp, int &lRealPort /*[out]*/);
-  bool start(const std::uint64_t pBufferSize);
+  bool start(const std::uint64_t pBufferSize, std::shared_ptr<ConcurrentQueue<ReceivedStfMeta> > pRecvQueue);
   void stop();
 
   void startAcceptingTfs();
@@ -89,12 +91,17 @@ class TfBuilderRpcImpl final : public TfBuilderRpc::Service
 
   /// reset counters on each new run
   void reset_run_counters() {
+    std::scoped_lock lLock(mTfIdSizesLock, mTopoTfIdLock);
+
     mTfIdSizes.clear();
     mCurrentTfBufferSize = mBufferSize;
     mLastBuiltTfId = 0;
     mNumBufferedTfs = 0;
     mNumTfsInBuilding = 0;
     mTfBuildRequests->flush();
+    // Reset Topo Tf Id renaming
+    mTopoStfId = 1;
+    mTopoTfIdRenameMap.clear();
   }
 
   std::optional<std::uint64_t> getNumberOfStfs(const TimeFrameIdType pTfId) {
@@ -113,6 +120,19 @@ class TfBuilderRpcImpl final : public TfBuilderRpc::Service
     }
   }
 
+  std::uint64_t getIdForTopoTf(const std::string &pStfSenderId, const std::uint64_t pTfId) {
+    std::scoped_lock lLock(mTopoTfIdLock);
+
+    assert (mTopoTfIdRenameMap.count(pStfSenderId) == 1);
+    assert (mTopoTfIdRenameMap[pStfSenderId].count(pTfId) == 1);
+
+    const auto lRet = mTopoTfIdRenameMap[pStfSenderId][pTfId];
+    mTopoTfIdRenameMap[pStfSenderId].erase(pTfId);
+
+    assert (lRet > 0);
+    return lRet;
+  }
+
 private:
   std::atomic_bool mRunning = false;
   std::atomic_bool mTerminateRequested = false;
@@ -125,8 +145,8 @@ class TfBuilderRpcImpl final : public TfBuilderRpc::Service
   std::thread mUpdateThread;
 
   // Stf request thread
-  // std::condition_variable mNewRequestCondition;
   std::thread mStfRequestThread;
+  std::shared_ptr<ConcurrentQueue<ReceivedStfMeta> > mReceivedDataQueue;
 
     struct StfRequests {
       std::string mStfSenderId;
@@ -140,7 +160,13 @@ class TfBuilderRpcImpl final : public TfBuilderRpc::Service
 
   std::atomic_int64_t mMaxNumReqInFlight = 64;
   std::atomic_int64_t mNumReqInFlight = 0;
-  ConcurrentFifo<std::pair<std::uint64_t, std::vector<StfRequests>> >  mStfRequestDeque; // <tfid, stf_reuests>
+
+  // <tfid, topo?, topo_id, stf_requests>
+  ConcurrentFifo<std::tuple<std::uint64_t, bool, std::uint64_t, std::vector<StfRequests>> >  mStfRequestQueue;
+
+  std::mutex mTopoTfIdLock;
+  std::uint64_t mTopoStfId = 1;
+  std::map<std::string, std::unordered_map<std::uint64_t, std::uint64_t> > mTopoTfIdRenameMap;
 
   // Store how many STFs were requested per TfID. nullopt if requests are not finished
   std::mutex mStfsCountMapLock;