From d95480255dfcf05b452192e09d5f7dd28e1e6712 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Tue, 4 Aug 2026 09:10:12 +0200 Subject: [PATCH 1/5] Add MessageContext message discarding Allow exception teardown paths to discard pending output messages without asserting that they were sent. --- Framework/Core/include/Framework/MessageContext.h | 5 +++++ Framework/Core/src/MessageContext.cxx | 7 +++++++ 2 files changed, 12 insertions(+) diff --git a/Framework/Core/include/Framework/MessageContext.h b/Framework/Core/include/Framework/MessageContext.h index 407bac0ceb00a..fce4b109a465e 100644 --- a/Framework/Core/include/Framework/MessageContext.h +++ b/Framework/Core/include/Framework/MessageContext.h @@ -466,6 +466,11 @@ class MessageContext /// discarded. void clear(); + /// Discard pending output messages without asserting that they were sent. This + /// is intended for exception teardown paths where normal post-processing will + /// not run. + void discard(); + FairMQDeviceProxy& proxy() { return mProxy; diff --git a/Framework/Core/src/MessageContext.cxx b/Framework/Core/src/MessageContext.cxx index 59dfc15837210..81bb3c7bb95c4 100644 --- a/Framework/Core/src/MessageContext.cxx +++ b/Framework/Core/src/MessageContext.cxx @@ -107,6 +107,13 @@ void MessageContext::clear() mMessages.clear(); } +void MessageContext::discard() +{ + mDidDispatch = false; + mScheduledMessages.clear(); + mMessages.clear(); +} + int64_t MessageContext::addToCache(std::unique_ptr& toCache) { auto&& cached = toCache->GetTransport()->CreateMessage(); From 98399e0183020910f980148e3c4bd27e1a270a1d Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Fri, 7 Aug 2026 14:05:48 +0200 Subject: [PATCH 2/5] Track message state as enum --- Framework/Core/include/Framework/MessageContext.h | 13 ++++++++++--- Framework/Core/src/CommonServices.cxx | 4 ++++ Framework/Core/src/MessageContext.cxx | 8 ++++---- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/Framework/Core/include/Framework/MessageContext.h b/Framework/Core/include/Framework/MessageContext.h index fce4b109a465e..4573f8e8bdc20 100644 --- a/Framework/Core/include/Framework/MessageContext.h +++ b/Framework/Core/include/Framework/MessageContext.h @@ -53,6 +53,12 @@ struct Output; class MessageContext { public: + enum class DispatchState { + NotDispatched, + Dispatched, + Discarded, + }; + constexpr static ServiceKind service_kind = ServiceKind::Stream; // so far we are only using one instance per named channel @@ -495,8 +501,9 @@ class MessageContext o2::header::DataHeader* findMessageHeader(const Output& spec); o2::header::Stack* findMessageHeaderStack(const Output& spec); [[nodiscard]] int countDeviceOutputs(bool excludeDPLOrigin = false) const; - void fakeDispatch() { mDidDispatch = true; } - bool didDispatch() { return mDidDispatch; } + void fakeDispatch() { mDispatchState = DispatchState::Dispatched; } + [[nodiscard]] bool didDispatch() const { return mDispatchState == DispatchState::Dispatched; } + [[nodiscard]] DispatchState dispatchState() const { return mDispatchState; } o2::framework::DataProcessingHeader* findMessageDataProcessingHeader(const Output& spec); std::pair findMessageHeaders(const Output& spec); @@ -504,7 +511,7 @@ class MessageContext FairMQDeviceProxy& mProxy; Messages mMessages; Messages mScheduledMessages; - bool mDidDispatch = false; + DispatchState mDispatchState = DispatchState::NotDispatched; DispatchControl mDispatchControl; /// Cached messages, in case we want to reuse them. std::unordered_map> mMessageCache; diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx index 2cdd046dedc34..ae3471546ccf0 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -185,6 +185,10 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec() auto& routes = processingContext.services().get().outputs; auto& timeslice = processingContext.services().get().timeslice; auto& messageContext = processingContext.services().get(); + // Do not report discarded messages as missing outputs. + if (messageContext.dispatchState() == MessageContext::DispatchState::Discarded) { + return; + } // Check if we never created any data for this timeslice // if we did not, but we still have didDispatched set to true // it means it was created out of band. diff --git a/Framework/Core/src/MessageContext.cxx b/Framework/Core/src/MessageContext.cxx index 81bb3c7bb95c4..db59e743aa8b8 100644 --- a/Framework/Core/src/MessageContext.cxx +++ b/Framework/Core/src/MessageContext.cxx @@ -84,7 +84,7 @@ int MessageContext::countDeviceOutputs(bool excludeDPLOrigin) const { // If we dispatched some messages before the end of the callback // we need to account for them as well. - int noutputs = mDidDispatch ? 1 : 0; + int noutputs = mDispatchState == DispatchState::Dispatched ? 1 : 0; constexpr o2::header::DataOrigin DataOriginDPL{"DPL"}; for (auto it = mMessages.rbegin(); it != mMessages.rend(); ++it) { if (!excludeDPLOrigin || (*it)->header()->dataOrigin != DataOriginDPL) { @@ -103,13 +103,13 @@ void MessageContext::clear() { // Verify that everything has been sent on clear. assert(std::all_of(mMessages.begin(), mMessages.end(), [](auto& m) { return m->empty(); })); - mDidDispatch = false; + mDispatchState = DispatchState::NotDispatched; mMessages.clear(); } void MessageContext::discard() { - mDidDispatch = false; + mDispatchState = DispatchState::Discarded; mScheduledMessages.clear(); mMessages.clear(); } @@ -164,7 +164,7 @@ void MessageContext::schedule(Messages::value_type&& message) } mDispatchControl.dispatch(std::move(parts), ChannelIndex{ci}, DefaultChannelIndex); } - mDidDispatch = mScheduledMessages.empty() == false; + mDispatchState = DispatchState::Dispatched; mScheduledMessages.clear(); } } From 308118cbedff08b69d9ad9cadc26131fbb30b4ac Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Fri, 7 Aug 2026 14:21:00 +0200 Subject: [PATCH 3/5] Add signpost for discarded messages --- Framework/Core/src/CommonServices.cxx | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx index ae3471546ccf0..83b0e45df2f4e 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -185,15 +185,16 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec() auto& routes = processingContext.services().get().outputs; auto& timeslice = processingContext.services().get().timeslice; auto& messageContext = processingContext.services().get(); + O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service); // Do not report discarded messages as missing outputs. if (messageContext.dispatchState() == MessageContext::DispatchState::Discarded) { + O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "Output messages discarded."); return; } // Check if we never created any data for this timeslice // if we did not, but we still have didDispatched set to true // it means it was created out of band. bool userDidCreate = false; - O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service); for (size_t ri = 0; ri < routes.size(); ++ri) { if (stream->routeCreated[ri] == true && stream->routeDPLCreated[ri] == false) { userDidCreate = true; From 2d5b1df8200eecb5e0bdbd3f1c3c1cc199803742 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Fri, 7 Aug 2026 14:53:59 +0200 Subject: [PATCH 4/5] Switch to error signpost for discarded messages --- Framework/Core/src/CommonServices.cxx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx index 83b0e45df2f4e..8a45fc631f3fe 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -188,7 +188,7 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec() O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service); // Do not report discarded messages as missing outputs. if (messageContext.dispatchState() == MessageContext::DispatchState::Discarded) { - O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "Output messages discarded."); + O2_SIGNPOST_EVENT_EMIT_ERROR(stream_context, cid, "postProcessingCallbacks", "Output messages discarded."); return; } // Check if we never created any data for this timeslice From c8f7184655a29f6f06f15afdb56630177b1fcf8f Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Fri, 7 Aug 2026 15:39:22 +0200 Subject: [PATCH 5/5] Replace didDispatch checks with explicit dispatch state --- Framework/Core/include/Framework/MessageContext.h | 1 - Framework/Core/src/CommonServices.cxx | 13 +++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/Framework/Core/include/Framework/MessageContext.h b/Framework/Core/include/Framework/MessageContext.h index 4573f8e8bdc20..dcf3433120fc5 100644 --- a/Framework/Core/include/Framework/MessageContext.h +++ b/Framework/Core/include/Framework/MessageContext.h @@ -502,7 +502,6 @@ class MessageContext o2::header::Stack* findMessageHeaderStack(const Output& spec); [[nodiscard]] int countDeviceOutputs(bool excludeDPLOrigin = false) const; void fakeDispatch() { mDispatchState = DispatchState::Dispatched; } - [[nodiscard]] bool didDispatch() const { return mDispatchState == DispatchState::Dispatched; } [[nodiscard]] DispatchState dispatchState() const { return mDispatchState; } o2::framework::DataProcessingHeader* findMessageDataProcessingHeader(const Output& spec); std::pair findMessageHeaders(const Output& spec); diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx index 8a45fc631f3fe..2ac9dab40d20a 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -185,14 +185,15 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec() auto& routes = processingContext.services().get().outputs; auto& timeslice = processingContext.services().get().timeslice; auto& messageContext = processingContext.services().get(); + auto dispatchState = messageContext.dispatchState(); O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service); // Do not report discarded messages as missing outputs. - if (messageContext.dispatchState() == MessageContext::DispatchState::Discarded) { + if (dispatchState == MessageContext::DispatchState::Discarded) { O2_SIGNPOST_EVENT_EMIT_ERROR(stream_context, cid, "postProcessingCallbacks", "Output messages discarded."); return; } // Check if we never created any data for this timeslice - // if we did not, but we still have didDispatched set to true + // if we did not, but messages were dispatched, // it means it was created out of band. bool userDidCreate = false; for (size_t ri = 0; ri < routes.size(); ++ri) { @@ -203,14 +204,14 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec() } O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "userDidCreate == %d && didDispatch == %d", userDidCreate, - messageContext.didDispatch()); - if (userDidCreate == false && messageContext.didDispatch() == true) { + dispatchState == MessageContext::DispatchState::Dispatched); + if (userDidCreate == false && dispatchState == MessageContext::DispatchState::Dispatched) { O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "Data created out of band userDidCreate == %d && messageContext.didDispatch == %d", userDidCreate, - messageContext.didDispatch()); + dispatchState == MessageContext::DispatchState::Dispatched); return; } - if (userDidCreate == false && messageContext.didDispatch() == false) { + if (userDidCreate == false && dispatchState == MessageContext::DispatchState::NotDispatched) { O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service); O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "No data created."); return;