GPU SA Multi-Threading: Distinguish between EoS from DPL and from asy…

…nc out-of-band channel
bazinski · Oct 23, 2023 · ced8b44 · ced8b44
1 parent 9892a91
commit ced8b44
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/GPU/Workflow/src/GPUWorkflowInternal.h b/GPU/Workflow/src/GPUWorkflowInternal.h
@@ -63,7 +63,8 @@ struct GPURecoWorkflowSpec_PipelineInternals {
   fair::mq::Device* fmqDevice = nullptr;
 
   fair::mq::State fmqState = fair::mq::State::Undefined;
-  volatile bool endOfStreamReceived = false;
+  volatile bool endOfStreamAsyncReceived = false;
+  volatile bool endOfStreamDplReceived = false;
   volatile bool runStarted = false;
   volatile bool shouldTerminate = false;
   std::mutex stateMutex;

diff --git a/GPU/Workflow/src/GPUWorkflowPipeline.cxx b/GPU/Workflow/src/GPUWorkflowPipeline.cxx
@@ -242,7 +242,7 @@ int GPURecoWorkflowSpec::handlePipeline(ProcessingContext& pc, GPUTrackingInOutP
 void GPURecoWorkflowSpec::handlePipelineEndOfStream(EndOfStreamContext& ec)
 {
   if (mSpecConfig.enableDoublePipeline == 1) {
-    mPipeline->endOfStreamReceived = true;
+    mPipeline->endOfStreamDplReceived = true;
     mPipeline->stateNotify.notify_all();
   }
   if (mSpecConfig.enableDoublePipeline == 2) {
@@ -262,7 +262,8 @@ void GPURecoWorkflowSpec::receiveFMQStateCallback(fair::mq::State newState)
   {
     std::lock_guard lk(mPipeline->stateMutex);
     if (mPipeline->fmqState != fair::mq::State::Running && newState == fair::mq::State::Running) {
-      mPipeline->endOfStreamReceived = false;
+      mPipeline->endOfStreamAsyncReceived = false;
+      mPipeline->endOfStreamDplReceived = false;
     }
     mPipeline->fmqState = newState;
     if (newState == fair::mq::State::Exiting) {
@@ -283,7 +284,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     do {
       {
         std::unique_lock lk(mPipeline->stateMutex);
-        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!
+        mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->fmqState == fair::mq::State::Running && !mPipeline->endOfStreamAsyncReceived) || mPipeline->shouldTerminate; }); // Do not check mPipeline->fmqDevice->NewStatePending() since we wait for EndOfStream!
       }
       if (mPipeline->shouldTerminate) {
         break;
@@ -310,7 +311,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     if (m->flagEndOfStream) {
       LOG(info) << "Received end-of-stream from out-of-band channel";
       std::lock_guard lk(mPipeline->stateMutex);
-      mPipeline->endOfStreamReceived = true;
+      mPipeline->endOfStreamAsyncReceived = true;
       mPipeline->mNTFReceived = 0;
       mPipeline->runStarted = false;
       continue;
@@ -324,7 +325,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
 
     {
       std::unique_lock lk(mPipeline->stateMutex);
-      mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->runStarted && !mPipeline->endOfStreamReceived) || mPipeline->shouldTerminate; });
+      mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->runStarted && !mPipeline->endOfStreamAsyncReceived) || mPipeline->shouldTerminate; });
       if (!mPipeline->runStarted) {
         continue;
       }