GPU Workflow: Fix possible deadlock with DoublePipeline when not receiving EndOfStream

GPU/GPUTracking/Base/GPUReconstruction.cxx

@@ -66,6 +66,7 @@ struct GPUReconstructionPipelineContext {
   std::queue<GPUReconstructionPipelineQueue*> pipelineQueue;
   std::mutex mutex;
   std::condition_variable cond;
+  bool workerRunning = false;
   bool terminate = false;
 };
 } // namespace o2::gpu
@@ -1094,6 +1095,7 @@ void GPUReconstruction::RunPipelineWorker()
     {
       std::unique_lock<std::mutex> lk(mPipelineContext->mutex);
       mPipelineContext->cond.wait(lk, [this] { return this->mPipelineContext->pipelineQueue.size() > 0; });
+      mPipelineContext->workerRunning = true;
     }
     GPUReconstructionPipelineQueue* q;
     {
@@ -1111,6 +1113,8 @@ void GPUReconstruction::RunPipelineWorker()
       q->done = true;
     }
     q->c.notify_one();
+    mPipelineContext->workerRunning = false;
+    mPipelineContext->cond.notify_one();
   }
   if (GetProcessingSettings().debugLevel >= 3) {
     GPUInfo("Pipeline worker ended");
@@ -1122,6 +1126,12 @@ void GPUReconstruction::TerminatePipelineWorker()
   EnqueuePipeline(true);
 }
 
+void GPUReconstruction::DrainPipeline()
+{
+  std::unique_lock<std::mutex> lk(mPipelineContext->mutex);
+  mPipelineContext->cond.wait(lk, [this] { return this->mPipelineContext->pipelineQueue.empty() && !this->mPipelineContext->workerRunning; });
+}
+
 int32_t GPUReconstruction::EnqueuePipeline(bool terminate)
 {
   ClearAllocatedMemory(true);

GPU/GPUTracking/Base/GPUReconstruction.h

@@ -98,6 +98,11 @@ class GPUReconstruction
   static constexpr GeometryType geometryType = GeometryType::O2;
 #endif
 
+  enum retValValue : uint32_t { ok = 0,
+                                error = 1,
+                                doExit = 2,
+                                nonFatalErrorCode = 3,
+                                abort = 4 };
   static DeviceType GetDeviceType(const char* type);
   enum InOutPointerType : uint32_t { CLUSTER_DATA = 0,
                                      SECTOR_OUT_TRACK = 1,
@@ -159,6 +164,7 @@ class GPUReconstruction
   int32_t CheckErrorCodes(bool cpuOnly = false, bool forceShowErrors = false, std::vector<std::array<uint32_t, 4>>* fillErrors = nullptr);
   void RunPipelineWorker();
   void TerminatePipelineWorker();
+  void DrainPipeline();
 
   // Helpers for memory allocation
   GPUMemoryResource& Res(int16_t num) { return mMemoryResources[num]; }

GPU/GPUTracking/Base/GPUReconstructionCPU.cxx

@@ -245,7 +245,7 @@ int32_t GPUReconstructionCPU::RunChains()
       retVal = mChains[i]->RunChain();
     }
   }
-  if (retVal != 0 && retVal != 2) {
+  if (retVal != GPUReconstruction::retValValue::ok && retVal != GPUReconstruction::retValValue::doExit) {
     return retVal;
   }
   mTimerTotal.Stop();

GPU/GPUTracking/Global/GPUChainTracking.cxx

@@ -677,7 +677,7 @@ int32_t GPUChainTracking::RunChain()
   const bool needQA = GPUQA::QAAvailable() && (GetProcessingSettings().runQA || (GetProcessingSettings().eventDisplay && (mIOPtrs.nMCInfosTPC || GetProcessingSettings().runMC)));
   if (needQA && GetQA()->IsInitialized() == false) {
     if (GetQA()->InitQA(GetProcessingSettings().runQA <= 0 ? -GetProcessingSettings().runQA : gpudatatypes::gpuqa::tasksAutomatic)) {
-      return 1;
+      return GPUReconstruction::retValValue::error;
     }
   }
   if (needQA) {
@@ -693,7 +693,7 @@ int32_t GPUChainTracking::RunChain()
     mRec->PrepareEvent();
   } catch (const std::bad_alloc& e) {
     GPUError("Memory Allocation Error");
-    return (1);
+    return GPUReconstruction::retValValue::error;
   }
   mRec->getGeneralStepTimer(GeneralStep::Prepare).Stop();
 
@@ -707,11 +707,11 @@ int32_t GPUChainTracking::RunChain()
 
   if (mIOPtrs.tpcCompressedClusters) {
     if (runRecoStep(RecoStep::TPCDecompression, &GPUChainTracking::RunTPCDecompression)) {
-      return 1;
+      return GPUReconstruction::retValValue::error;
     }
   } else if (mIOPtrs.tpcPackedDigits || mIOPtrs.tpcZS) {
     if (runRecoStep(RecoStep::TPCClusterFinding, &GPUChainTracking::RunTPCClusterizer, false)) {
-      return 1;
+      return GPUReconstruction::retValValue::error;
     }
   }
 
@@ -720,17 +720,17 @@ int32_t GPUChainTracking::RunChain()
   }
 
   if (mIOPtrs.clustersNative && runRecoStep(RecoStep::TPCConversion, &GPUChainTracking::ConvertNativeToClusterData)) {
-    return 1;
+    return GPUReconstruction::retValValue::error;
   }
 
   mRec->PushNonPersistentMemory(qStr2Tag("TPCSLCD1")); // 1st stack level for TPC tracking sector data
   mTPCSectorScratchOnStack = true;
   if (runRecoStep(RecoStep::TPCSectorTracking, &GPUChainTracking::RunTPCTrackingSectors)) {
-    return 1;
+    return GPUReconstruction::retValValue::error;
   }
 
   if (runRecoStep(RecoStep::TPCMerging, &GPUChainTracking::RunTPCTrackingMerger, false)) {
-    return 1;
+    return GPUReconstruction::retValValue::error;
   }
   if (mTPCSectorScratchOnStack) {
     mRec->PopNonPersistentMemory(RecoStep::TPCSectorTracking, qStr2Tag("TPCSLCD1")); // Release 1st stack level, TPC sector data not needed after merger
@@ -750,16 +750,16 @@ int32_t GPUChainTracking::RunChain()
       }
     }
     if (runRecoStep(RecoStep::TPCCompression, &GPUChainTracking::RunTPCCompression)) {
-      return 1;
+      return GPUReconstruction::retValValue::error;
     }
   }
 
   if (runRecoStep(RecoStep::TRDTracking, &GPUChainTracking::RunTRDTracking)) {
-    return 1;
+    return GPUReconstruction::retValValue::error;
   }
 
   if (runRecoStep(RecoStep::Refit, &GPUChainTracking::RunRefit)) {
-    return 1;
+    return GPUReconstruction::retValValue::error;
   }
 
   if (!GetProcessingSettings().doublePipeline) { // Synchronize with output copies running asynchronously
@@ -770,9 +770,9 @@ int32_t GPUChainTracking::RunChain()
     mRec->SetNActiveThreads(-1);
   }
 
-  int32_t retVal = 0;
+  int32_t retVal = GPUReconstruction::retValValue::ok;
   if (CheckErrorCodes(false, false, mRec->getErrorCodeOutput())) { // TODO: Eventually, we should use GPUReconstruction::CheckErrorCodes
-    retVal = 3;
+    retVal = GPUReconstruction::retValValue::nonFatalErrorCode;
     if (!GetProcessingSettings().ignoreNonFatalGPUErrors) {
       return retVal;
     }
@@ -820,7 +820,7 @@ int32_t GPUChainTracking::RunChainFinalize()
       GPUInfo("Starting Event Display...");
       if (mEventDisplay->StartDisplay()) {
         GPUError("Error starting Event Display");
-        return (1);
+        return GPUReconstruction::retValValue::error;
       }
       mDisplayRunning = true;
     } else {
@@ -857,15 +857,15 @@ int32_t GPUChainTracking::RunChainFinalize()
       mDisplayRunning = false;
       GetProcessingSettings().eventDisplay->DisplayExit();
       const_cast<GPUSettingsProcessing&>(GetProcessingSettings()).eventDisplay = nullptr; // TODO: fixme - eventDisplay should probably not be put into ProcessingSettings in the first place
-      return (2);
+      return GPUReconstruction::retValValue::doExit;
     }
     GetProcessingSettings().eventDisplay->setDisplayControl(0);
     GPUInfo("Loading next event...");
 
     mEventDisplay->BlockTillNextEvent();
   }
 
-  return 0;
+  return GPUReconstruction::retValValue::ok;
 }
 
 int32_t GPUChainTracking::FinalizePipelinedProcessing()

GPU/GPUTracking/Global/GPUChainTracking.h

@@ -195,7 +195,7 @@ class GPUChainTracking : public GPUChain
   void SetCalibObjects(const GPUCalibObjects& obj);
   void SetUpdateCalibObjects(const GPUCalibObjectsConst& obj, const GPUNewCalibValues& vals);
   void SetSubOutputControl(int32_t i, GPUOutputControl* v) { mSubOutputControls[i] = v; }
-  void SetFinalInputCallback(std::function<void()> v) { mWaitForFinalInputs = v; }
+  void SetFinalInputCallback(std::function<int32_t()> v) { mWaitForFinalInputs = v; }
 
   const GPUSettingsDisplay* mConfigDisplay = nullptr; // Abstract pointer to Standalone Display Configuration Structure
   const GPUSettingsQA* mConfigQA = nullptr;           // Abstract pointer to Standalone QA Configuration Structure
@@ -321,7 +321,7 @@ class GPUChainTracking : public GPUChain
   std::mutex mMutexUpdateCalib;
   std::unique_ptr<GPUChainTrackingFinalContext> mPipelineFinalizationCtx;
   GPUChainTrackingFinalContext* mPipelineNotifyCtx = nullptr;
-  std::function<void()> mWaitForFinalInputs;
+  std::function<int32_t()> mWaitForFinalInputs;
 
   int32_t OutputStream() const { return mRec->NStreams() - 2; }
 };

GPU/GPUTracking/Global/GPUChainTrackingClusterizer.cxx

@@ -769,7 +769,7 @@ int32_t GPUChainTracking::RunTPCClusterizer(bool synchronizeOutput)
 #endif
 
   if (RunTPCClusterizer_prepare(mPipelineNotifyCtx && GetProcessingSettings().doublePipelineClusterizer, extraADCs)) {
-    return 1;
+    return GPUReconstruction::retValValue::error;
   }
   if (GetProcessingSettings().autoAdjustHostThreads && !doGPU) {
     mRec->SetNActiveThreads(mRec->MemoryScalers()->nTPCdigits / 6000);
@@ -1471,7 +1471,9 @@ int32_t GPUChainTracking::RunTPCClusterizer(bool synchronizeOutput)
       notifyForeignChainFinished();
     }
     if (mWaitForFinalInputs && iSectorBase >= 30 && (int32_t)iSectorBase < 30 + GetProcessingSettings().nTPCClustererLanes) {
-      mWaitForFinalInputs();
+      if (mWaitForFinalInputs()) {
+        return GPUReconstruction::retValValue::abort;
+      }
       synchronizeCalibUpdate = DoQueuedUpdates(0, false);
     }
   }

GPU/GPUTracking/Interface/GPUO2Interface.cxx

@@ -137,6 +137,11 @@ void GPUO2Interface::Deinitialize()
   mNContexts = 0;
 }
 
+void GPUO2Interface::DrainPipeline()
+{
+  mCtx[0].mRec->DrainPipeline();
+}
+
 void GPUO2Interface::DumpEvent(int32_t nEvent, GPUTrackingInOutPointers* data, uint32_t iThread, const char* dir)
 {
   const auto oldPtrs = mCtx[iThread].mChain->mIOPtrs;
@@ -185,19 +190,23 @@ int32_t GPUO2Interface::RunTracking(GPUTrackingInOutPointers* data, GPUInterface
     }
   };
 
-  auto inputWaitCallback = [this, iThread, inputUpdateCallback, &data, &outputs, &setOutputs]() {
+  auto inputWaitCallback = [this, iThread, inputUpdateCallback, &data, &outputs, &setOutputs]() -> int32_t {
     GPUTrackingInOutPointers* updatedData;
     GPUInterfaceOutputs* updatedOutputs;
+    int32_t retVal = 0;
     if (inputUpdateCallback->callback) {
-      inputUpdateCallback->callback(updatedData, updatedOutputs);
-      mCtx[iThread].mChain->mIOPtrs = *updatedData;
-      outputs = updatedOutputs;
-      data = updatedData;
-      setOutputs(outputs);
+      retVal = inputUpdateCallback->callback(updatedData, updatedOutputs);
+      if (retVal == 0) {
+        mCtx[iThread].mChain->mIOPtrs = *updatedData;
+        outputs = updatedOutputs;
+        data = updatedData;
+        setOutputs(outputs);
+      }
     }
     if (inputUpdateCallback->notifyCallback) {
       inputUpdateCallback->notifyCallback();
     }
+    return retVal;
   };
 
   if (inputUpdateCallback) {
@@ -210,8 +219,8 @@ int32_t GPUO2Interface::RunTracking(GPUTrackingInOutPointers* data, GPUInterface
   }
 
   int32_t retVal = mCtx[iThread].mRec->RunChains();
-  if (retVal == 2) {
-    retVal = 0; // 2 signals end of event display, ignore
+  if (retVal == GPUReconstruction::retValValue::doExit) {
+    retVal = GPUReconstruction::retValValue::ok; // Ignore exit signal from event display
   }
   if (mConfig->configQA.shipToQC && mCtx[iThread].mChain->QARanForTF()) {
     outputs->qa.hist1 = &mCtx[iThread].mChain->GetQA()->getHistograms1D();

GPU/GPUTracking/Interface/GPUO2Interface.h

@@ -71,6 +71,7 @@ class GPUO2Interface
 
   int32_t Initialize(const GPUO2InterfaceConfiguration& config);
   void Deinitialize();
+  void DrainPipeline();
 
   int32_t RunTracking(GPUTrackingInOutPointers* data, GPUInterfaceOutputs* outputs = nullptr, uint32_t iThread = 0, GPUInterfaceInputUpdate* inputUpdateCallback = nullptr);
   void Clear(bool clearOutputs, uint32_t iThread = 0);

GPU/GPUTracking/Interface/GPUO2InterfaceConfiguration.h

@@ -58,8 +58,8 @@ struct GPUInterfaceOutputs : public GPUTrackingOutputs {
 };
 
 struct GPUInterfaceInputUpdate {
-  std::function<void(GPUTrackingInOutPointers*& data, GPUInterfaceOutputs*& outputs)> callback; // Callback which provides final data ptrs / outputRegions after Clusterization stage
-  std::function<void()> notifyCallback;                                                         // Callback called to notify that Clusterization state has finished without update
+  std::function<int32_t(GPUTrackingInOutPointers*& data, GPUInterfaceOutputs*& outputs)> callback; // Callback which provides final data ptrs / outputRegions after Clusterization stage
+  std::function<void()> notifyCallback;                                                            // Callback called to notify that Clusterization state has finished without update
 };
 
 // Full configuration structure with all available settings of GPU...

GPU/GPUTracking/Standalone/Benchmark/standalone.cxx

@@ -685,11 +685,11 @@ int32_t RunBenchmark(GPUReconstruction* recUse, GPUChainTracking* chainTrackingU
       }
     }
 
-    if (tmpRetVal == 0 || tmpRetVal == 2) {
+    if (tmpRetVal == GPUReconstruction::retValValue::ok || tmpRetVal == GPUReconstruction::retValValue::doExit) {
       OutputStat(chainTrackingUse, iRun == 0 ? nTracksTotal : nullptr, iRun == 0 ? nClustersTotal : nullptr);
     }
 
-    if (tmpRetVal == 0 && configStandalone.testSyncAsync) {
+    if (tmpRetVal == GPUReconstruction::retValValue::ok && configStandalone.testSyncAsync) {
       vecpod<char> compressedTmpMem(chainTracking->mIOPtrs.tpcCompressedClusters->totalDataSize);
       memcpy(compressedTmpMem.data(), (const void*)chainTracking->mIOPtrs.tpcCompressedClusters, chainTracking->mIOPtrs.tpcCompressedClusters->totalDataSize);
       o2::tpc::CompressedClusters tmp(*chainTracking->mIOPtrs.tpcCompressedClusters);
@@ -717,7 +717,7 @@ int32_t RunBenchmark(GPUReconstruction* recUse, GPUChainTracking* chainTrackingU
         recAsync->SetResetTimers(iRun < configStandalone.runsInit);
       }
       tmpRetVal = recAsync->RunChains();
-      if (tmpRetVal == 0 || tmpRetVal == 2) {
+      if (tmpRetVal == GPUReconstruction::retValValue::ok || tmpRetVal == GPUReconstruction::retValValue::doExit) {
         OutputStat(chainTrackingAsync, nullptr, nullptr);
       }
       recAsync->ClearAllocatedMemory();
@@ -726,14 +726,14 @@ int32_t RunBenchmark(GPUReconstruction* recUse, GPUChainTracking* chainTrackingU
       recUse->ClearAllocatedMemory();
     }
 
-    if (tmpRetVal == 2) {
+    if (tmpRetVal == GPUReconstruction::retValValue::doExit) {
       configStandalone.continueOnError = 0; // Forced exit from event display loop
       configStandalone.noprompt = 1;
     }
-    if (tmpRetVal == 3 && configStandalone.proc.ignoreNonFatalGPUErrors) {
+    if (tmpRetVal == GPUReconstruction::retValValue::nonFatalErrorCode && configStandalone.proc.ignoreNonFatalGPUErrors) {
       printf("GPU Standalone Benchmark: Non-FATAL GPU error occured, ignoring\n");
     } else if (tmpRetVal && !configStandalone.continueOnError) {
-      if (tmpRetVal != 2) {
+      if (tmpRetVal != GPUReconstruction::retValValue::doExit) {
         printf("GPU Standalone Benchmark: Error occured\n");
       }
       return 1;

GPU/Workflow/src/GPUWorkflowInternal.h

@@ -64,11 +64,12 @@ struct GPURecoWorkflowSpec_PipelineInternals {
   fair::mq::Device* fmqDevice = nullptr;
 
   volatile fair::mq::State fmqState = fair::mq::State::Undefined, fmqPreviousState = fair::mq::State::Undefined;
-  volatile bool endOfStreamAsyncReceived = false;
+  volatile bool endOfStreamAsyncWaiting = true;
   volatile bool endOfStreamDplReceived = false;
   volatile bool runStarted = false;
   volatile bool shouldTerminate = false;
-  std::mutex stateMutex;
+  volatile bool pipelineAbort = false;
+  std::mutex stateMutex, receiveMutex;
   std::condition_variable stateNotify;
 
   std::thread receiveThread;