Skip to content

Commit 37b06fc

Browse files
ktfktf
committed
DPL: registering service needs a salt as well
1 parent db2329c commit 37b06fc

11 files changed

Lines changed: 63 additions & 47 deletions

‎Framework/Core/include/Framework/ServiceRegistry.h‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,9 @@ struct ServiceRegistry {
209209
/// If it is of kind "Stream" we will create the Service only
210210
/// when requested by a given thread. This function is not
211211
/// thread safe.
212-
void declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options);
212+
/// @a salt is used to create the service in the proper context
213+
/// FIXME: for now we create everything in the global context
214+
void declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options, ServiceRegistry::Salt salt = ServiceRegistry::threadSalt());
213215

214216
/// Bind the callbacks of a service spec to a given service.
215217
void bindService(ServiceSpec const& spec, void* service);
@@ -292,9 +294,9 @@ struct ServiceRegistry {
292294
}
293295

294296
/// Register a service given an handle
295-
void registerService(ServiceHandle handle)
297+
void registerService(ServiceHandle handle, Salt salt)
296298
{
297-
ServiceRegistry::registerService({handle.hash}, handle.instance, handle.kind, ServiceRegistry::threadSalt(), handle.name.c_str());
299+
ServiceRegistry::registerService({handle.hash}, handle.instance, handle.kind, salt, handle.name.c_str());
298300
}
299301

300302
mutable std::vector<ServiceSpec> mSpecs;
@@ -305,28 +307,28 @@ struct ServiceRegistry {
305307

306308
/// @deprecated old API to be substituted with the ServiceHandle one
307309
template <class I, class C, enum ServiceKind K = ServiceKind::Serial>
308-
void registerService(C* service)
310+
void registerService(C* service, Salt salt)
309311
{
310312
// This only works for concrete implementations of the type T.
311313
// We need type elision as we do not want to know all the services in
312314
// advance
313315
static_assert(std::is_base_of<I, C>::value == true,
314316
"Registered service is not derived from declared interface");
315317
constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<I>()};
316-
ServiceRegistry::registerService(typeHash, reinterpret_cast<void*>(service), K, ServiceRegistry::threadSalt(), typeid(C).name());
318+
ServiceRegistry::registerService(typeHash, reinterpret_cast<void*>(service), K, salt, typeid(C).name());
317319
}
318320

319321
/// @deprecated old API to be substituted with the ServiceHandle one
320322
template <class I, class C, enum ServiceKind K = ServiceKind::Serial>
321-
void registerService(C const* service)
323+
void registerService(C const* service, Salt salt)
322324
{
323325
// This only works for concrete implementations of the type T.
324326
// We need type elision as we do not want to know all the services in
325327
// advance
326328
static_assert(std::is_base_of<I, C>::value == true,
327329
"Registered service is not derived from declared interface");
328330
constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<I const>()};
329-
this->registerService(typeHash, reinterpret_cast<void*>(const_cast<C*>(service)), K, ServiceRegistry::threadSalt(), typeid(C).name());
331+
this->registerService(typeHash, reinterpret_cast<void*>(const_cast<C*>(service)), K, salt, typeid(C).name());
330332
}
331333

332334
/// Check if service of type T is currently active.

‎Framework/Core/include/Framework/ServiceRegistryRef.h‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,20 @@ class ServiceRegistryRef
6464
mRegistry.preSendingMessagesCallbacks(mRegistry, parts, channelindex);
6565
}
6666

67+
void registerService(ServiceTypeHash typeHash, void* service, ServiceKind kind, char const* name = nullptr) const {
68+
mRegistry.registerService(typeHash, service, kind, mSalt, name);
69+
}
70+
71+
/// Register a service given an handle, notice how
72+
/// the service will be created in the current salt,
73+
/// so that from a dataprocessor you cannot create a service
74+
/// globally, or in a stream you cannot create services for
75+
/// a dataprocessor.
76+
void registerService(ServiceHandle handle)
77+
{
78+
mRegistry.registerService({handle.hash}, handle.instance, handle.kind, mSalt, handle.name.c_str());
79+
}
80+
6781
private:
6882
ServiceRegistry& mRegistry;
6983
ServiceRegistry::Salt mSalt;

‎Framework/Core/include/Framework/ServiceSpec.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -96,10 +96,10 @@ using ServicePostDispatching = void (*)(ProcessingContext&, void*);
9696
using ServicePostForwarding = void (*)(ProcessingContext&, void*);
9797

9898
/// Callback invoked when the driver enters the init phase.
99-
using ServiceDriverInit = void (*)(ServiceRegistry&, boost::program_options::variables_map const&);
99+
using ServiceDriverInit = void (*)(ServiceRegistryRef, boost::program_options::variables_map const&);
100100

101101
/// Callback invoked when the driver enters the init phase.
102-
using ServiceDriverStartup = void (*)(ServiceRegistry&, boost::program_options::variables_map const&);
102+
using ServiceDriverStartup = void (*)(ServiceRegistryRef, boost::program_options::variables_map const&);
103103

104104
/// Callback invoked when we inject internal devices in the topology
105105
using ServiceTopologyInject = void (*)(WorkflowSpecNode&, ConfigContext&);

‎Framework/Core/src/ArrowSupport.cxx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -364,7 +364,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
364364
monitoring.send(Metric{(uint64_t)arrow->bytesDestroyed(), "arrow-bytes-destroyed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
365365
monitoring.send(Metric{(uint64_t)arrow->messagesDestroyed(), "arrow-messages-destroyed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
366366
monitoring.flushBuffer(); },
367-
.driverInit = [](ServiceRegistry& registry, boost::program_options::variables_map const& vm) {
367+
.driverInit = [](ServiceRegistryRef registry, boost::program_options::variables_map const& vm) {
368368
auto config = new RateLimitConfig{};
369369
int readers = std::stoll(vm["readers"].as<std::string>());
370370
if (vm.count("aod-memory-rate-limit") && vm["aod-memory-rate-limit"].defaulted() == false) {

‎Framework/Core/src/CommonServices.cxx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,7 @@ o2::framework::ServiceSpec CommonServices::configurationSpec()
209209
ConfigurationFactory::getConfiguration(backend).release()};
210210
},
211211
.configure = noConfiguration(),
212-
.driverStartup = [](ServiceRegistry& registry, boost::program_options::variables_map const& vmap) {
212+
.driverStartup = [](ServiceRegistryRef registry, boost::program_options::variables_map const& vmap) {
213213
if (vmap.count("configuration") == 0) {
214214
registry.registerService(ServiceHandle{0, nullptr});
215215
return;

‎Framework/Core/src/ControlWebSocketHandler.cxx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ void ControlWebSocketHandler::endChunk()
7878
}
7979
size_t timestamp = uv_now(mContext.loop);
8080
for (auto& callback : *mContext.metricProcessingCallbacks) {
81-
callback(*mContext.registry, *mContext.metrics, *mContext.specs, *mContext.infos, mContext.driver->metrics, timestamp);
81+
callback(mContext.registry, *mContext.metrics, *mContext.specs, *mContext.infos, mContext.driver->metrics, timestamp);
8282
}
8383
for (auto& metricsInfo : *mContext.metrics) {
8484
std::fill(metricsInfo.changed.begin(), metricsInfo.changed.end(), false);

‎Framework/Core/src/DriverServerContext.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@ struct ServiceRegistry;
3030
struct GuiCallbackContext;
3131

3232
struct DriverServerContext {
33+
ServiceRegistryRef registry;
3334
uv_loop_t* loop = nullptr;
34-
ServiceRegistry* registry = nullptr;
3535
std::vector<DeviceControl>* controls = nullptr;
3636
std::vector<DeviceInfo>* infos = nullptr;
3737
std::vector<DeviceSpec>* specs = nullptr;

‎Framework/Core/src/ServiceRegistry.cxx‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -88,13 +88,13 @@ void ServiceRegistry::registerService(ServiceTypeHash typeHash, void* service, S
8888
throw runtime_error_f("Unable to find a spot in the registry for service %d. Make sure you use const / non-const correctly.", typeHash.hash);
8989
}
9090

91-
void ServiceRegistry::declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options)
91+
void ServiceRegistry::declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options, ServiceRegistry::Salt salt)
9292
{
9393
mSpecs.push_back(spec);
9494
// Services which are not stream must have a single instance created upfront.
9595
if (spec.kind != ServiceKind::Stream) {
96-
ServiceHandle handle = spec.init(*this, state, options);
97-
this->registerService({handle.hash}, handle.instance, handle.kind, GLOBAL_CONTEXT_SALT, handle.name.c_str());
96+
ServiceHandle handle = spec.init({*this}, state, options);
97+
this->registerService({handle.hash}, handle.instance, handle.kind, salt, handle.name.c_str());
9898
this->bindService(spec, handle.instance);
9999
}
100100
}

‎Framework/Core/src/runDataProcessing.cxx‎

Lines changed: 23 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -526,7 +526,7 @@ struct ControlWebSocketHandler : public WebSocketHandler {
526526
}
527527
size_t timestamp = uv_now(mContext.loop);
528528
for (auto& callback : *mContext.metricProcessingCallbacks) {
529-
callback(*mContext.registry, *mContext.metrics, *mContext.specs, *mContext.infos, mContext.driver->metrics, timestamp);
529+
callback(mContext.registry, *mContext.metrics, *mContext.specs, *mContext.infos, mContext.driver->metrics, timestamp);
530530
}
531531
for (auto& metricsInfo : *mContext.metrics) {
532532
std::fill(metricsInfo.changed.begin(), metricsInfo.changed.end(), false);
@@ -1112,29 +1112,29 @@ int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
11121112
&processingPolicies,
11131113
&deviceContext,
11141114
&loop](fair::mq::DeviceRunner& r) {
1115+
ServiceRegistryRef serviceRef = {serviceRegistry};
11151116
simpleRawDeviceService = std::make_unique<SimpleRawDeviceService>(nullptr, spec);
1116-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<RawDeviceService>(simpleRawDeviceService.get()));
1117+
serviceRef.registerService(ServiceRegistryHelpers::handleForService<RawDeviceService>(simpleRawDeviceService.get()));
11171118

11181119
deviceState = std::make_unique<DeviceState>();
11191120
deviceState->loop = loop;
11201121
deviceState->tracingFlags = DeviceStateHelpers::parseTracingFlags(r.fConfig.GetPropertyAsString("dpl-tracing-flags"));
1121-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<DeviceState>(deviceState.get()));
1122+
serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceState>(deviceState.get()));
11221123

11231124
quotaEvaluator = std::make_unique<ComputingQuotaEvaluator>(uv_now(loop));
1124-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<ComputingQuotaEvaluator>(quotaEvaluator.get()));
1125+
serviceRef.registerService(ServiceRegistryHelpers::handleForService<ComputingQuotaEvaluator>(quotaEvaluator.get()));
11251126

11261127
deviceContext = std::make_unique<DeviceContext>();
1127-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<DeviceSpec const>(&spec));
1128-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<RunningWorkflowInfo const>(&runningWorkflow));
1129-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<DeviceContext>(deviceContext.get()));
1128+
serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceSpec const>(&spec));
1129+
serviceRef.registerService(ServiceRegistryHelpers::handleForService<RunningWorkflowInfo const>(&runningWorkflow));
1130+
serviceRef.registerService(ServiceRegistryHelpers::handleForService<DeviceContext>(deviceContext.get()));
11301131

11311132
// The decltype stuff is to be able to compile with both new and old
11321133
// FairMQ API (one which uses a shared_ptr, the other one a unique_ptr.
11331134
decltype(r.fDevice) device;
11341135
device = make_matching<decltype(device), DataProcessingDevice>(ref, serviceRegistry, processingPolicies);
11351136

1136-
ServiceRegistryRef ref{serviceRegistry};
1137-
ref.get<RawDeviceService>().setDevice(device.get());
1137+
serviceRef.get<RawDeviceService>().setDevice(device.get());
11381138
r.fDevice = std::move(device);
11391139
fair::Logger::SetConsoleColor(false);
11401140

@@ -1144,7 +1144,7 @@ int doChild(int argc, char** argv, ServiceRegistry& serviceRegistry,
11441144
serviceRegistry.declareService(service, *deviceState.get(), r.fConfig);
11451145
}
11461146
if (ResourcesMonitoringHelper::isResourcesMonitoringEnabled(spec.resourceMonitoringInterval)) {
1147-
ref.get<Monitoring>().enableProcessMonitoring(spec.resourceMonitoringInterval, {PmMeasurement::Cpu, PmMeasurement::Mem, PmMeasurement::Smaps});
1147+
serviceRef.get<Monitoring>().enableProcessMonitoring(spec.resourceMonitoringInterval, {PmMeasurement::Cpu, PmMeasurement::Mem, PmMeasurement::Smaps});
11481148
}
11491149
};
11501150

@@ -1339,7 +1339,8 @@ int runStateMachine(DataProcessorSpecs const& workflow,
13391339
service.driverStartup(serviceRegistry, varmap);
13401340
}
13411341

1342-
serviceRegistry.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
1342+
ServiceRegistryRef ref{serviceRegistry};
1343+
ref.registerService(ServiceRegistryHelpers::handleForService<DevicesManager>(devicesManager));
13431344

13441345
GuiCallbackContext guiContext;
13451346
guiContext.plugin = debugGUI;
@@ -1350,17 +1351,17 @@ int runStateMachine(DataProcessorSpecs const& workflow,
13501351

13511352
// This is to make sure we can process metrics, commands, configuration
13521353
// changes coming from websocket (or even via any standard uv_stream_t, I guess).
1353-
DriverServerContext serverContext;
1354-
serverContext.registry = &serviceRegistry;
1355-
serverContext.loop = loop;
1356-
serverContext.controls = &controls;
1357-
serverContext.infos = &infos;
1358-
serverContext.specs = &runningWorkflow.devices;
1359-
serverContext.metrics = &metricsInfos;
1360-
serverContext.driver = &driverInfo;
1361-
serverContext.metricProcessingCallbacks = &metricProcessingCallbacks;
1362-
serverContext.gui = &guiContext;
1363-
serverContext.isDriver = frameworkId.empty();
1354+
DriverServerContext serverContext{
1355+
.registry = {serviceRegistry},
1356+
.loop = loop,
1357+
.controls = &controls,
1358+
.infos = &infos,
1359+
.specs = &runningWorkflow.devices,
1360+
.metrics = &metricsInfos,
1361+
.metricProcessingCallbacks = &metricProcessingCallbacks,
1362+
.driver = &driverInfo,
1363+
.gui = &guiContext,
1364+
.isDriver = frameworkId.empty()};
13641365

13651366
uv_tcp_t serverHandle;
13661367
serverHandle.data = &serverContext;

‎Framework/Core/test/test_Services.cxx‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -50,12 +50,13 @@ BOOST_AUTO_TEST_CASE(TestServiceRegistry)
5050
};
5151

5252
ServiceRegistry registry;
53+
ServiceRegistryRef ref{registry};
5354
ConcreteA serviceA;
5455
ConcreteB serviceB;
5556
ConcreteC const serviceC;
56-
registry.registerService(ServiceRegistryHelpers::handleForService<InterfaceA>(&serviceA));
57-
registry.registerService(ServiceRegistryHelpers::handleForService<InterfaceB>(&serviceB));
58-
registry.registerService(ServiceRegistryHelpers::handleForService<InterfaceC const>(&serviceC));
57+
ref.registerService(ServiceRegistryHelpers::handleForService<InterfaceA>(&serviceA));
58+
ref.registerService(ServiceRegistryHelpers::handleForService<InterfaceB>(&serviceB));
59+
ref.registerService(ServiceRegistryHelpers::handleForService<InterfaceC const>(&serviceC));
5960
BOOST_CHECK(registry.get<InterfaceA>(ServiceRegistry::threadSalt()).method() == true);
6061
BOOST_CHECK(registry.get<InterfaceB>(ServiceRegistry::threadSalt()).method() == false);
6162
BOOST_CHECK(registry.get<InterfaceC const>(ServiceRegistry::threadSalt()).method() == false);
@@ -70,8 +71,9 @@ BOOST_AUTO_TEST_CASE(TestCallbackService)
7071
{
7172
using namespace o2::framework;
7273
ServiceRegistry registry;
74+
ServiceRegistryRef ref{registry};
7375
auto service = std::make_unique<CallbackService>();
74-
registry.registerService(ServiceRegistryHelpers::handleForService<CallbackService>(service.get()));
76+
ref.registerService(ServiceRegistryHelpers::handleForService<CallbackService>(service.get()));
7577

7678
// the callback simply sets the captured variable to indicated that it was called
7779
bool cbCalled = false;

0 commit comments

Comments
 (0)