tfbuilder: implement timeouts and retries for StfDataRequest grpcs

Author: ironMann

src/StfSender/StfSenderOutput.cxx

@@ -406,10 +406,20 @@ void StfSenderOutput::sendStfToTfBuilder(const std::uint64_t pStfId, const std::
   assert(!pTfBuilderId.empty());
   std::scoped_lock lLock(mScheduledStfMapLock);
 
+  { // Check if we handled this grpc call before, and we are handling a retry now
+    const auto lResIter = mSchedulingResult.find(pTfBuilderId);
+    if (lResIter != mSchedulingResult.end()) {
+
+      if (lResIter->second.first == pStfId) {
+        pRes.set_status(lResIter->second.second);
+        return;
+      }
+    }
+  }
+
   const auto lStfIter = mScheduledStfMap.find(pStfId);
   // verify we have the STF.
   if (lStfIter == mScheduledStfMap.end()) {
-
     if (pTfBuilderId != "-1") {
       // request for Stf we don't have is an error
       pRes.set_status(StfDataResponse::DATA_DROPPED_UNKNOWN);
@@ -442,21 +452,23 @@ void StfSenderOutput::sendStfToTfBuilder(const std::uint64_t pStfId, const std::
     } else if (mOutputFairMQ) {
       lOk = mOutputFairMQ->sendStfToTfBuilder(pTfBuilderId, std::move(lStf));
     }
-    if (!lOk) {
-      pRes.set_status(StfDataResponse::TF_BUILDER_UNKNOWN);
-      mDropQueue.push(std::move(lStf));
-      EDDLOG_GRL(1000, "sendStfToTfBuilder: TfBuilder not known to StfSender. tfb_id={}", pTfBuilderId);
-      return;
-    }
-
-    // update status and counters
-    pRes.set_status(StfDataResponse::OK);
-    {
+    if (lOk) {
+      // update status and counters
+      pRes.set_status(StfDataResponse::OK);
       std::scoped_lock lCntLock(mCounters.mCountersLock);
       mCounters.mValues.mInSending.mSize += lStfSize;
       mCounters.mValues.mInSending.mCnt += 1;
+    } else {
+      pRes.set_status(StfDataResponse::TF_BUILDER_UNKNOWN);
+      mDropQueue.push(std::move(lStf));
+      EDDLOG_GRL(1000, "sendStfToTfBuilder: TfBuilder not known to StfSender. tfb_id={}", pTfBuilderId);
     }
   }
+
+  // remember the result for the stf
+  auto &lResultInfo = mSchedulingResult[pTfBuilderId];
+  lResultInfo.first = pStfId;
+  lResultInfo.second = pRes.status();
 }
 
 /// Drop thread

src/StfSender/StfSenderOutput.h

@@ -86,6 +86,12 @@ class StfSenderOutput
       mCounters.mValues = StdSenderOutputCounters::Values();
       mLastStfId = 0;
     }
+
+    { // clear the stf maps
+      std::scoped_lock lLock(mScheduledStfMapLock);
+      mScheduledStfMap.clear();
+      mSchedulingResult.clear();
+    }
   }
 
  private:
@@ -119,6 +125,7 @@ class StfSenderOutput
   std::uint64_t mLastStfId = 0;
   std::mutex mScheduledStfMapLock;
     std::map<std::uint64_t, ScheduledStfInfo> mScheduledStfMap;
+    std::map<std::string, std::pair<std::uint64_t, StfDataResponse::StfDataStatus> > mSchedulingResult; // use for answering retried StfRequest() grpc
 
   /// Buffer utilization counters
   StdSenderOutputCounters mCounters;

src/StfSender/StfSenderOutputUCX.cxx

@@ -420,6 +420,10 @@ void StfSenderOutputUCX::DataHandlerThread(unsigned pThreadIdx)
 {
   DDDLOG("StfSenderOutputUCX: Starting meta thread {}", pThreadIdx);
 
+#if defined(__linux__)
+  if (nice(-10)) { }
+#endif
+
   // local worker we advance here
   assert (pThreadIdx < mDataWorkers.size());
   ucx::dd_ucp_worker &lWorker = mDataWorkers[pThreadIdx];

src/StfSender/StfSenderRpc.cxx

@@ -166,7 +166,7 @@ ::grpc::Status StfSenderRpcImpl::TerminatePartition(::grpc::ServerContext* /*con
     return Status::CANCELLED;
   }
 
-  IDDLOG("TerminatePartition request received. partition_id={}", request->partition_id());
+  IDDLOG_GRL(5000, "TerminatePartition request received. partition_id={}", request->partition_id());
 
   mTerminateRequested = true;
   return Status::OK;

src/TfBuilder/TfBuilderInput.cxx

@@ -174,13 +174,13 @@ void TfBuilderInput::StfPacingThread()
     }
 
     assert (lStfInfo.mType == ReceivedStfMeta::MetaType::INFO);
-    assert (lStfInfo.mRecvStfdata);
+    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.get(), *lStfInfo.mRecvStfdata));
-      lStfInfo.mRecvStfdata = nullptr;
+      lStfInfo.mStf = std::move(lStfReceiver.deserialize(*lStfInfo.mRecvStfHeaderMeta.get(), *lStfInfo.mRecvStfData));
+      lStfInfo.mRecvStfData = nullptr;
 
       const std::uint64_t lNewTfId = mRpc->getIdForTopoTf(lStfInfo.mStfSenderId, lStfInfo.mStfId);
 
@@ -229,8 +229,8 @@ void TfBuilderInput::deserialize_headers(std::vector<ReceivedStfMeta> &pStfs)
     }
 
     // deserialize the data
-    lStfInfo.mStf = std::move(lStfReceiver.deserialize(*lStfInfo.mRecvStfHeaderMeta.get(), *lStfInfo.mRecvStfdata));
-    lStfInfo.mRecvStfdata = nullptr;
+    lStfInfo.mStf = std::move(lStfReceiver.deserialize(*lStfInfo.mRecvStfHeaderMeta.get(), *lStfInfo.mRecvStfData));
+    lStfInfo.mRecvStfData = nullptr;
   }
 }
 
@@ -259,6 +259,7 @@ void TfBuilderInput::StfDeserializingThread()
     // check if the TF should be dropped
     if (mStfIdsToDrop.count(lStfId) > 0) {
       mStfMergeMap.erase(mStfMergeMap.begin());
+      continue;
     }
 
     // deserialize headers of any new STFs
@@ -273,7 +274,7 @@ void TfBuilderInput::StfDeserializingThread()
     }
 
     // Check if the TF is completed
-    if (!lStfVector.empty() && (lStfVector.size() == lNumStfsOpt)) {
+    if (!lStfVector.empty() && (lStfVector.size() == lNumStfsOpt.value())) {
       mStfsForMerging.push(std::move(lStfVector));
 
       mStfMergeMap.erase(lStfId);

src/TfBuilder/TfBuilderInputDefs.h

@@ -34,7 +34,7 @@ struct ReceivedStfMeta {
     std::chrono::time_point<std::chrono::steady_clock> mTimeReceived;
 
     std::unique_ptr<IovStfHdrMeta> mRecvStfHeaderMeta;
-    std::unique_ptr<std::vector<FairMQMessagePtr>> mRecvStfdata;
+    std::unique_ptr<std::vector<FairMQMessagePtr>> mRecvStfData;
     std::unique_ptr<SubTimeFrame> mStf;
     std::string mStfSenderId;
 
@@ -47,13 +47,13 @@ struct ReceivedStfMeta {
                     const SubTimeFrame::Header::Origin pStfOrigin,
                     const std::string &pStfSenderId,
                     std::unique_ptr<IovStfHdrMeta> &&pRcvHdrMeta,
-                    std::unique_ptr<std::vector<FairMQMessagePtr>> &&pRecvStfdata)
+                    std::unique_ptr<std::vector<FairMQMessagePtr>> &&pRecvStfData)
     : mType(INFO),
       mStfId(pStfId),
       mStfOrigin(pStfOrigin),
       mTimeReceived(std::chrono::steady_clock::now()),
       mRecvStfHeaderMeta(std::move(pRcvHdrMeta)),
-      mRecvStfdata(std::move(pRecvStfdata)),
+      mRecvStfData(std::move(pRecvStfData)),
       mStf(nullptr),
       mStfSenderId(pStfSenderId)
     { }

src/TfBuilder/TfBuilderInputUCX.cxx

@@ -187,8 +187,8 @@ bool TfBuilderInputUCX::start()
 {
   // setting configuration options
   mThreadPoolSize = std::clamp(mConfig->getUInt64Param(UcxTfBuilderThreadPoolSizeKey, UcxTfBuilderThreadPoolSizeDefault), std::size_t(1), std::size_t(256));
-  mRdmaPollingWait =mConfig->getBoolParam(UcxPollForRDMACompletionKey, UcxPollForRDMACompletionDefault);
-  mRdmaConcurrentStfSizeMax =mConfig->getUInt64Param(UcxMaxStfSizeForConcurrentFetchBKey, UcxMaxStfSizeForConcurrentFetchBDefault);
+  mRdmaPollingWait = mConfig->getBoolParam(UcxPollForRDMACompletionKey, UcxPollForRDMACompletionDefault);
+  mRdmaConcurrentStfSizeMax = mConfig->getUInt64Param(UcxMaxStfSizeForConcurrentFetchBKey, UcxMaxStfSizeForConcurrentFetchBDefault);
   IDDLOG("TfBuilderInputUCX: Configuration loaded. thread_pool={} polling={} concurrent_size={}", mThreadPoolSize, mRdmaPollingWait, mRdmaConcurrentStfSizeMax);
 
   auto &lConfStatus = mConfig->status();