DPL: make TimingInfo a service (#7436)

ktf · web-flow · commit 72a2bce69478 · 2021-10-28T13:47:17.000+02:00
diff --git a/Framework/Core/include/Framework/CommonServices.h b/Framework/Core/include/Framework/CommonServices.h
@@ -66,6 +66,7 @@ struct CommonServices {
   static ServiceSpec tracingSpec();
   static ServiceSpec threadPool(int numWorkers);
   static ServiceSpec dataProcessingStats();
+  static ServiceSpec timingInfoSpec();
 
   static std::vector<ServiceSpec> defaultServices(int numWorkers = 0);
   static std::vector<ServiceSpec> requiredServices();
diff --git a/Framework/Core/include/Framework/DataAllocator.h b/Framework/Core/include/Framework/DataAllocator.h
@@ -86,8 +86,7 @@ class DataAllocator
     using value_type = T;
   };
 
-  DataAllocator(TimingInfo* timingInfo,
-                ServiceRegistry* contextes,
+  DataAllocator(ServiceRegistry* contextes,
                 const AllowedOutputRoutes& routes);
 
   DataChunk& newChunk(const Output&, size_t);
@@ -108,7 +107,8 @@ class DataAllocator
     if constexpr (is_specialization<T, UninitializedVector>::value) {
       // plain buffer as polymorphic spectator std::vector, which does not run constructors / destructors
       using ValueType = typename T::value_type;
-      std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+      auto& timingInfo = mRegistry->get<TimingInfo>();
+      std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
       auto& context = mRegistry->get<MessageContext>();
 
       // Note: initial payload size is 0 and will be set by the context before sending
@@ -120,7 +120,8 @@ class DataAllocator
       // this catches all std::vector objects with messageable value type before checking if is also
       // has a root dictionary, so non-serialized transmission is preferred
       using ValueType = typename T::value_type;
-      std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+      auto& timingInfo = mRegistry->get<TimingInfo>();
+      std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
       auto& context = mRegistry->get<MessageContext>();
 
       // Note: initial payload size is 0 and will be set by the context before sending
@@ -129,7 +130,8 @@ class DataAllocator
     } else if constexpr (has_root_dictionary<T>::value == true && is_messageable<T>::value == false) {
       // Extended support for types implementing the Root ClassDef interface, both TObject
       // derived types and others
-      std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+      auto& timingInfo = mRegistry->get<TimingInfo>();
+      std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
       auto& context = mRegistry->get<MessageContext>();
 
       // Note: initial payload size is 0 and will be set by the context before sending
@@ -171,7 +173,8 @@ class DataAllocator
         if constexpr (is_messageable<T>::value == true) {
           auto [nElements] = std::make_tuple(args...);
           auto size = nElements * sizeof(T);
-          std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+          auto& timingInfo = mRegistry->get<TimingInfo>();
+          std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
           auto& context = mRegistry->get<MessageContext>();
 
           FairMQMessagePtr headerMessage = headerMessageFromOutput(spec, channel, o2::header::gSerializationMethodNone, size);
@@ -237,7 +240,8 @@ class DataAllocator
     using type = T;
 
     char* payload = reinterpret_cast<char*>(ptr);
-    std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+    auto& timingInfo = mRegistry->get<TimingInfo>();
+    std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
     // the correct payload size is set later when sending the
     // RawBufferContext, see DataProcessor::doSend
     auto header = headerMessageFromOutput(spec, channel, o2::header::gSerializationMethodNone, 0);
@@ -400,7 +404,8 @@ class DataAllocator
   //get the memory resource associated with an output
   o2::pmr::FairMQMemoryResource* getMemoryResource(const Output& spec)
   {
-    std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+    auto& timingInfo = mRegistry->get<TimingInfo>();
+    std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
     auto& context = mRegistry->get<MessageContext>();
     return *context.proxy().getTransport(channel);
   }
@@ -421,7 +426,8 @@ class DataAllocator
   {
     // Find a matching channel, extract the message for it form the container
     // and put it in the queue to be sent at the end of the processing
-    std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+    auto& timingInfo = mRegistry->get<TimingInfo>();
+    std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
 
     auto& context = mRegistry->get<MessageContext>();
     FairMQMessagePtr payloadMessage = o2::pmr::getMessage(std::forward<ContainerT>(container), *context.proxy().getTransport(channel));
@@ -460,7 +466,6 @@ class DataAllocator
 
  private:
   AllowedOutputRoutes mAllowedOutputRoutes;
-  TimingInfo* mTimingInfo;
   ServiceRegistry* mRegistry;
 
   std::string const& matchDataHeader(const Output& spec, size_t timeframeId);
diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx
@@ -115,6 +115,18 @@ o2::framework::ServiceSpec CommonServices::monitoringSpec()
     .kind = ServiceKind::Serial};
 }
 
+// Make it a service so that it can be used easily from the analysis
+// FIXME: Moreover, it makes sense that this will be duplicated on a per thread
+// basis when we get to it.
+o2::framework::ServiceSpec CommonServices::timingInfoSpec()
+{
+  return ServiceSpec{
+    .name = "timing-info",
+    .init = simpleServiceInit<TimingInfo, TimingInfo>(),
+    .configure = noConfiguration(),
+    .kind = ServiceKind::Serial};
+}
+
 o2::framework::ServiceSpec CommonServices::datatakingContextSpec()
 {
   return ServiceSpec{
@@ -588,6 +600,7 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats()
 std::vector<ServiceSpec> CommonServices::defaultServices(int numThreads)
 {
   std::vector<ServiceSpec> specs{
+    timingInfoSpec(),
     timesliceIndex(),
     driverClientSpec(),
     datatakingContextSpec(),
diff --git a/Framework/Core/src/DataAllocator.cxx b/Framework/Core/src/DataAllocator.cxx
@@ -35,11 +35,9 @@ using DataHeader = o2::header::DataHeader;
 using DataDescription = o2::header::DataDescription;
 using DataProcessingHeader = o2::framework::DataProcessingHeader;
 
-DataAllocator::DataAllocator(TimingInfo* timingInfo,
-                             ServiceRegistry* contextRegistry,
+DataAllocator::DataAllocator(ServiceRegistry* contextRegistry,
                              const AllowedOutputRoutes& routes)
   : mAllowedOutputRoutes{routes},
-    mTimingInfo{timingInfo},
     mRegistry{contextRegistry}
 {
 }
@@ -62,7 +60,8 @@ std::string const& DataAllocator::matchDataHeader(const Output& spec, size_t tim
 
 DataChunk& DataAllocator::newChunk(const Output& spec, size_t size)
 {
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  auto& timingInfo = mRegistry->get<TimingInfo>();
+  std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
   auto& context = mRegistry->get<MessageContext>();
 
   FairMQMessagePtr headerMessage = headerMessageFromOutput(spec, channel,                        //
@@ -77,7 +76,7 @@ void DataAllocator::adoptChunk(const Output& spec, char* buffer, size_t size, fa
 {
   // Find a matching channel, create a new message for it and put it in the
   // queue to be sent at the end of the processing
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  std::string const& channel = matchDataHeader(spec, mRegistry->get<TimingInfo>().timeslice);
 
   FairMQMessagePtr headerMessage = headerMessageFromOutput(spec, channel,                        //
                                                            o2::header::gSerializationMethodNone, //
@@ -94,17 +93,18 @@ FairMQMessagePtr DataAllocator::headerMessageFromOutput(Output const& spec,
                                                         o2::header::SerializationMethod method, //
                                                         size_t payloadSize)                     //
 {
+  auto& timingInfo = mRegistry->get<TimingInfo>();
   DataHeader dh;
   dh.dataOrigin = spec.origin;
   dh.dataDescription = spec.description;
   dh.subSpecification = spec.subSpec;
   dh.payloadSize = payloadSize;
   dh.payloadSerializationMethod = method;
-  dh.tfCounter = mTimingInfo->tfCounter;
-  dh.firstTForbit = mTimingInfo->firstTFOrbit;
-  dh.runNumber = mTimingInfo->runNumber;
+  dh.tfCounter = timingInfo.tfCounter;
+  dh.firstTForbit = timingInfo.firstTFOrbit;
+  dh.runNumber = timingInfo.runNumber;
 
-  DataProcessingHeader dph{mTimingInfo->timeslice, 1};
+  DataProcessingHeader dph{timingInfo.timeslice, 1};
   auto& context = mRegistry->get<MessageContext>();
 
   auto channelAlloc = o2::pmr::getTransportAllocator(context.proxy().getTransport(channel, 0));
@@ -114,7 +114,8 @@ FairMQMessagePtr DataAllocator::headerMessageFromOutput(Output const& spec,
 void DataAllocator::addPartToContext(FairMQMessagePtr&& payloadMessage, const Output& spec,
                                      o2::header::SerializationMethod serializationMethod)
 {
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  auto& timingInfo = mRegistry->get<TimingInfo>();
+  std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
   auto headerMessage = headerMessageFromOutput(spec, channel, serializationMethod, 0);
 
   // FIXME: this is kind of ugly, we know that we can change the content of the
@@ -132,7 +133,8 @@ void DataAllocator::addPartToContext(FairMQMessagePtr&& payloadMessage, const Ou
 void DataAllocator::adopt(const Output& spec, std::string* ptr)
 {
   std::unique_ptr<std::string> payload(ptr);
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  auto& timingInfo = mRegistry->get<TimingInfo>();
+  std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
   // the correct payload size is set later when sending the
   // StringContext, see DataProcessor::doSend
   auto header = headerMessageFromOutput(spec, channel, o2::header::gSerializationMethodNone, 0);
@@ -189,7 +191,8 @@ void doWriteTable(std::shared_ptr<FairMQResizableBuffer> b, arrow::Table* table)
 
 void DataAllocator::adopt(const Output& spec, TableBuilder* tb)
 {
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  auto& timingInfo = mRegistry->get<TimingInfo>();
+  std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
   auto header = headerMessageFromOutput(spec, channel, o2::header::gSerializationMethodArrow, 0);
   auto& context = mRegistry->get<ArrowContext>();
 
@@ -210,7 +213,8 @@ void DataAllocator::adopt(const Output& spec, TableBuilder* tb)
 
 void DataAllocator::adopt(const Output& spec, TreeToTable* t2t)
 {
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  auto& timingInfo = mRegistry->get<TimingInfo>();
+  std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
 
   auto header = headerMessageFromOutput(spec, channel, o2::header::gSerializationMethodArrow, 0);
   auto& context = mRegistry->get<ArrowContext>();
@@ -234,7 +238,8 @@ void DataAllocator::adopt(const Output& spec, TreeToTable* t2t)
 
 void DataAllocator::adopt(const Output& spec, std::shared_ptr<arrow::Table> ptr)
 {
-  std::string const& channel = matchDataHeader(spec, mTimingInfo->timeslice);
+  auto& timingInfo = mRegistry->get<TimingInfo>();
+  std::string const& channel = matchDataHeader(spec, timingInfo.timeslice);
   auto header = headerMessageFromOutput(spec, channel, o2::header::gSerializationMethodArrow, 0);
   auto& context = mRegistry->get<ArrowContext>();
 
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -102,7 +102,7 @@ DataProcessingDevice::DataProcessingDevice(RunningDeviceRef ref, ServiceRegistry
     mError{mSpec.algorithm.onError},
     mConfigRegistry{nullptr},
     mServiceRegistry{registry},
-    mAllocator{&mTimingInfo, &registry, mSpec.outputs},
+    mAllocator{&registry, mSpec.outputs},
     mQuotaEvaluator{registry.get<ComputingQuotaEvaluator>()},
     mAwakeHandle{nullptr},
     mProcessingPolicies{policies}
@@ -535,7 +535,7 @@ void DataProcessingDevice::fillContext(DataProcessorContext& context, DeviceCont
   context.registry = &mServiceRegistry;
   context.completed = &mCompleted;
   context.expirationHandlers = &mExpirationHandlers;
-  context.timingInfo = &mTimingInfo;
+  context.timingInfo = &mServiceRegistry.get<TimingInfo>();
   context.allocator = &mAllocator;
   context.statefulProcess = &mStatefulProcess;
   context.statelessProcess = &mStatelessProcess;
diff --git a/Framework/Core/test/test_DataAllocator.cxx b/Framework/Core/test/test_DataAllocator.cxx
@@ -43,10 +43,9 @@ using namespace o2::framework;
 // this function is only used to do the static checks for API return types
 void doTypeChecks()
 {
-  TimingInfo* timingInfo = nullptr;
   ServiceRegistry* contextes = nullptr;
   std::vector<OutputRoute> routes;
-  DataAllocator allocator(timingInfo, contextes, routes);
+  DataAllocator allocator(contextes, routes);
   const Output output{"TST", "DUMMY", 0, Lifetime::Timeframe};
   // we require references to objects owned by allocator context
   static_assert(std::is_lvalue_reference<decltype(allocator.make<int>(output))>::value);
diff --git a/Framework/Core/test/test_TableBuilder.cxx b/Framework/Core/test/test_TableBuilder.cxx
@@ -347,7 +347,7 @@ BOOST_AUTO_TEST_CASE(TestSoAIntegration)
 BOOST_AUTO_TEST_CASE(TestDataAllocatorReturnType)
 {
   std::vector<OutputRoute> routes;
-  DataAllocator allocator(nullptr, nullptr, routes);
+  DataAllocator allocator(nullptr, routes);
   const Output output{"TST", "DUMMY", 0, Lifetime::Timeframe};
   // we require reference to object owned by allocator context
   static_assert(std::is_lvalue_reference<decltype(allocator.make<TableBuilder>(output))>::value);
diff --git a/Framework/TestWorkflows/src/o2AnalysisTaskExample.cxx b/Framework/TestWorkflows/src/o2AnalysisTaskExample.cxx
@@ -19,12 +19,13 @@ using namespace o2::framework;
 
 // This is a stateful task, where we send the state downstream.
 struct ATask {
-  explicit ATask(int state)
-    : mSomeState{state} {}
+  Service<TimingInfo> info;
+
+  // explicit ATask(int state)
+  //   : mSomeState{state} {}
 
   void init(InitContext& ic)
   {
-    mSomeState += 1;
   }
 
   void run(ProcessingContext& pc)
@@ -47,12 +48,10 @@ struct ATask {
     hEta->SetName("Eta");
     hEta->Write();
   }
-
-  int mSomeState;
 };
 
 WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
 {
   return WorkflowSpec{
-    adaptAnalysisTask<ATask>(cfgc, TaskName{"mySimpleTrackAnalysis"}, 0)};
+    adaptAnalysisTask<ATask>(cfgc, TaskName{"mySimpleTrackAnalysis"})};
 }

Original file line number	Diff line number	Diff line change
`@@ -347,7 +347,7 @@ BOOST_AUTO_TEST_CASE(TestSoAIntegration)`
`347`	`347`	`BOOST_AUTO_TEST_CASE(TestDataAllocatorReturnType)`
`348`	`348`	`{`
`349`	`349`	`std::vector<OutputRoute> routes;`
`350`		`- DataAllocator allocator(nullptr, nullptr, routes);`
	`350`	`+ DataAllocator allocator(nullptr, routes);`
`351`	`351`	`const Output output{"TST", "DUMMY", 0, Lifetime::Timeframe};`
`352`	`352`	`// we require reference to object owned by allocator context`
`353`	`353`	`static_assert(std::is_lvalue_reference<decltype(allocator.make<TableBuilder>(output))>::value);`
Original file line number	Diff line number	Diff line change
`@@ -19,12 +19,13 @@ using namespace o2::framework;`
`19`	`19`
`20`	`20`	`// This is a stateful task, where we send the state downstream.`
`21`	`21`	`struct ATask {`
`22`		`- explicit ATask(int state)`
`23`		`- : mSomeState{state} {}`
	`22`	`+ Service<TimingInfo> info;`
	`23`	`+`
	`24`	`+ // explicit ATask(int state)`
	`25`	`+ // : mSomeState{state} {}`
`24`	`26`
`25`	`27`	`void init(InitContext& ic)`
`26`	`28`	`{`
`27`		`- mSomeState += 1;`
`28`	`29`	`}`
`29`	`30`
`30`	`31`	`void run(ProcessingContext& pc)`
`@@ -47,12 +48,10 @@ struct ATask {`
`47`	`48`	`hEta->SetName("Eta");`
`48`	`49`	`hEta->Write();`
`49`	`50`	`}`
`50`		`-`
`51`		`- int mSomeState;`
`52`	`51`	`};`
`53`	`52`
`54`	`53`	`WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)`
`55`	`54`	`{`
`56`	`55`	`return WorkflowSpec{`
`57`		`- adaptAnalysisTask<ATask>(cfgc, TaskName{"mySimpleTrackAnalysis"}, 0)};`
	`56`	`+ adaptAnalysisTask<ATask>(cfgc, TaskName{"mySimpleTrackAnalysis"})};`
`58`	`57`	`}`