GPU: Delay only processing of async message not receiving, to fill completionPolicyQueue in time and do not deadlock

Author: davidrohr

GPU/Workflow/src/GPUWorkflowInternal.h

@@ -84,7 +84,7 @@ struct GPURecoWorkflowSpec_PipelineInternals {
   std::condition_variable queueNotify;
 
   std::queue<o2::framework::DataProcessingHeader::StartTime> completionPolicyQueue;
-  bool pipelineSenderTerminating = false;
+  volatile bool pipelineSenderTerminating = false;
   std::mutex completionPolicyMutex;
   std::condition_variable completionPolicyNotify;
 

GPU/Workflow/src/GPUWorkflowPipeline.cxx

@@ -283,7 +283,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     do {
       {
         std::unique_lock lk(mPipeline->stateMutex);
-        mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->runStarted && mPipeline->fmqState == fair::mq::State::Running && !mPipeline->endOfStreamReceived) || mPipeline->shouldTerminate; }); // Do not check mPipeline->fmqDevice->NewStatePending() since we wait for EndOfStream!
+        mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->fmqState == fair::mq::State::Running && !mPipeline->endOfStreamReceived) || mPipeline->shouldTerminate; }); // Do not check mPipeline->fmqDevice->NewStatePending() since we wait for EndOfStream!
       }
       if (mPipeline->shouldTerminate) {
         break;
@@ -316,6 +316,20 @@ void GPURecoWorkflowSpec::RunReceiveThread()
       continue;
     }
 
+    {
+      std::lock_guard lk(mPipeline->completionPolicyMutex);
+      mPipeline->completionPolicyQueue.emplace(m->timeSliceId);
+    }
+    mPipeline->completionPolicyNotify.notify_one();
+
+    {
+      std::unique_lock lk(mPipeline->stateMutex);
+      mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->runStarted && !mPipeline->endOfStreamReceived) || mPipeline->shouldTerminate; });
+      if (!mPipeline->runStarted) {
+        continue;
+      }
+    }
+
     auto context = std::make_unique<GPURecoWorkflow_QueueObject>();
     context->timeSliceId = m->timeSliceId;
     context->tfSettings = m->tfSettings;
@@ -358,21 +372,13 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     if (mPipeline->mNTFReceived++ >= mPipeline->workers.size()) { // Do not inject the first workers.size() TFs, since we need a first round of calib updates from DPL before starting
       enqueuePipelinedJob(&context->ptrs, nullptr, context.get(), false);
     }
-    {
-      std::lock_guard lk(mPipeline->completionPolicyMutex);
-      mPipeline->completionPolicyQueue.emplace(m->timeSliceId);
-    }
-    mPipeline->completionPolicyNotify.notify_one();
     {
       std::lock_guard lk(mPipeline->queueMutex);
       mPipeline->pipelineQueue.emplace(std::move(context));
     }
     mPipeline->queueNotify.notify_one();
   }
-  {
-    std::lock_guard lk(mPipeline->completionPolicyMutex);
-    mPipeline->pipelineSenderTerminating = true;
-  }
+  mPipeline->pipelineSenderTerminating = true;
   mPipeline->completionPolicyNotify.notify_one();
 }