Track message state as enum

Author: autumn-mck

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,16 +501,17 @@ 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<o2::header::DataHeader*, o2::framework::DataProcessingHeader*> findMessageHeaders(const Output& spec);
 
  private:
   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<int64_t, std::unique_ptr<fair::mq::Message>> mMessageCache;

Framework/Core/src/CommonServices.cxx

@@ -185,6 +185,10 @@ o2::framework::ServiceSpec CommonServices::streamContextSpec()
       auto& routes = processingContext.services().get<DeviceSpec const>().outputs;
       auto& timeslice = processingContext.services().get<TimingInfo>().timeslice;
       auto& messageContext = processingContext.services().get<MessageContext>();
+      // 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.

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();
     }
   }