stfsender: fix deadlock when FLP-standalone memory pressure test is active

Author: ironMann

script/start_Discovery-3FLP-3EPN.sh

@@ -160,6 +160,10 @@ STF_SENDER+=" --mq-config $chainConfig"
 STF_SENDER+=" --monitoring-interval=2.0"
 STF_SENDER+=" --monitoring-log"
 STF_SENDER+=" --shm-monitor=false"
+if [[ $EPN_CNT -eq 0 ]]; then
+  STF_SENDER+=" --stand-alone"
+fi
+
 
 TF_BUILDER="TfBuilder"
 TF_BUILDER+=" --discovery-net-if=$EPN_NETIF"

src/StfSender/StfSenderOutput.cxx

@@ -202,27 +202,18 @@ void StfSenderOutput::StfKeepThread()
   std::default_random_engine lGen;
   std::uniform_int_distribution<unsigned> lUniformDist(0, 99);
 
-  std::unique_ptr<SubTimeFrame> lStf;
-
   mDeletePercentage = mDiscoveryConfig->getUInt64Param(StandaloneStfDeleteChanceKey, StandaloneStfDeleteChanceDefault);
   mDeletePercentage = std::clamp(mDeletePercentage.load(), std::uint64_t(0), std::uint64_t(100));
 
-  while ((lStf = mPipelineI.dequeue(eSenderIn)) != nullptr) {
-    std::scoped_lock lMapLock(mStfKeepMapLock);
-
-    const auto lStfId = lStf->id();
-    const auto lStfSize = lStf->getDataSize();
+  while (mRunning) {
+    auto lStfOpt = mPipelineI.dequeue_for(eSenderIn, 50ms);
 
-    if (mStfKeepMap.count(lStfId) == 0) {
-      mStfKeepMap[lStfId] = std::move(lStf);
-    }
+    std::scoped_lock lMapLock(mStfKeepMapLock);
 
-    // update buffer sizes
+    // get buffer sizes
     StdSenderOutputCounters::Values lCounters;
     {
       std::scoped_lock lLock(mCounters.mCountersLock);
-      mCounters.mValues.mBuffered.mSize += lStfSize;
-      mCounters.mValues.mBuffered.mCnt += 1;
       lCounters = mCounters.mValues;
     }
 
@@ -253,6 +244,23 @@ void StfSenderOutput::StfKeepThread()
       mDropQueue.push(std::move(lIter->second));
       mStfKeepMap.erase(lIter->first);
     }
+
+    if (lStfOpt) {
+      std::unique_ptr<SubTimeFrame> lStf = std::move(lStfOpt.value());
+
+      const auto lStfId = lStf->id();
+      const auto lStfSize = lStf->getDataSize();
+
+      if (mStfKeepMap.count(lStfId) == 0) {
+        mStfKeepMap[lStfId] = std::move(lStf);
+        // update buffer sizes
+        {
+          std::scoped_lock lLock(mCounters.mCountersLock);
+          mCounters.mValues.mBuffered.mSize += lStfSize;
+          mCounters.mValues.mBuffered.mCnt += 1;
+        }
+      }
+    }
   }
 
   std::scoped_lock lMapLock(mStfKeepMapLock);