DPL: introduce a preStart callback

Author: ktf

Framework/Core/include/Framework/ServiceRegistry.h

@@ -94,6 +94,8 @@ struct ServiceRegistry {
   std::vector<ServiceEOSHandle> mPostEOSHandles;
   /// Callbacks for services to be executed after every dispatching
   std::vector<ServiceDispatchingHandle> mPostDispatchingHandles;
+  /// Callbacks for services to be executed before Start
+  std::vector<ServiceStartHandle> mPreStartHandles;
   /// Callbacks for services to be executed on exit
   std::vector<ServiceExitHandle> mPreExitHandles;
 
@@ -129,6 +131,8 @@ struct ServiceRegistry {
     return *this;
   }
 
+  /// Invoke callbacks to be executed in PreRun(), before the User Start callbacks
+  void preStartCallbacks();
   /// Invoke callbacks to be executed before every process method invokation
   void preProcessingCallbacks(ProcessingContext&);
   /// Invoke callbacks to be executed after every process method invokation

Framework/Core/include/Framework/ServiceSpec.h

@@ -37,6 +37,9 @@ class DanglingContext;
 
 /// A callback to create a given Service.
 using ServiceInit = std::function<ServiceHandle(ServiceRegistry&, DeviceState&, fair::mq::ProgOptions&)>;
+/// A callback invoked whenever we start running, before the user callback.
+using ServiceStartCallback = std::function<void(ServiceRegistry&, void*)>;
+/// A callback invoked whenever we stop running, before we exit.
 using ServiceExitCallback = std::function<void(ServiceRegistry&, void*)>;
 
 /// A callback to configure a given Service. Notice that the
@@ -130,6 +133,8 @@ struct ServiceSpec {
   /// dispatched.
   ServicePostDispatching postDispatching = nullptr;
 
+  /// Callback invoked on Start
+  ServiceStartCallback start = nullptr;
   /// Callback invoked on exit
   ServiceExitCallback exit = nullptr;
 
@@ -162,6 +167,11 @@ struct ServiceDispatchingHandle {
   void* service;
 };
 
+struct ServiceStartHandle {
+  ServiceExitCallback callback;
+  void* service;
+};
+
 struct ServiceExitHandle {
   ServiceExitCallback callback;
   void* service;

Framework/Core/src/CommonMessageBackends.cxx

@@ -128,6 +128,7 @@ o2::framework::ServiceSpec CommonMessageBackends::rateLimitingSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 
@@ -361,6 +362,7 @@ o2::framework::ServiceSpec CommonMessageBackends::arrowBackendSpec()
                        monitoring.flushBuffer();
                      },
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 
@@ -403,6 +405,7 @@ o2::framework::ServiceSpec CommonMessageBackends::fairMQBackendSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 
@@ -425,6 +428,7 @@ o2::framework::ServiceSpec CommonMessageBackends::stringBackendSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 
@@ -447,6 +451,7 @@ o2::framework::ServiceSpec CommonMessageBackends::rawBufferBackendSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 

Framework/Core/src/CommonServices.cxx

@@ -96,6 +96,7 @@ o2::framework::ServiceSpec CommonServices::monitoringSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      [](ServiceRegistry& registry, void* service) {
                        Monitoring* monitoring = reinterpret_cast<Monitoring*>(service);
                        delete monitoring;
@@ -122,6 +123,7 @@ o2::framework::ServiceSpec CommonServices::infologgerContextSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 
@@ -214,6 +216,7 @@ o2::framework::ServiceSpec CommonServices::infologgerSpec()
                      nullptr,
                      nullptr,
                      nullptr,
+                     nullptr,
                      ServiceKind::Serial};
 }
 
@@ -244,6 +247,7 @@ o2::framework::ServiceSpec CommonServices::configurationSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Global};
 }
 
@@ -276,6 +280,7 @@ o2::framework::ServiceSpec CommonServices::driverClientSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Global};
 }
 
@@ -302,6 +307,7 @@ o2::framework::ServiceSpec CommonServices::controlSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -325,6 +331,7 @@ o2::framework::ServiceSpec CommonServices::rootFileSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -352,6 +359,7 @@ o2::framework::ServiceSpec CommonServices::parallelSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -375,6 +383,7 @@ o2::framework::ServiceSpec CommonServices::timesliceIndex()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -398,6 +407,7 @@ o2::framework::ServiceSpec CommonServices::callbacksSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -428,6 +438,7 @@ o2::framework::ServiceSpec CommonServices::dataRelayer()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -463,6 +474,7 @@ o2::framework::ServiceSpec CommonServices::tracingSpec()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -501,6 +513,7 @@ o2::framework::ServiceSpec CommonServices::threadPool(int numWorkers)
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 
@@ -607,6 +620,7 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats()
     nullptr,
     nullptr,
     nullptr,
+    nullptr,
     ServiceKind::Serial};
 }
 

Framework/Core/src/DataProcessingDevice.cxx

@@ -409,6 +409,7 @@ void DataProcessingDevice::fillContext(DataProcessorContext& context)
 
 void DataProcessingDevice::PreRun()
 {
+  mServiceRegistry.preStartCallbacks();
   mServiceRegistry.get<CallbackService>()(CallbackService::Id::Start);
 }
 

Framework/Core/src/ServiceRegistry.cxx

@@ -99,6 +99,9 @@ void ServiceRegistry::bindService(ServiceSpec const& spec, void* service)
   if (spec.postDispatching) {
     mPostDispatchingHandles.push_back(ServiceDispatchingHandle{spec.postDispatching, service});
   }
+  if (spec.start) {
+    mPreStartHandles.push_back(ServiceStartHandle{spec.start, service});
+  }
   if (spec.exit) {
     mPreExitHandles.push_back(ServiceExitHandle{spec.exit, service});
   }
@@ -158,6 +161,16 @@ void ServiceRegistry::postDispatchingCallbacks(ProcessingContext& processContext
   }
 }
 
+/// Callbacks to be called in FairMQDevice::PreRun()
+void ServiceRegistry::preStartCallbacks()
+{
+  // FIXME: we need to call the callback only once for the global services
+  /// I guess...
+  for (auto startHandle = mPreStartHandles.begin(); startHandle != mPreStartHandles.end(); ++startHandle) {
+    startHandle->callback(*this, startHandle->service);
+  }
+}
+
 /// Invoke callback to be executed on exit, in reverse order.
 void ServiceRegistry::preExitCallbacks()
 {