DPL: Proper ServiceRegistryRef implementation

ktf · ktf · commit f216f34fbb6f · 2022-10-20T17:12:21.000+02:00
For now it simply forwards to the underlying service registry.
diff --git a/Framework/Core/include/Framework/CommonServices.h b/Framework/Core/include/Framework/CommonServices.h
@@ -12,6 +12,7 @@
 #define O2_FRAMEWORK_COMMONSERVICES_H_
 
 #include "Framework/ServiceSpec.h"
+#include "Framework/ServiceRegistryRef.h"
 #include "Framework/TypeIdHelpers.h"
 
 class TDatabasePDG;
diff --git a/Framework/Core/include/Framework/DataProcessingDevice.h b/Framework/Core/include/Framework/DataProcessingDevice.h
@@ -122,7 +122,7 @@ struct DeviceConfigurationHelpers {
 class DataProcessingDevice : public fair::mq::Device
 {
  public:
-  DataProcessingDevice(RunningDeviceRef ref, ServiceRegistryRef, ProcessingPolicies& policies);
+  DataProcessingDevice(RunningDeviceRef ref, ServiceRegistry&, ProcessingPolicies& policies);
   void Init() final;
   void InitTask() final;
   void PreRun() final;
@@ -159,7 +159,7 @@ class DataProcessingDevice : public fair::mq::Device
   AlgorithmSpec::ErrorCallback mError;
   std::function<void(RuntimeErrorRef e, InputRecord& record)> mErrorHandling;
   std::unique_ptr<ConfigParamRegistry> mConfigRegistry;
-  ServiceRegistryRef mServiceRegistry;
+  ServiceRegistry& mServiceRegistry;
   DataAllocator mAllocator;
   DataRelayer* mRelayer = nullptr;
   /// Expiration handler
diff --git a/Framework/Core/include/Framework/DataSender.h b/Framework/Core/include/Framework/DataSender.h
@@ -40,6 +40,8 @@ class DataSender
 
  private:
   FairMQDeviceProxy& mProxy;
+  // We need the ServiceRegistry and not a ref, to be able
+  // to call the callbacks after sending.
   ServiceRegistryRef mRegistry;
   DeviceSpec const& mSpec;
   std::vector<OutputSpec> mOutputs;
diff --git a/Framework/Core/include/Framework/DebugGUI.h b/Framework/Core/include/Framework/DebugGUI.h
@@ -24,6 +24,7 @@
 
 namespace o2::framework
 {
+struct ServiceRegistry;
 /// Plugin interface for DPL GUIs.
 struct DebugGUI {
   virtual std::function<void(void)> getGUIDebugger(std::vector<o2::framework::DeviceInfo> const& infos,
@@ -41,7 +42,7 @@ struct DebugGUI {
   virtual void keyUp(char key) = 0;
   virtual void charIn(char key) = 0;
 
-  virtual void* initGUI(char const* windowTitle, ServiceRegistryRef registry) = 0;
+  virtual void* initGUI(char const* windowTitle, ServiceRegistry& registry) = 0;
   virtual void getFrameJSON(void* data, std::ostream& json_data) = 0;
   virtual void getFrameRaw(void* data, void** raw_data, int* size) = 0;
   virtual bool pollGUIPreRender(void* context, float delta) = 0;
diff --git a/Framework/Core/include/Framework/ErrorContext.h b/Framework/Core/include/Framework/ErrorContext.h
@@ -12,7 +12,7 @@
 #define O2_FRAMEWORK_ERROR_CONTEXT_H_
 
 #include "Framework/InputRecord.h"
-#include "Framework/ServiceRegistry.h"
+#include "Framework/ServiceRegistryRef.h"
 #include "Framework/RuntimeError.h"
 
 namespace o2::framework
@@ -31,7 +31,7 @@ class ErrorContext
   }
 
   InputRecord const& inputs() { return mInputs; }
-  ServiceRegistry const& services() { return mServices; }
+  ServiceRegistryRef services() { return mServices; }
   RuntimeErrorRef exception() { return mExceptionRef; }
 
  private:
diff --git a/Framework/Core/include/Framework/ServiceRegistry.h b/Framework/Core/include/Framework/ServiceRegistry.h
@@ -370,10 +370,6 @@ struct ServiceRegistry {
   }
 };
 
-// This is to migrate the code in QC.
-// FIXME: move this to the proper smart reference class once done
-using ServiceRegistryRef = o2::framework::ServiceRegistry &;
-
 } // namespace o2::framework
 
 #endif // O2_FRAMEWORK_SERVICEREGISTRY_H_
diff --git a/Framework/Core/include/Framework/ServiceRegistryRef.h b/Framework/Core/include/Framework/ServiceRegistryRef.h
@@ -11,7 +11,64 @@
 #ifndef O2_FRAMEWORK_SERVICEREGISTRYREF_H_
 #define O2_FRAMEWORK_SERVICEREGISTRYREF_H_
 
-// Until we merge the proper ServiceRegistryRef...
 #include "Framework/ServiceRegistry.h"
-#endif // O2_FRAMEWORK_SERVICEREGISTRYREF_H_
+#include "Framework/Logger.h"
 
+namespace o2::framework
+{
+
+class ServiceRegistryRef
+{
+ public:
+  // The streamId is used to identify the stream in case we have multiple
+  // threads. We cannot merely used the thread id because that does not
+  // work in case the thread is created ad-hoc, like it appears to happen
+  // for both libuv and FairMQ. This behaviour, BTW, makes usage
+  // of thread local storage basically impossible (i.e. you lose state).
+  // We use the following convention:
+  // - streamId == 0 means the main thread
+  // - streamId > 0 means one of the libuv worker threads
+  // - streamId == -1 means the region callback thread
+  // - streamId < -1 means some other worker thread of FairMQ which
+  //  we do not know about.
+  //
+  // The getter will also make sure that a service of kind Stream
+  // cannot be accessed if the streamId is <= 0 and complain accordingly.
+  // The dataProcessorId will be used to distinguish between different
+  // data processors when
+  ServiceRegistryRef(ServiceRegistry& registry, short streamId = 0, short dataProcessorId = 0)
+    : mRegistry(registry),
+      mContext{streamId, dataProcessorId}
+  {
+  }
+
+  /// Check if service of type T is currently active.
+  template <typename T>
+  std::enable_if_t<std::is_const_v<T> == false, bool> active() const
+  {
+    return mRegistry.active<T>();
+  }
+
+  /// Get a service for the given interface T. The returned reference exposed to
+  /// the user is actually of the last concrete type C registered, however this
+  /// should not be a problem.
+  template <typename T>
+  T& get() const
+  {
+    return mRegistry.get<T>();
+  }
+
+  /// Invoke before sending messages @a parts on a channel @a channelindex
+  void preSendingMessagesCallbacks(fair::mq::Parts& parts, ChannelIndex channelindex)
+  {
+    mRegistry.preSendingMessagesCallbacks(mRegistry, parts, channelindex);
+  }
+
+ private:
+  ServiceRegistry& mRegistry;
+  ServiceRegistry::Context mContext;
+};
+
+} // namespace o2::framework
+
+#endif // O2_FRAMEWORK_SERVICEREGISTRY_H_
diff --git a/Framework/Core/include/Framework/ServiceSpec.h b/Framework/Core/include/Framework/ServiceSpec.h
@@ -30,7 +30,7 @@ namespace o2::framework
 struct InitContext;
 struct DeviceSpec;
 struct ServiceRegistry;
-using ServiceRegistryRef = ServiceRegistry&;
+struct ServiceRegistryRef;
 struct DeviceState;
 struct ProcessingContext;
 class EndOfStreamContext;
@@ -96,10 +96,10 @@ using ServicePostDispatching = void (*)(ProcessingContext&, void*);
 using ServicePostForwarding = void (*)(ProcessingContext&, void*);
 
 /// Callback invoked when the driver enters the init phase.
-using ServiceDriverInit = void (*)(ServiceRegistryRef, boost::program_options::variables_map const&);
+using ServiceDriverInit = void (*)(ServiceRegistry&, boost::program_options::variables_map const&);
 
 /// Callback invoked when the driver enters the init phase.
-using ServiceDriverStartup = void (*)(ServiceRegistryRef, boost::program_options::variables_map const&);
+using ServiceDriverStartup = void (*)(ServiceRegistry&, boost::program_options::variables_map const&);
 
 /// Callback invoked when we inject internal devices in the topology
 using ServiceTopologyInject = void (*)(WorkflowSpecNode&, ConfigContext&);
diff --git a/Framework/Core/src/ArrowSupport.cxx b/Framework/Core/src/ArrowSupport.cxx
@@ -364,7 +364,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
                        monitoring.send(Metric{(uint64_t)arrow->bytesDestroyed(), "arrow-bytes-destroyed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
                        monitoring.send(Metric{(uint64_t)arrow->messagesDestroyed(), "arrow-messages-destroyed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
                        monitoring.flushBuffer(); },
-    .driverInit = [](ServiceRegistryRef registry, boost::program_options::variables_map const& vm) {
+    .driverInit = [](ServiceRegistry& registry, boost::program_options::variables_map const& vm) {
                        auto config = new RateLimitConfig{};
                        int readers = std::stoll(vm["readers"].as<std::string>());
                        if (vm.count("aod-memory-rate-limit") && vm["aod-memory-rate-limit"].defaulted() == false) {
diff --git a/Framework/Core/src/CallbacksPolicy.cxx b/Framework/Core/src/CallbacksPolicy.cxx
@@ -11,6 +11,7 @@
 #include "Framework/CallbacksPolicy.h"
 #include "Framework/CallbackService.h"
 #include "Framework/CompletionPolicy.h"
+#include "Framework/ServiceRegistryRef.h"
 #include "Framework/TimingInfo.h"
 #include "Framework/Logger.h"
 #include <cstdlib>
diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx
@@ -18,7 +18,7 @@
 #include "Framework/TimesliceIndex.h"
 #include "Framework/DataTakingContext.h"
 #include "Framework/DataSender.h"
-#include "Framework/ServiceRegistry.h"
+#include "Framework/ServiceRegistryRef.h"
 #include "Framework/DeviceSpec.h"
 #include "Framework/LocalRootFileService.h"
 #include "Framework/DataRelayer.h"
@@ -207,7 +207,7 @@ o2::framework::ServiceSpec CommonServices::configurationSpec()
                            ConfigurationFactory::getConfiguration(backend).release()};
     },
     .configure = noConfiguration(),
-    .driverStartup = [](ServiceRegistryRef registry, boost::program_options::variables_map const& vmap) {
+    .driverStartup = [](ServiceRegistry& registry, boost::program_options::variables_map const& vmap) {
       if (vmap.count("configuration") == 0) {
         registry.registerService(ServiceHandle{0, nullptr});
         return;
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -108,7 +108,7 @@ void on_communication_requested(uv_async_t* s)
   state->loopReason |= DeviceState::METRICS_MUST_FLUSH;
 }
 
-DataProcessingDevice::DataProcessingDevice(RunningDeviceRef ref, ServiceRegistryRef registry, ProcessingPolicies& policies)
+DataProcessingDevice::DataProcessingDevice(RunningDeviceRef ref, ServiceRegistry& registry, ProcessingPolicies& policies)
   : mSpec{registry.get<RunningWorkflowInfo const>().devices[ref.index]},
     mState{registry.get<DeviceState>()},
     mInit{mSpec.algorithm.onInit},
diff --git a/Framework/Core/src/DataSender.cxx b/Framework/Core/src/DataSender.cxx
@@ -79,7 +79,7 @@ std::unique_ptr<fair::mq::Message> DataSender::create(RouteIndex routeIndex)
 
 void DataSender::send(fair::mq::Parts& parts, ChannelIndex channelIndex)
 {
-  mRegistry.preSendingMessagesCallbacks(mRegistry, parts, channelIndex);
+  mRegistry.preSendingMessagesCallbacks(parts, channelIndex);
   mPolicy.send(mProxy, parts, channelIndex, mRegistry);
 }
 
diff --git a/Framework/Core/src/ExternalFairMQDeviceProxy.cxx b/Framework/Core/src/ExternalFairMQDeviceProxy.cxx
@@ -433,7 +433,7 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
     // will be multiple channels. At least we throw a more informative exception.
     // fair::mq::Device calls the custom init before the channels have been configured
     // so we do the check before starting in a dedicated callback
-    auto channelConfigurationChecker = [device, deviceName, &services = ctx.services()]() {
+    auto channelConfigurationChecker = [device, deviceName, services = ctx.services()]() {
       auto& deviceState = services.get<DeviceState>();
       channels.clear();
       numberOfEoS.clear();
diff --git a/Framework/Core/src/ServiceRegistry.cxx b/Framework/Core/src/ServiceRegistry.cxx
@@ -9,6 +9,7 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 #include "Framework/ServiceRegistry.h"
+#include "Framework/ServiceRegistryRef.h"
 #include "Framework/Tracing.h"
 #include "Framework/Logger.h"
 #include <iostream>
@@ -28,7 +29,7 @@ ServiceRegistry::ServiceRegistry(ServiceRegistry const& other)
   }
 }
 
-ServiceRegistryRef ServiceRegistry::operator=(ServiceRegistry const& other)
+ServiceRegistry& ServiceRegistry::operator=(ServiceRegistry const& other)
 {
   for (size_t i = 0; i < MAX_SERVICES; ++i) {
     mServicesKey[i].store(other.mServicesKey[i].load());
@@ -234,18 +235,18 @@ void ServiceRegistry::preExitCallbacks()
   // FIXME: we need to call the callback only once for the global services
   /// I guess...
   for (auto exitHandle = mPreExitHandles.rbegin(); exitHandle != mPreExitHandles.rend(); ++exitHandle) {
-    exitHandle->callback(*this, exitHandle->service);
+    exitHandle->callback(ServiceRegistryRef{*this}, exitHandle->service);
   }
 }
 
-void ServiceRegistry::domainInfoUpdatedCallback(ServiceRegistryRef registry, size_t oldestPossibleTimeslice, ChannelIndex channelIndex)
+void ServiceRegistry::domainInfoUpdatedCallback(ServiceRegistry& registry, size_t oldestPossibleTimeslice, ChannelIndex channelIndex)
 {
   for (auto& handle : mDomainInfoHandles) {
     handle.callback(*this, oldestPossibleTimeslice, channelIndex);
   }
 }
 
-void ServiceRegistry::preSendingMessagesCallbacks(ServiceRegistryRef registry, fair::mq::Parts& parts, ChannelIndex channelIndex)
+void ServiceRegistry::preSendingMessagesCallbacks(ServiceRegistry& registry, fair::mq::Parts& parts, ChannelIndex channelIndex)
 {
   for (auto& handle : mPreSendingMessagesHandles) {
     handle.callback(*this, parts, channelIndex);
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx
@@ -1042,7 +1042,7 @@ void doDefaultWorkflowTerminationHook()
   // LOG(info) << "Process " << getpid() << " is exiting.";
 }
 
-int doChild(int argc, char** argv, ServiceRegistryRef serviceRegistry,
+int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
             RunningWorkflowInfo const& runningWorkflow,
             RunningDeviceRef ref,
             ProcessingPolicies processingPolicies,
diff --git a/Framework/GUISupport/src/Plugin.cxx b/Framework/GUISupport/src/Plugin.cxx
@@ -68,7 +68,7 @@ struct ImGUIDebugGUI : o2::framework::DebugGUI {
     o2::framework::gui::charIn(key);
   }
 
-  void* initGUI(char const* windowTitle, ServiceRegistryRef registry_) override
+  void* initGUI(char const* windowTitle, ServiceRegistry& registry_) override
   {
     registry = &registry_;
     return o2::framework::initGUI(windowTitle);
diff --git a/Framework/GUISupport/src/SpyService.h b/Framework/GUISupport/src/SpyService.h
@@ -14,6 +14,7 @@
 
 #include "Framework/ServiceHandle.h"
 #include "Framework/ServiceSpec.h"
+#include "Framework/ServiceRegistryRef.h"
 #include <mutex>
 #include <vector>
 #include <fairmq/FwdDecls.h>

Original file line number	Diff line number	Diff line change
`@@ -370,10 +370,6 @@ struct ServiceRegistry {`
`370`	`370`	`}`
`371`	`371`	`};`
`372`	`372`
`373`		`-// This is to migrate the code in QC.`
`374`		`-// FIXME: move this to the proper smart reference class once done`
`375`		`-using ServiceRegistryRef = o2::framework::ServiceRegistry &;`
`376`		`-`
`377`	`373`	`} // namespace o2::framework`
`378`	`374`
`379`	`375`	`#endif // O2_FRAMEWORK_SERVICEREGISTRY_H_`