GPU SA Multi-Threading: Make sure we do not try to receive while not …

…in Running or Stopping
bazinski · Oct 23, 2023 · 0f125fe · 0f125fe
1 parent ced8b44
commit 0f125fe
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 5 deletions.
diff --git a/GPU/Workflow/src/GPUWorkflowInternal.h b/GPU/Workflow/src/GPUWorkflowInternal.h
@@ -62,7 +62,7 @@ struct GPURecoWorkflowSpec_PipelineInternals {
 
   fair::mq::Device* fmqDevice = nullptr;
 
-  fair::mq::State fmqState = fair::mq::State::Undefined;
+  volatile fair::mq::State fmqState = fair::mq::State::Undefined, fmqPreviousState = fair::mq::State::Undefined;
   volatile bool endOfStreamAsyncReceived = false;
   volatile bool endOfStreamDplReceived = false;
   volatile bool runStarted = false;
@@ -91,8 +91,8 @@ struct GPURecoWorkflowSpec_PipelineInternals {
 
   unsigned long mNTFReceived = 0;
 
-  bool mayInject = true;
-  unsigned long mayInjectTFId = 0;
+  volatile bool mayInject = true;
+  volatile unsigned long mayInjectTFId = 0;
   std::mutex mayInjectMutex;
   std::condition_variable mayInjectCondition;
 };

diff --git a/GPU/Workflow/src/GPUWorkflowPipeline.cxx b/GPU/Workflow/src/GPUWorkflowPipeline.cxx
@@ -265,6 +265,7 @@ void GPURecoWorkflowSpec::receiveFMQStateCallback(fair::mq::State newState)
       mPipeline->endOfStreamAsyncReceived = false;
       mPipeline->endOfStreamDplReceived = false;
     }
+    mPipeline->fmqPreviousState = mPipeline->fmqState;
     mPipeline->fmqState = newState;
     if (newState == fair::mq::State::Exiting) {
       mPipeline->fmqDevice->UnsubscribeFromStateChange(GPURecoWorkflowSpec_FMQCallbackKey);
@@ -281,17 +282,22 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     int recvTimeot = 1000;
     fair::mq::MessagePtr msg;
     LOG(debug) << "Waiting for out of band message";
+    auto shouldReceive = [this]() { return ((mPipeline->fmqState == fair::mq::State::Running || (mPipeline->fmqState == fair::mq::State::Ready && mPipeline->fmqPreviousState == fair::mq::State::Running)) && !mPipeline->endOfStreamAsyncReceived); };
     do {
       {
         std::unique_lock lk(mPipeline->stateMutex);
-        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!
+        mPipeline->stateNotify.wait(lk, [this, shouldReceive]() { return shouldReceive() || mPipeline->shouldTerminate; }); // Do not check mPipeline->fmqDevice->NewStatePending() since we wait for EndOfStream!
       }
       if (mPipeline->shouldTerminate) {
         break;
       }
       try {
-        msg = device->NewMessageFor("gpu-prepare-channel", 0, 0);
         do {
+          std::unique_lock lk(mPipeline->stateMutex);
+          if (!shouldReceive()) {
+            break;
+          }
+          msg = device->NewMessageFor("gpu-prepare-channel", 0, 0);
           received = device->Receive(msg, "gpu-prepare-channel", 0, recvTimeot) > 0;
         } while (!received && !mPipeline->shouldTerminate);
       } catch (...) {