diff --git a/GPU/GPUTracking/Base/GPUReconstruction.cxx b/GPU/GPUTracking/Base/GPUReconstruction.cxx
index ac67c617872bd..eaf3afbdc9cf3 100644
--- a/GPU/GPUTracking/Base/GPUReconstruction.cxx
+++ b/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);
diff --git a/GPU/GPUTracking/Base/GPUReconstruction.h b/GPU/GPUTracking/Base/GPUReconstruction.h
index 4479eb696808e..d85c29371f8fb 100644
--- a/GPU/GPUTracking/Base/GPUReconstruction.h
+++ b/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]; }
diff --git a/GPU/GPUTracking/Base/GPUReconstructionCPU.cxx b/GPU/GPUTracking/Base/GPUReconstructionCPU.cxx
index 9fbe9e1171af3..ac2c2d0a78a78 100644
--- a/GPU/GPUTracking/Base/GPUReconstructionCPU.cxx
+++ b/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();
diff --git a/GPU/GPUTracking/Global/GPUChainTracking.cxx b/GPU/GPUTracking/Global/GPUChainTracking.cxx
index 067c4dff92810..480ec76408584 100644
--- a/GPU/GPUTracking/Global/GPUChainTracking.cxx
+++ b/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,7 +857,7 @@ 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...");
@@ -865,7 +865,7 @@ int32_t GPUChainTracking::RunChainFinalize()
     mEventDisplay->BlockTillNextEvent();
   }
 
-  return 0;
+  return GPUReconstruction::retValValue::ok;
 }
 
 int32_t GPUChainTracking::FinalizePipelinedProcessing()
diff --git a/GPU/GPUTracking/Global/GPUChainTracking.h b/GPU/GPUTracking/Global/GPUChainTracking.h
index 3c3531530372a..759aaf818028e 100644
--- a/GPU/GPUTracking/Global/GPUChainTracking.h
+++ b/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; }
 };
diff --git a/GPU/GPUTracking/Global/GPUChainTrackingClusterizer.cxx b/GPU/GPUTracking/Global/GPUChainTrackingClusterizer.cxx
index 681914414d401..a558ed85c5516 100644
--- a/GPU/GPUTracking/Global/GPUChainTrackingClusterizer.cxx
+++ b/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);
     }
   }
diff --git a/GPU/GPUTracking/Interface/GPUO2Interface.cxx b/GPU/GPUTracking/Interface/GPUO2Interface.cxx
index ced3016dc15b1..184597b12ceba 100644
--- a/GPU/GPUTracking/Interface/GPUO2Interface.cxx
+++ b/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();
diff --git a/GPU/GPUTracking/Interface/GPUO2Interface.h b/GPU/GPUTracking/Interface/GPUO2Interface.h
index ca56018908b41..eed7c119f5328 100644
--- a/GPU/GPUTracking/Interface/GPUO2Interface.h
+++ b/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);
diff --git a/GPU/GPUTracking/Interface/GPUO2InterfaceConfiguration.h b/GPU/GPUTracking/Interface/GPUO2InterfaceConfiguration.h
index 0f8a3784f0a88..155048859a507 100644
--- a/GPU/GPUTracking/Interface/GPUO2InterfaceConfiguration.h
+++ b/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...
diff --git a/GPU/GPUTracking/Standalone/Benchmark/standalone.cxx b/GPU/GPUTracking/Standalone/Benchmark/standalone.cxx
index cfa3ad7c81b83..c452e973d97e0 100644
--- a/GPU/GPUTracking/Standalone/Benchmark/standalone.cxx
+++ b/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;
diff --git a/GPU/Workflow/src/GPUWorkflowInternal.h b/GPU/Workflow/src/GPUWorkflowInternal.h
index 1ad6f3df13f5a..aa6ab80af576b 100644
--- a/GPU/Workflow/src/GPUWorkflowInternal.h
+++ b/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;
diff --git a/GPU/Workflow/src/GPUWorkflowPipeline.cxx b/GPU/Workflow/src/GPUWorkflowPipeline.cxx
index f0aeb8089e27a..305772331abb3 100644
--- a/GPU/Workflow/src/GPUWorkflowPipeline.cxx
+++ b/GPU/Workflow/src/GPUWorkflowPipeline.cxx
@@ -116,11 +116,15 @@ void GPURecoWorkflowSpec::enqueuePipelinedJob(GPUTrackingInOutPointers* ptrs, GP
   context->jobInputUpdateCallback = std::make_unique<GPUInterfaceInputUpdate>();
 
   if (!inputFinal) {
-    context->jobInputUpdateCallback->callback = [context](GPUTrackingInOutPointers*& data, GPUInterfaceOutputs*& outputs) {
+    context->jobInputUpdateCallback->callback = [context, this](GPUTrackingInOutPointers*& data, GPUInterfaceOutputs*& outputs) -> int32_t {
       std::unique_lock lk(context->jobInputFinalMutex);
-      context->jobInputFinalNotify.wait(lk, [context]() { return context->jobInputFinal; });
+      context->jobInputFinalNotify.wait(lk, [context, this]() { return context->jobInputFinal || mPipeline->pipelineAbort; });
+      if (mPipeline->pipelineAbort) {
+        return 1;
+      }
       data = context->jobPtrs;
       outputs = context->jobOutputRegions;
+      return 0;
     };
   }
   context->jobInputUpdateCallback->notifyCallback = [this]() {
@@ -195,15 +199,17 @@ int32_t GPURecoWorkflowSpec::handlePipeline(ProcessingContext& pc, GPUTrackingIn
     }
 
     size_t prepareBufferSize = sizeof(pipelinePrepareMessage) + ptrsTotal * sizeof(size_t) * 4;
-    std::vector<size_t> messageBuffer(prepareBufferSize / sizeof(size_t));
-    pipelinePrepareMessage& preMessage = *(pipelinePrepareMessage*)messageBuffer.data();
+    fair::mq::MessagePtr payload(device->NewMessage());
+    payload->Rebuild(prepareBufferSize, fair::mq::Alignment(sizeof(size_t)));
+    auto* messageBuffer = (size_t*)payload->GetData();
+    pipelinePrepareMessage& preMessage = *(pipelinePrepareMessage*)messageBuffer;
     preMessage.magicWord = preMessage.MAGIC_WORD;
     preMessage.timeSliceId = tinfo.timeslice;
     preMessage.pointersTotal = ptrsTotal;
     preMessage.flagEndOfStream = false;
     memcpy((void*)&preMessage.tfSettings, (const void*)ptrs.settingsTF, sizeof(preMessage.tfSettings));
 
-    size_t* ptrBuffer = messageBuffer.data() + sizeof(preMessage) / sizeof(size_t);
+    size_t* ptrBuffer = messageBuffer + sizeof(preMessage) / sizeof(size_t);
     size_t ptrsCopied = 0;
     int32_t lastRegion = -1;
     for (uint32_t i = 0; i < GPUTrackingInOutZS::NSECTORS; i++) {
@@ -234,9 +240,7 @@ int32_t GPURecoWorkflowSpec::handlePipeline(ProcessingContext& pc, GPUTrackingIn
     }
 
     auto channel = device->GetChannels().find("gpu-prepare-channel");
-    fair::mq::MessagePtr payload(device->NewMessage());
     LOG(info) << "Sending gpu-reco-workflow prepare message of size " << prepareBufferSize;
-    payload->Rebuild(messageBuffer.data(), prepareBufferSize, nullptr, nullptr);
     channel->second[0].Send(payload);
     return 2;
   }
@@ -251,12 +255,13 @@ void GPURecoWorkflowSpec::handlePipelineEndOfStream(EndOfStreamContext& ec)
   }
   if (mSpecConfig.enableDoublePipeline == 2) {
     auto* device = ec.services().get<RawDeviceService>().device();
-    pipelinePrepareMessage preMessage;
-    preMessage.flagEndOfStream = true;
-    auto channel = device->GetChannels().find("gpu-prepare-channel");
     fair::mq::MessagePtr payload(device->NewMessage());
+    payload->Rebuild(sizeof(pipelinePrepareMessage), fair::mq::Alignment(alignof(pipelinePrepareMessage)));
+    auto* preMessage = (pipelinePrepareMessage*)payload->GetData();
+    new (preMessage) pipelinePrepareMessage;
+    preMessage->flagEndOfStream = true;
+    auto channel = device->GetChannels().find("gpu-prepare-channel");
     LOG(info) << "Sending end-of-stream message over out-of-bands channel";
-    payload->Rebuild(&preMessage, sizeof(preMessage), nullptr, nullptr);
     channel->second[0].Send(payload);
   }
 }
@@ -264,7 +269,33 @@ void GPURecoWorkflowSpec::handlePipelineEndOfStream(EndOfStreamContext& ec)
 void GPURecoWorkflowSpec::handlePipelineStop()
 {
   if (mSpecConfig.enableDoublePipeline == 1) {
-    mPipeline->mayInjectTFId = 0;
+    {
+      std::unique_lock lk(mPipeline->queueMutex);
+      mPipeline->pipelineAbort = mPipeline->pipelineQueue.size();
+    }
+    if (mPipeline->pipelineAbort) {
+      mPipeline->pipelineQueue.front()->jobInputFinalNotify.notify_one();
+      mGPUReco->DrainPipeline();
+      {
+        std::unique_lock lk(mPipeline->queueMutex);
+        mPipeline->pipelineQueue = {};
+      }
+      {
+        std::lock_guard lk(mPipeline->completionPolicyMutex);
+        mPipeline->completionPolicyQueue = {};
+      }
+      mPipeline->pipelineAbort = false;
+      {
+        std::lock_guard lk(mPipeline->stateMutex);
+        mPipeline->endOfStreamAsyncWaiting = false;
+        mPipeline->mNTFReceived = 0;
+        mPipeline->runStarted = false;
+      }
+    }
+    {
+      std::unique_lock lk(mPipeline->mayInjectMutex);
+      mPipeline->mayInjectTFId = 0;
+    }
   }
 }
 
@@ -273,16 +304,19 @@ 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->endOfStreamAsyncReceived = false;
+      mPipeline->endOfStreamAsyncWaiting = true;
       mPipeline->endOfStreamDplReceived = false;
     }
     mPipeline->fmqPreviousState = mPipeline->fmqState;
     mPipeline->fmqState = newState;
+  }
+  mPipeline->stateNotify.notify_all();
+  {
+    std::lock_guard lk(mPipeline->receiveMutex);
     if (newState == fair::mq::State::Exiting) {
       mPipeline->fmqDevice->UnsubscribeFromStateChange(GPURecoWorkflowSpec_FMQCallbackKey);
     }
   }
-  mPipeline->stateNotify.notify_all();
 }
 
 void GPURecoWorkflowSpec::RunReceiveThread()
@@ -293,7 +327,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     int32_t 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); };
+    auto shouldReceive = [this]() { return ((mPipeline->fmqState == fair::mq::State::Running || (mPipeline->fmqState == fair::mq::State::Ready && mPipeline->fmqPreviousState == fair::mq::State::Running)) && mPipeline->endOfStreamAsyncWaiting); };
     do {
       {
         std::unique_lock lk(mPipeline->stateMutex);
@@ -304,7 +338,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
       }
       try {
         do {
-          std::unique_lock lk(mPipeline->stateMutex);
+          std::unique_lock lk(mPipeline->receiveMutex);
           if (!shouldReceive()) {
             break;
           }
@@ -327,10 +361,13 @@ void GPURecoWorkflowSpec::RunReceiveThread()
     }
     if (m->flagEndOfStream) {
       LOG(info) << "Received end-of-stream from out-of-band channel";
-      std::lock_guard lk(mPipeline->stateMutex);
-      mPipeline->endOfStreamAsyncReceived = true;
-      mPipeline->mNTFReceived = 0;
-      mPipeline->runStarted = false;
+      {
+        std::lock_guard lk(mPipeline->stateMutex);
+        mPipeline->endOfStreamAsyncWaiting = false;
+        mPipeline->mNTFReceived = 0;
+        mPipeline->runStarted = false;
+      }
+      mPipeline->stateNotify.notify_all();
       continue;
     }
 
@@ -342,7 +379,7 @@ void GPURecoWorkflowSpec::RunReceiveThread()
 
     {
       std::unique_lock lk(mPipeline->stateMutex);
-      mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->runStarted && !mPipeline->endOfStreamAsyncReceived) || mPipeline->shouldTerminate; });
+      mPipeline->stateNotify.wait(lk, [this]() { return (mPipeline->runStarted && mPipeline->endOfStreamAsyncWaiting) || mPipeline->shouldTerminate; });
       if (!mPipeline->runStarted) {
         continue;
       }