Merge pull request #8550 from ktf/oldest-possible-inputs

DPL: Add support for propagating oldest possible timeslice

Author: ktf

Detectors/CTP/workflowScalers/src/ctp-proxy.cxx

@@ -101,7 +101,7 @@ InjectorFunction dcs2dpl()
       FairMQParts outParts;
       outParts.AddPart(std::move(hdMessageF));
       outParts.AddPart(std::move(plMessageF));
-      sendOnChannel(device, outParts, channel);
+      sendOnChannel(device, outParts, channel, *timesliceId);
       LOG(info) << "Sent CTP counters DPL message" << std::flush;
     }
   };

Detectors/DCS/testWorkflow/src/DCStoDPLconverter.h

@@ -123,7 +123,7 @@ o2f::InjectorFunction dcs2dpl(std::unordered_map<DPID, o2h::DataDescription>& dp
         FairMQParts outParts;
         outParts.AddPart(std::move(hdMessage));
         outParts.AddPart(std::move(plMessage));
-        o2f::sendOnChannel(device, outParts, channel);
+        o2f::sendOnChannel(device, outParts, channel, *timesliceId);
       }
 
       timer = timerNow;

Detectors/DCS/testWorkflow/src/dcs-config-proxy.cxx

@@ -42,7 +42,7 @@ void sendAnswer(const std::string& what, const std::string& ack_chan, FairMQDevi
     memcpy(msg->GetData(), what.c_str(), what.size());
     FairMQParts outParts;
     outParts.AddPart(std::move(msg));
-    sendOnChannel(device, outParts, ack_chan);
+    sendOnChannel(device, outParts, ack_chan, (size_t)-1);
   }
 }
 
@@ -127,12 +127,12 @@ InjectorFunction dcs2dpl(const std::string& acknowledge)
     FairMQParts outPartsF;
     outPartsF.AddPart(std::move(hdMessageF));
     outPartsF.AddPart(std::move(plMessageF));
-    sendOnChannel(device, outPartsF, channel);
+    sendOnChannel(device, outPartsF, channel, (size_t)-1);
 
     FairMQParts outPartsN;
     outPartsN.AddPart(std::move(hdMessageN));
     outPartsN.AddPart(std::move(plMessageN));
-    sendOnChannel(device, outPartsN, channel);
+    sendOnChannel(device, outPartsN, channel, *timesliceId);
 
     sendAnswer(fmt::format("{}:ok", filename), acknowledge, device);
     LOG(info) << "Sent DPL message and acknowledgment for file " << filename;

Detectors/Raw/src/RawFileReaderWorkflow.cxx

@@ -18,8 +18,10 @@
 #include "Framework/ControlService.h"
 #include "Framework/SourceInfoHeader.h"
 #include "Framework/DataProcessingHeader.h"
+#include "Framework/DataProcessingHelpers.h"
 #include "Framework/Task.h"
 #include "Framework/Logger.h"
+#include "Framework/DomainInfoHeader.h"
 
 #include "DetectorsRaw/RawFileReader.h"
 #include "DetectorsRaw/RDHUtils.h"
@@ -351,7 +353,10 @@ void RawReaderSpecs::run(o2f::ProcessingContext& ctx)
 
   mSentSize += tfSize;
   mSentMessages += tfNParts;
-
+  for (auto& msgIt : messagesPerRoute) {
+    auto& channel = device->GetChannel(msgIt.first, 0);
+    o2::framework::DataProcessingHelpers::sendOldestPossibleTimeframe(channel, mTFCounter);
+  }
   mReader->setNextTFToRead(++tfID);
   ++mTFCounter;
 }

Framework/Core/CMakeLists.txt

@@ -38,6 +38,7 @@ o2_add_library(Framework
                        src/CCDBHelpers.cxx
                        src/DispatchPolicy.cxx
                        src/DataSender.cxx
+                       src/DomainInfoHeader.cxx
                        src/ProcessingPoliciesHelpers.cxx
                        src/ConfigParamStore.cxx
                        src/ConfigParamsHelper.cxx

Framework/Core/include/Framework/ChannelInfo.h

@@ -11,6 +11,7 @@
 #ifndef O2_FRAMEWORK_CHANNELINFO_H
 #define O2_FRAMEWORK_CHANNELINFO_H
 
+#include "Framework/RoutingIndices.h"
 #include <string>
 #include <fairmq/Parts.h>
 
@@ -47,6 +48,7 @@ struct InputChannelInfo {
   /// We start with false since we assume there is no
   /// backpressure to start with.
   bool backpressureNotified = false;
+  ChannelIndex id = {-1};
 };
 
 } // namespace o2::framework

Framework/Core/include/Framework/CommonServices.h

@@ -65,6 +65,7 @@ struct CommonServices {
   static ServiceSpec dataRelayer();
   static ServiceSpec dataSender();
   static ServiceSpec tracingSpec();
+  static ServiceSpec summaryServiceSpec();
   static ServiceSpec threadPool(int numWorkers);
   static ServiceSpec dataProcessingStats();
   static ServiceSpec objectCache();

Framework/Core/include/Framework/DataAllocator.h

@@ -285,7 +285,7 @@ class DataAllocator
     RouteIndex routeIndex = matchDataHeader(spec, mRegistry->get<TimingInfo>().timeslice);
     if constexpr (is_messageable<T>::value == true) {
       // Serialize a snapshot of a trivially copyable, non-polymorphic object,
-      payloadMessage = proxy.createMessage(routeIndex, sizeof(T));
+      payloadMessage = proxy.createOutputMessage(routeIndex, sizeof(T));
       memcpy(payloadMessage->GetData(), &object, sizeof(T));
 
       serializationType = o2::header::gSerializationMethodNone;
@@ -298,7 +298,7 @@ class DataAllocator
         // reference object
         constexpr auto elementSizeInBytes = sizeof(ElementType);
         auto sizeInBytes = elementSizeInBytes * object.size();
-        payloadMessage = proxy.createMessage(routeIndex, sizeInBytes);
+        payloadMessage = proxy.createOutputMessage(routeIndex, sizeInBytes);
 
         if constexpr (std::is_pointer<typename T::value_type>::value == false) {
           // vector of elements
@@ -326,7 +326,7 @@ class DataAllocator
       }
     } else if constexpr (has_root_dictionary<T>::value == true || is_specialization_v<T, ROOTSerialized> == true) {
       // Serialize a snapshot of an object with root dictionary
-      payloadMessage = proxy.createMessage(routeIndex);
+      payloadMessage = proxy.createOutputMessage(routeIndex);
       if constexpr (is_specialization_v<T, ROOTSerialized> == true) {
         // Explicitely ROOT serialize a snapshot of object.
         // An object wrapped into type `ROOTSerialized` is explicitely marked to be ROOT serialized
@@ -407,7 +407,7 @@ class DataAllocator
     auto& timingInfo = mRegistry->get<TimingInfo>();
     auto& proxy = mRegistry->get<FairMQDeviceProxy>();
     RouteIndex routeIndex = matchDataHeader(spec, timingInfo.timeslice);
-    return *proxy.getTransport(routeIndex);
+    return *proxy.getOutputTransport(routeIndex);
   }
 
   //make a stl (pmr) vector
@@ -509,7 +509,7 @@ DataAllocator::CacheId DataAllocator::adoptContainer(const Output& spec, Contain
   auto routeIndex = matchDataHeader(spec, timingInfo.timeslice);
 
   auto& context = mRegistry->get<MessageContext>();
-  auto* transport = mRegistry->get<FairMQDeviceProxy>().getTransport(routeIndex);
+  auto* transport = mRegistry->get<FairMQDeviceProxy>().getOutputTransport(routeIndex);
   FairMQMessagePtr payloadMessage = o2::pmr::getMessage(std::forward<ContainerT>(container), *transport);
   FairMQMessagePtr headerMessage = headerMessageFromOutput(spec, routeIndex,         //
                                                            method,                   //

…amework/Core/src/DataProcessingHelpers.h …nclude/Framework/DataProcessingHelpers.hFramework/Core/src/DataProcessingHelpers.h renamed to Framework/Core/include/Framework/DataProcessingHelpers.h Framework/Core/src/DataProcessingHelpers.h renamed to Framework/Core/include/Framework/DataProcessingHelpers.h

@@ -11,12 +11,14 @@
 #ifndef O2_FRAMEWORK_DATAPROCESSINGHELPERS_H_
 #define O2_FRAMEWORK_DATAPROCESSINGHELPERS_H_
 
+#include "Framework/TimesliceIndex.h"
 #include <fairmq/FwdDecls.h>
 
 namespace o2::framework
 {
 
 struct OutputChannelSpec;
+class FairMQDeviceProxy;
 
 /// Generic helpers for DataProcessing releated functions.
 struct DataProcessingHelpers {
@@ -25,6 +27,8 @@ struct DataProcessingHelpers {
   /// @param channel the OutputChannelSpec of the channel which needs to be signaled
   ///        for EndOfStream
   static void sendEndOfStream(FairMQDevice& device, OutputChannelSpec const& channel);
+  static void sendOldestPossibleTimeframe(fair::mq::Channel& channel, size_t timeslice);
+  static void broadcastOldestPossibleTimeslice(FairMQDeviceProxy& proxy, size_t timeslice);
 };
 
 } // namespace o2::framework

Framework/Core/include/Framework/DataRelayer.h

@@ -101,6 +101,14 @@ class DataRelayer
                     size_t nMessages,
                     size_t nPayloads = 1);
 
+  /// This is to set the oldest possible @a timeslice this relayer can
+  /// possibly see on an input channel @a channel.
+  void setOldestPossibleInput(TimesliceId timeslice, ChannelIndex channel);
+
+  /// This is to retrieve the oldest possible @a timeslice this relayer can
+  /// possibly have in output.
+  TimesliceIndex::OldestOutputInfo getOldestPossibleOutput() const;
+
   /// @returns the actions ready to be performed.
   void getReadyToProcess(std::vector<RecordAction>& completed);