GPU Workflow: Add dumpFirst and dumpLast options

Author: davidrohr

GPU/GPUTracking/Definitions/GPUSettingsList.h

@@ -625,6 +625,9 @@ AddOption(deviceType, std::string, "CPU", "", 0, "Device type, CPU | CUDA | HIP
 AddOption(forceDeviceType, bool, true, "", 0, "force device type, otherwise allows fall-back to CPU")
 AddOption(synchronousProcessing, bool, false, "", 0, "Apply performance shortcuts for synchronous processing, disable unneeded steps")
 AddOption(dump, int32_t, 0, "", 0, "Dump events for standalone benchmark: 1 = dump events, 2 = dump events and skip processing in workflow")
+AddOption(dumpFirst, int32_t, 0, "", 0, "First event to dump (referring to tfCounter)")
+AddOption(dumpLast, int32_t, -1, "", 0, "Last event to dump (-1 = all)")
+AddOption(dumpFolder, std::string, "", "", 0, "Folder to which to write dump files, [P] is replaced by process id")
 AddOption(display, bool, false, "", 0, "Enable standalone gpu tracking visualizaion")
 AddOption(rundEdx, int32_t, -1, "", 0, "Enable/disable dEdx processing (-1 for autoselect)")
 AddOption(dEdxSplineTopologyCorrFile, std::string, "", "", 0, "File name of the dE/dx spline track topology correction file")

GPU/GPUTracking/Interface/GPUO2Interface.cxx

@@ -137,29 +137,30 @@ void GPUO2Interface::Deinitialize()
   mNContexts = 0;
 }
 
-void GPUO2Interface::DumpEvent(int32_t nEvent, GPUTrackingInOutPointers* data)
+void GPUO2Interface::DumpEvent(int32_t nEvent, GPUTrackingInOutPointers* data, uint32_t iThread, const char* dir)
 {
-  mCtx[0].mChain->ClearIOPointers();
-  mCtx[0].mChain->mIOPtrs = *data;
+  const auto oldPtrs = mCtx[iThread].mChain->mIOPtrs;
+  mCtx[iThread].mChain->mIOPtrs = *data;
   char fname[1024];
-  snprintf(fname, 1024, "event.%d.dump", nEvent);
-  mCtx[0].mChain->DumpData(fname);
+  snprintf(fname, 1024, "%sevent.%d.dump", dir, nEvent);
+  mCtx[iThread].mChain->DumpData(fname);
   if (nEvent == 0) {
 #ifdef GPUCA_BUILD_QA
     if (mConfig->configProcessing.runMC) {
-      mCtx[0].mChain->ForceInitQA();
+      mCtx[iThread].mChain->ForceInitQA();
       snprintf(fname, 1024, "mc.%d.dump", nEvent);
-      mCtx[0].mChain->GetQA()->UpdateChain(mCtx[0].mChain);
-      mCtx[0].mChain->GetQA()->DumpO2MCData(fname);
+      mCtx[iThread].mChain->GetQA()->UpdateChain(mCtx[iThread].mChain);
+      mCtx[iThread].mChain->GetQA()->DumpO2MCData(fname);
     }
 #endif
   }
+  mCtx[iThread].mChain->mIOPtrs = oldPtrs;
 }
 
-void GPUO2Interface::DumpSettings()
+void GPUO2Interface::DumpSettings(uint32_t iThread, const char* dir)
 {
-  mCtx[0].mChain->DoQueuedUpdates(-1);
-  mCtx[0].mRec->DumpSettings();
+  mCtx[iThread].mChain->DoQueuedUpdates(-1);
+  mCtx[iThread].mRec->DumpSettings(dir);
 }
 
 int32_t GPUO2Interface::RunTracking(GPUTrackingInOutPointers* data, GPUInterfaceOutputs* outputs, uint32_t iThread, GPUInterfaceInputUpdate* inputUpdateCallback)

GPU/GPUTracking/Interface/GPUO2Interface.h

@@ -77,8 +77,8 @@ class GPUO2Interface
 
   int32_t RunTracking(GPUTrackingInOutPointers* data, GPUInterfaceOutputs* outputs = nullptr, uint32_t iThread = 0, GPUInterfaceInputUpdate* inputUpdateCallback = nullptr);
   void Clear(bool clearOutputs, uint32_t iThread = 0);
-  void DumpEvent(int32_t nEvent, GPUTrackingInOutPointers* data);
-  void DumpSettings();
+  void DumpEvent(int32_t nEvent, GPUTrackingInOutPointers* data, uint32_t iThread, const char* dir = "");
+  void DumpSettings(uint32_t iThread, const char* dir = "");
 
   void GetITSTraits(o2::its::TrackerTraits<7>*& trackerTraits, o2::its::VertexerTraits<7>*& vertexerTraits, o2::its::TimeFrame<7>*& timeFrame);
   const o2::base::Propagator* GetDeviceO2Propagator(int32_t iThread = 0) const;

GPU/Workflow/include/GPUWorkflow/GPUWorkflowSpec.h

@@ -225,6 +225,7 @@ class GPURecoWorkflowSpec : public o2::framework::Task
   int64_t mCreationForCalib = -1; ///< creation time for calib manipulation
   int32_t mVerbosity = 0;
   uint32_t mNTFs = 0;
+  uint32_t mNTFDumps = 0;
   uint32_t mNDebugDumps = 0;
   uint32_t mNextThreadIndex = 0;
   bool mUpdateGainMapCCDB = true;

GPU/Workflow/src/GPUWorkflowInternal.h

@@ -47,6 +47,7 @@ struct GPURecoWorkflow_QueueObject {
   bool jobSubmitted = false;
   bool jobFinished = false;
   int32_t jobReturnValue = 0;
+  volatile int32_t jobThreadIndex = -1;
   std::mutex jobFinishedMutex;
   std::condition_variable jobFinishedNotify;
   bool jobInputFinal = false;

GPU/Workflow/src/GPUWorkflowPipeline.cxx

@@ -90,6 +90,7 @@ void GPURecoWorkflowSpec::RunWorkerThread(int32_t id)
       context = workerContext.inputQueue.front();
       workerContext.inputQueue.pop();
     }
+    context->jobThreadIndex = id;
     context->jobReturnValue = runMain(nullptr, context->jobPtrs, context->jobOutputRegions, id, context->jobInputUpdateCallback.get());
     {
       std::lock_guard lk(context->jobFinishedMutex);
@@ -179,8 +180,7 @@ int32_t GPURecoWorkflowSpec::handlePipeline(ProcessingContext& pc, GPUTrackingIn
       }
       mPipeline->completionPolicyQueue.pop();
     }
-  }
-  if (mSpecConfig.enableDoublePipeline == 2) {
+  } else if (mSpecConfig.enableDoublePipeline == 2) {
     auto prepareDummyMessage = pc.outputs().make<DataAllocator::UninitializedVector<char>>(Output{gDataOriginGPU, "PIPELINEPREPARE", 0}, 0u);
 
     size_t ptrsTotal = 0;

GPU/Workflow/src/GPUWorkflowSpec.cxx

@@ -825,11 +825,31 @@ void GPURecoWorkflowSpec::run(ProcessingContext& pc)
 
   lockDecodeInput.reset();
 
+  uint32_t threadIndex;
   if (mConfParam->dump) {
-    if (mNTFs == 1) {
-      mGPUReco->DumpSettings();
+    if (mSpecConfig.enableDoublePipeline && pipelineContext->jobSubmitted) {
+      while (pipelineContext->jobThreadIndex == -1) {
+      }
+      threadIndex = pipelineContext->jobThreadIndex;
+    } else {
+      threadIndex = 0; // TODO: Not sure if this is safe, but it is not yet known which threadIndex will pick up the enqueued job
+    }
+
+    std::string dir = "";
+    if (mConfParam->dumpFolder != "") {
+      dir = std::regex_replace(mConfParam->dumpFolder, std::regex("\\[P\\]"), std::to_string(getpid()));
+      if (mNTFs == 1) {
+        mkdir(dir.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
+      }
+      dir += "/";
+    }
+    if (mNTFs == 1) { // Must dump with first TF, since will enforce enqueued calib updates
+      mGPUReco->DumpSettings(threadIndex, dir.c_str());
+    }
+    if (tinfo.tfCounter >= mConfParam->dumpFirst && (mConfParam->dumpLast == -1 || tinfo.tfCounter <= mConfParam->dumpLast)) {
+      mGPUReco->DumpEvent(mNTFDumps, &ptrs, threadIndex, dir.c_str());
+      mNTFDumps++;
     }
-    mGPUReco->DumpEvent(mNTFs - 1, &ptrs);
   }
   std::unique_ptr<GPUTrackingInOutPointers> ptrsDump;
   if (mConfParam->dumpBadTFMode == 2) {
@@ -847,9 +867,10 @@ void GPURecoWorkflowSpec::run(ProcessingContext& pc)
     std::unique_lock lk(pipelineContext->jobFinishedMutex);
     pipelineContext->jobFinishedNotify.wait(lk, [context = pipelineContext.get()]() { return context->jobFinished; });
     retVal = pipelineContext->jobReturnValue;
+    threadIndex = pipelineContext->jobThreadIndex;
   } else {
     // uint32_t threadIndex = pc.services().get<ThreadPool>().threadIndex;
-    uint32_t threadIndex = mNextThreadIndex;
+    threadIndex = mNextThreadIndex;
     if (mConfig->configProcessing.doublePipeline) {
       mNextThreadIndex = (mNextThreadIndex + 1) % 2;
     }
@@ -879,7 +900,7 @@ void GPURecoWorkflowSpec::run(ProcessingContext& pc)
       }
       fclose(fp);
     } else if (mConfParam->dumpBadTFMode == 2) {
-      mGPUReco->DumpEvent(mNDebugDumps - 1, ptrsDump.get());
+      mGPUReco->DumpEvent(mNDebugDumps - 1, ptrsDump.get(), threadIndex);
     }
   }
 

prodtests/full-system-test/dpl-workflow.sh

@@ -235,6 +235,9 @@ if [[ $EPNSYNCMODE == 1 ]]; then
     fi
   fi
 fi
+if [[ $GPUTYPE != "CPU" && $NGPUS > 1 ]]; then
+  GPU_CONFIG_KEY+="GPU_global.dumpFolder=gpu_dump_[P];"
+fi
 if [[ $SYNCRAWMODE == 1 ]]; then
   GPU_CONFIG_KEY+="GPU_proc.tpcIncreasedMinClustersPerRow=500000;GPU_proc.ignoreNonFatalGPUErrors=1;GPU_proc.throttleAlarms=1;"
   if [[ $RUNTYPE == "PHYSICS" || $RUNTYPE == "COSMICS" || $RUNTYPE == "TECHNICAL" ]]; then