Skip to content

Commit 4679fed

Browse files
autumn-mckautumn-mck
authored
Add MessageContext message discarding (#15662)
1 parent ba7f425 commit 4679fed

3 files changed

Lines changed: 38 additions & 12 deletions

File tree

‎Framework/Core/include/Framework/MessageContext.h‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,12 @@ struct Output;
5353
class MessageContext
5454
{
5555
public:
56+
enum class DispatchState {
57+
NotDispatched,
58+
Dispatched,
59+
Discarded,
60+
};
61+
5662
constexpr static ServiceKind service_kind = ServiceKind::Stream;
5763

5864
// so far we are only using one instance per named channel
@@ -466,6 +472,11 @@ class MessageContext
466472
/// discarded.
467473
void clear();
468474

475+
/// Discard pending output messages without asserting that they were sent. This
476+
/// is intended for exception teardown paths where normal post-processing will
477+
/// not run.
478+
void discard();
479+
469480
FairMQDeviceProxy& proxy()
470481
{
471482
return mProxy;
@@ -490,16 +501,16 @@ class MessageContext
490501
o2::header::DataHeader* findMessageHeader(const Output& spec);
491502
o2::header::Stack* findMessageHeaderStack(const Output& spec);
492503
[[nodiscard]] int countDeviceOutputs(bool excludeDPLOrigin = false) const;
493-
void fakeDispatch() { mDidDispatch = true; }
494-
bool didDispatch() { return mDidDispatch; }
504+
void fakeDispatch() { mDispatchState = DispatchState::Dispatched; }
505+
[[nodiscard]] DispatchState dispatchState() const { return mDispatchState; }
495506
o2::framework::DataProcessingHeader* findMessageDataProcessingHeader(const Output& spec);
496507
std::pair<o2::header::DataHeader*, o2::framework::DataProcessingHeader*> findMessageHeaders(const Output& spec);
497508

498509
private:
499510
FairMQDeviceProxy& mProxy;
500511
Messages mMessages;
501512
Messages mScheduledMessages;
502-
bool mDidDispatch = false;
513+
DispatchState mDispatchState = DispatchState::NotDispatched;
503514
DispatchControl mDispatchControl;
504515
/// Cached messages, in case we want to reuse them.
505516
std::unordered_map<int64_t, std::unique_ptr<fair::mq::Message>> mMessageCache;

‎Framework/Core/src/CommonServices.cxx‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -185,11 +185,17 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec()
185185
auto& routes = processingContext.services().get<DeviceSpec const>().outputs;
186186
auto& timeslice = processingContext.services().get<TimingInfo>().timeslice;
187187
auto& messageContext = processingContext.services().get<MessageContext>();
188+
auto dispatchState = messageContext.dispatchState();
189+
O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service);
190+
// Do not report discarded messages as missing outputs.
191+
if (dispatchState == MessageContext::DispatchState::Discarded) {
192+
O2_SIGNPOST_EVENT_EMIT_ERROR(stream_context, cid, "postProcessingCallbacks", "Output messages discarded.");
193+
return;
194+
}
188195
// Check if we never created any data for this timeslice
189-
// if we did not, but we still have didDispatched set to true
196+
// if we did not, but messages were dispatched,
190197
// it means it was created out of band.
191198
bool userDidCreate = false;
192-
O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service);
193199
for (size_t ri = 0; ri < routes.size(); ++ri) {
194200
if (stream->routeCreated[ri] == true && stream->routeDPLCreated[ri] == false) {
195201
userDidCreate = true;
@@ -198,14 +204,14 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec()
198204
}
199205
O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "userDidCreate == %d && didDispatch == %d",
200206
userDidCreate,
201-
messageContext.didDispatch());
202-
if (userDidCreate == false && messageContext.didDispatch() == true) {
207+
dispatchState == MessageContext::DispatchState::Dispatched);
208+
if (userDidCreate == false && dispatchState == MessageContext::DispatchState::Dispatched) {
203209
O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "Data created out of band userDidCreate == %d && messageContext.didDispatch == %d",
204210
userDidCreate,
205-
messageContext.didDispatch());
211+
dispatchState == MessageContext::DispatchState::Dispatched);
206212
return;
207213
}
208-
if (userDidCreate == false && messageContext.didDispatch() == false) {
214+
if (userDidCreate == false && dispatchState == MessageContext::DispatchState::NotDispatched) {
209215
O2_SIGNPOST_ID_FROM_POINTER(cid, stream_context, service);
210216
O2_SIGNPOST_EVENT_EMIT(stream_context, cid, "postProcessingCallbacks", "No data created.");
211217
return;

‎Framework/Core/src/MessageContext.cxx‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ int MessageContext::countDeviceOutputs(bool excludeDPLOrigin) const
8484
{
8585
// If we dispatched some messages before the end of the callback
8686
// we need to account for them as well.
87-
int noutputs = mDidDispatch ? 1 : 0;
87+
int noutputs = mDispatchState == DispatchState::Dispatched ? 1 : 0;
8888
constexpr o2::header::DataOrigin DataOriginDPL{"DPL"};
8989
for (auto it = mMessages.rbegin(); it != mMessages.rend(); ++it) {
9090
if (!excludeDPLOrigin || (*it)->header()->dataOrigin != DataOriginDPL) {
@@ -103,7 +103,16 @@ void MessageContext::clear()
103103
{
104104
// Verify that everything has been sent on clear.
105105
assert(std::all_of(mMessages.begin(), mMessages.end(), [](auto& m) { return m->empty(); }));
106-
mDidDispatch = false;
106+
assert(mScheduledMessages.empty());
107+
mDispatchState = DispatchState::NotDispatched;
108+
mScheduledMessages.clear();
109+
mMessages.clear();
110+
}
111+
112+
void MessageContext::discard()
113+
{
114+
mDispatchState = DispatchState::Discarded;
115+
mScheduledMessages.clear();
107116
mMessages.clear();
108117
}
109118

@@ -157,7 +166,7 @@ void MessageContext::schedule(Messages::value_type&& message)
157166
}
158167
mDispatchControl.dispatch(std::move(parts), ChannelIndex{ci}, DefaultChannelIndex);
159168
}
160-
mDidDispatch = mScheduledMessages.empty() == false;
169+
mDispatchState = DispatchState::Dispatched;
161170
mScheduledMessages.clear();
162171
}
163172
}

0 commit comments

Comments
 (0)