Skip to content

Commit 45d1a0a

Browse files
ktfktf
committed
DPL: use ServiceRegistry Salt and Context
For now the behaviour is the same, but with the multithreading we will have a proper .streamId and eventually a .dataProcessorId
1 parent d76bb43 commit 45d1a0a

3 files changed

Lines changed: 64 additions & 49 deletions

File tree

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

Lines changed: 25 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -212,7 +212,7 @@ struct ServiceRegistry {
212212
/// hash used to identify the service, @a service is
213213
/// a type erased pointer to the service itself.
214214
/// This method is supposed to be thread safe
215-
void registerService(ServiceTypeHash typeHash, void* service, ServiceKind kind, uint64_t threadId, char const* name = nullptr) const;
215+
void registerService(ServiceTypeHash typeHash, void* service, ServiceKind kind, Salt salt, char const* name = nullptr) const;
216216

217217
// Lookup a given @a typeHash for a given @a threadId at
218218
// a unique (per typeHash) location. There might
@@ -222,12 +222,13 @@ struct ServiceRegistry {
222222
// as guaranteed by the atomic, mServicesKey[i + id] will
223223
// either be 0 or the final value.
224224
// This method should NEVER register a new service, event when requested.
225-
int getPos(ServiceTypeHash typeHash, uint64_t threadId) const
225+
int getPos(ServiceTypeHash typeHash, Salt salt) const
226226
{
227-
auto threadHashId = (typeHash.hash ^ threadId) & MAX_SERVICES_MASK;
227+
InstanceId instanceId = instanceFromTypeSalt(typeHash, salt);
228+
Index index = indexFromInstance(instanceId);
228229
for (uint8_t i = 0; i < MAX_DISTANCE; ++i) {
229-
if (mServicesKey[i + threadHashId].load() == typeHash.hash) {
230-
return i + threadHashId;
230+
if (mServicesKey[i + index.index].load() == typeHash.hash) {
231+
return i + index.index;
231232
}
232233
}
233234
return -1;
@@ -241,15 +242,15 @@ struct ServiceRegistry {
241242
// if the service is not a stream service and the global
242243
// zero service is available.
243244
// Use this API only if you know what you are doing.
244-
void* get(ServiceTypeHash typeHash, uint64_t threadId, ServiceKind kind, char const* name = nullptr) const
245+
void* get(ServiceTypeHash typeHash, Salt salt, ServiceKind kind, char const* name = nullptr) const
245246
{
246247
// Look for the service. If found, return it.
247248
// Notice how due to threading issues, we might
248249
// find it with getPos, but the value can still
249250
// be nullptr.
250-
auto pos = getPos(typeHash, threadId);
251-
if (pos != -1 && mServicesMeta[pos].kind == ServiceKind::Stream && mServicesMeta[pos].threadId != threadId) {
252-
throwError(runtime_error_f("Inconsistent registry for thread %d. Expected %d", threadId, mServicesMeta[pos].threadId));
251+
auto pos = getPos(typeHash, salt);
252+
if (pos != -1 && mServicesMeta[pos].kind == ServiceKind::Stream && mServicesMeta[pos].salt.value != salt.value) {
253+
throwError(runtime_error_f("Inconsistent registry for thread %d. Expected %d", salt.context.streamId, mServicesMeta[pos].salt.context.streamId));
253254
O2_BUILTIN_UNREACHABLE();
254255
}
255256

@@ -264,12 +265,12 @@ struct ServiceRegistry {
264265
// We are looking up a service which is not of
265266
// stream kind and was not looked up by this thread
266267
// before.
267-
if (threadId != 0) {
268-
int pos = getPos(typeHash, 0);
268+
if (salt.value != GLOBAL_CONTEXT_SALT.value) {
269+
int pos = getPos(typeHash, GLOBAL_CONTEXT_SALT);
269270
if (pos != -1 && kind != ServiceKind::Stream) {
270271
mServicesKey[pos].load();
271272
std::atomic_thread_fence(std::memory_order_acquire);
272-
registerService(typeHash, mServicesValue[pos], kind, threadId, name);
273+
registerService(typeHash, mServicesValue[pos], kind, salt, name);
273274
}
274275
if (pos != -1) {
275276
mServicesKey[pos].load();
@@ -289,13 +290,14 @@ struct ServiceRegistry {
289290
{
290291
auto tid = std::this_thread::get_id();
291292
std::hash<std::thread::id> hasher;
292-
ServiceRegistry::registerService({handle.hash}, handle.instance, handle.kind, hasher(tid), handle.name.c_str());
293+
Salt salt{Context{.streamId = (short)hasher(tid)}};
294+
ServiceRegistry::registerService({handle.hash}, handle.instance, handle.kind, salt, handle.name.c_str());
293295
}
294296

295297
mutable std::vector<ServiceSpec> mSpecs;
296298
mutable std::array<std::atomic<uint32_t>, MAX_SERVICES + MAX_DISTANCE> mServicesKey;
297299
mutable std::array<void*, MAX_SERVICES + MAX_DISTANCE> mServicesValue;
298-
mutable std::array<ServiceMeta, MAX_SERVICES + MAX_DISTANCE> mServicesMeta;
300+
mutable std::array<Meta, MAX_SERVICES + MAX_DISTANCE> mServicesMeta;
299301
mutable std::array<std::atomic<bool>, MAX_SERVICES + MAX_DISTANCE> mServicesBooked;
300302

301303
/// @deprecated old API to be substituted with the ServiceHandle one
@@ -310,7 +312,8 @@ struct ServiceRegistry {
310312
constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<I>()};
311313
auto tid = std::this_thread::get_id();
312314
std::hash<std::thread::id> hasher;
313-
ServiceRegistry::registerService(typeHash, reinterpret_cast<void*>(service), K, hasher(tid), typeid(C).name());
315+
Salt salt = Salt{Context{.streamId = (short)hasher(tid)}};
316+
ServiceRegistry::registerService(typeHash, reinterpret_cast<void*>(service), K, salt, typeid(C).name());
314317
}
315318

316319
/// @deprecated old API to be substituted with the ServiceHandle one
@@ -325,7 +328,8 @@ struct ServiceRegistry {
325328
constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<I const>()};
326329
auto tid = std::this_thread::get_id();
327330
std::hash<std::thread::id> hasher;
328-
this->registerService(typeHash, reinterpret_cast<void*>(const_cast<C*>(service)), K, hasher(tid), typeid(C).name());
331+
Salt salt = Salt{Context{.streamId = (short)hasher(tid)}};
332+
this->registerService(typeHash, reinterpret_cast<void*>(const_cast<C*>(service)), K, salt, typeid(C).name());
329333
}
330334

331335
/// Check if service of type T is currently active.
@@ -335,10 +339,11 @@ struct ServiceRegistry {
335339
constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<T>()};
336340
auto tid = std::this_thread::get_id();
337341
std::hash<std::thread::id> hasher;
338-
if (this->getPos(typeHash, 0) != -1) {
342+
if (this->getPos(typeHash, GLOBAL_CONTEXT_SALT) != -1) {
339343
return true;
340344
}
341-
auto result = this->getPos(typeHash, hasher(tid)) != -1;
345+
Salt salt = Salt{Context{.streamId = (short)hasher(tid)}};
346+
auto result = this->getPos(typeHash, salt) != -1;
342347
return result;
343348
}
344349

@@ -351,7 +356,8 @@ struct ServiceRegistry {
351356
constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<T>()};
352357
auto tid = std::this_thread::get_id();
353358
std::hash<std::thread::id> hasher;
354-
auto ptr = this->get(typeHash, hasher(tid), ServiceKind::Serial, typeid(T).name());
359+
Salt salt = Salt{Context{.streamId = (short)hasher(tid)}};
360+
auto ptr = this->get(typeHash, salt, ServiceKind::Serial, typeid(T).name());
355361
if (O2_BUILTIN_LIKELY(ptr != nullptr)) {
356362
if constexpr (std::is_const_v<T>) {
357363
return *reinterpret_cast<T const*>(ptr);

‎Framework/Core/src/ServiceRegistry.cxx‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -57,28 +57,29 @@ ServiceRegistry::ServiceRegistry()
5757
/// hash used to identify the service, @a service is
5858
/// a type erased pointer to the service itself.
5959
/// This method is supposed to be thread safe
60-
void ServiceRegistry::registerService(ServiceTypeHash typeHash, void* service, ServiceKind kind, uint64_t threadId, const char* name) const
60+
void ServiceRegistry::registerService(ServiceTypeHash typeHash, void* service, ServiceKind kind, Salt salt, const char* name) const
6161
{
62-
hash_type threadHashId = (typeHash.hash ^ threadId) & MAX_SERVICES_MASK;
62+
InstanceId id = instanceFromTypeSalt(typeHash, salt);
63+
Index index = indexFromInstance(id);
6364
// If kind is not stream, there is only one copy of our service.
6465
// So we look if it is already registered and reused it if it is.
6566
// If not, we register it as thread id 0 and as the passed one.
66-
if (kind != ServiceKind::Stream && threadId != 0) {
67-
void* oldService = this->get(typeHash, 0, kind);
67+
if (kind != ServiceKind::Stream && salt.context.streamId != 0) {
68+
void* oldService = this->get(typeHash, GLOBAL_CONTEXT_SALT, kind);
6869
if (oldService == nullptr) {
69-
registerService(typeHash, service, kind, 0);
70+
registerService(typeHash, service, kind, GLOBAL_CONTEXT_SALT);
7071
} else {
7172
service = oldService;
7273
}
7374
}
7475
for (uint8_t i = 0; i < MAX_DISTANCE; ++i) {
7576
// If the service slot was not taken, take it atomically
7677
bool expected = false;
77-
if (mServicesBooked[i + threadHashId].compare_exchange_strong(expected, true,
78+
if (mServicesBooked[i + index.index].compare_exchange_strong(expected, true,
7879
std::memory_order_seq_cst)) {
79-
mServicesValue[i + threadHashId] = service;
80-
mServicesMeta[i + threadHashId] = ServiceMeta{kind, threadId};
81-
mServicesKey[i + threadHashId] = typeHash.hash;
80+
mServicesValue[i + index.index] = service;
81+
mServicesMeta[i + index.index] = Meta{kind, salt};
82+
mServicesKey[i + index.index] = typeHash.hash;
8283
std::atomic_thread_fence(std::memory_order_release);
8384
return;
8485
}
@@ -92,7 +93,7 @@ void ServiceRegistry::declareService(ServiceSpec const& spec, DeviceState& state
9293
// Services which are not stream must have a single instance created upfront.
9394
if (spec.kind != ServiceKind::Stream) {
9495
ServiceHandle handle = spec.init(*this, state, options);
95-
this->registerService({handle.hash}, handle.instance, handle.kind, 0, handle.name.c_str());
96+
this->registerService({handle.hash}, handle.instance, handle.kind, GLOBAL_CONTEXT_SALT, handle.name.c_str());
9697
this->bindService(spec, handle.instance);
9798
}
9899
}

‎Framework/Core/test/test_Services.cxx‎

Lines changed: 28 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -42,11 +42,11 @@ BOOST_AUTO_TEST_CASE(TestServiceRegistry)
4242
};
4343

4444
struct InterfaceC {
45-
virtual bool method() const = 0;
45+
[[nodiscard]] virtual bool method() const = 0;
4646
};
4747

4848
struct ConcreteC : InterfaceC {
49-
bool method() const final { return false; }
49+
[[nodiscard]] bool method() const final { return false; }
5050
};
5151

5252
ServiceRegistry registry;
@@ -90,18 +90,25 @@ struct DummyService {
9090
int threadId;
9191
};
9292

93+
namespace o2::framework
94+
{
95+
static ServiceRegistry::Salt salt_0 = ServiceRegistry::Salt{ServiceRegistry::Context{0,0}};
96+
static ServiceRegistry::Salt salt_1 = ServiceRegistry::Salt{ServiceRegistry::Context{1,0}};
97+
static ServiceRegistry::Salt salt_2 = ServiceRegistry::Salt{ServiceRegistry::Context{2,0}};
98+
}
99+
93100
BOOST_AUTO_TEST_CASE(TestSerialServices)
94101
{
95102
using namespace o2::framework;
96103
ServiceRegistry registry;
97104

98105
DummyService t0{0};
99106
/// We register it pretending to be on thread 0
100-
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Serial, 0);
107+
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Serial, salt_0);
101108

102-
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 0, ServiceKind::Serial));
103-
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 1, ServiceKind::Serial));
104-
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 2, ServiceKind::Serial));
109+
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_0, ServiceKind::Serial));
110+
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_1, ServiceKind::Serial));
111+
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_2, ServiceKind::Serial));
105112
BOOST_CHECK_EQUAL(tt0->threadId, 0);
106113
BOOST_CHECK_EQUAL(tt1->threadId, 0);
107114
BOOST_CHECK_EQUAL(tt2->threadId, 0);
@@ -114,11 +121,11 @@ BOOST_AUTO_TEST_CASE(TestGlobalServices)
114121

115122
DummyService t0{0};
116123
/// We register it pretending to be on thread 0
117-
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Global, 0);
124+
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Global, salt_0);
118125

119-
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 0, ServiceKind::Serial));
120-
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 1, ServiceKind::Serial));
121-
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 2, ServiceKind::Serial));
126+
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_0, ServiceKind::Serial));
127+
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_1, ServiceKind::Serial));
128+
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_2, ServiceKind::Serial));
122129
BOOST_CHECK_EQUAL(tt0->threadId, 0);
123130
BOOST_CHECK_EQUAL(tt1->threadId, 0);
124131
BOOST_CHECK_EQUAL(tt2->threadId, 0);
@@ -131,16 +138,17 @@ BOOST_AUTO_TEST_CASE(TestGlobalServices02)
131138

132139
DummyService t0{1};
133140
/// We register it pretending to be on thread 0
134-
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Global, 1);
141+
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Global, salt_1);
135142

136-
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 0, ServiceKind::Global));
137-
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 1, ServiceKind::Global));
138-
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 2, ServiceKind::Global));
143+
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_0, ServiceKind::Global));
144+
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_1, ServiceKind::Global));
145+
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_2, ServiceKind::Global));
139146
BOOST_CHECK_EQUAL(tt0->threadId, 1);
140147
BOOST_CHECK_EQUAL(tt1->threadId, 1);
141148
BOOST_CHECK_EQUAL(tt2->threadId, 1);
142149
}
143150

151+
144152
BOOST_AUTO_TEST_CASE(TestStreamServices)
145153
{
146154
using namespace o2::framework;
@@ -150,13 +158,13 @@ BOOST_AUTO_TEST_CASE(TestStreamServices)
150158
DummyService t1{1};
151159
DummyService t2{2};
152160
/// We register it pretending to be on thread 0
153-
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Stream, 0);
154-
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t1, ServiceKind::Stream, 1);
155-
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t2, ServiceKind::Stream, 2);
161+
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t0, ServiceKind::Stream, ServiceRegistry::Salt{ServiceRegistry::Context{0,0}});
162+
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t1, ServiceKind::Stream, ServiceRegistry::Salt{ServiceRegistry::Context{1,0}});
163+
registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &t2, ServiceKind::Stream, ServiceRegistry::Salt{ServiceRegistry::Context{2,0}});
156164

157-
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 0, ServiceKind::Stream));
158-
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 1, ServiceKind::Stream));
159-
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, 2, ServiceKind::Stream));
165+
auto tt0 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_0, ServiceKind::Stream));
166+
auto tt1 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_1, ServiceKind::Stream));
167+
auto tt2 = reinterpret_cast<DummyService*>(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_2, ServiceKind::Stream));
160168
BOOST_CHECK_EQUAL(tt0->threadId, 0);
161169
BOOST_CHECK_EQUAL(tt1->threadId, 1);
162170
BOOST_CHECK_EQUAL(tt2->threadId, 2);

0 commit comments

Comments
 (0)