Skip to content

Commit ebcc8b7

Browse files
ktfktf
committed
DPL: give each stream its own slot in the service registry
The registry derives a service's slot from the low 8 bits of type.hash ^ valueFromSalt(salt), but valueFromSalt puts streamId at bit 16, so streamId contributes nothing to it: every stream lands on the same slot for a given service, they pile into one MAX_DISTANCE-deep probe window, and the first which does not fit is refused -- with the error naming whichever service lost the race rather than the one which filled the window. Mix the salt into the low bits instead: only the starting slot changes, since valueFromSalt still carries key equality and both getPos and registerService go through instanceFromTypeSalt, and unlike widening the table this raises no new ceiling, the mask staying below bit 16 until MAX_SERVICES passes 65536. The added test registers one instance of the same stream service for each of 32 streams and asks for each of them back; on the old slot calculation it gets seven in before registration is refused.
1 parent 6339a5a commit ebcc8b7

2 files changed

Lines changed: 57 additions & 1 deletion

File tree

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

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -177,7 +177,18 @@ struct ServiceRegistry {
177177

178178
constexpr InstanceId instanceFromTypeSalt(ServiceTypeHash type, Salt salt) const
179179
{
180-
return InstanceId{type.hash ^ valueFromSalt(salt)};
180+
// Fold the whole salt down into the low bits. The slot is the low bits of
181+
// this (see indexFromInstance) while streamId sits at bit 16 of
182+
// valueFromSalt, so using that directly gives every stream the same slot
183+
// for a given service: they pile into one probe window, and once it is
184+
// MAX_DISTANCE deep the next registration is refused -- reported against
185+
// whichever service happened to lose, not the one which filled it.
186+
//
187+
// Widening the table does not help on its own: the mask stays below bit 16
188+
// until MAX_SERVICES passes 65536.
189+
uint32_t mixed = static_cast<uint32_t>(static_cast<uint16_t>(salt.streamId)) * 0x9E3779B9u ^
190+
static_cast<uint32_t>(static_cast<uint16_t>(salt.dataProcessorId));
191+
return InstanceId{type.hash ^ mixed};
181192
}
182193

183194
constexpr Index indexFromInstance(InstanceId id) const

‎Framework/Core/test/test_Services.cxx‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
#include <catch_amalgamated.hpp>
1818
#include <fairmq/ProgOptions.h>
1919
#include <memory>
20+
#include <vector>
2021

2122
TEST_CASE("TestServiceRegistry")
2223
{
@@ -213,6 +214,50 @@ TEST_CASE("TestStreamServices")
213214
REQUIRE_THROWS_AS(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_1_1, ServiceKind::Stream), RuntimeErrorRef);
214215
}
215216

217+
TEST_CASE("TestStreamServicesDoNotShareASlot")
218+
{
219+
using namespace o2::framework;
220+
ServiceRegistry registry;
221+
222+
ServiceSpec spec{.name = "dummy-service",
223+
.uniqueId = CommonServices::simpleServiceId<DummyService>(),
224+
.init = CommonServices::simpleServiceInit<DummyService, DummyService>(),
225+
.configure = CommonServices::noConfiguration(),
226+
.kind = ServiceKind::Stream};
227+
228+
DeviceState state;
229+
fair::mq::ProgOptions options;
230+
registry.declareService(spec, state, options, ServiceRegistry::globalDeviceSalt());
231+
232+
// One instance of the same service per stream, and more streams than a probe
233+
// window is deep. The slot comes from the low bits of the type hash combined
234+
// with the salt, so if the salt's streamId does not reach those bits every one
235+
// of these lands on the same slot, and the first which does not fit in the
236+
// window is refused outright.
237+
constexpr short STREAMS = 32;
238+
std::vector<DummyService> services(STREAMS);
239+
for (short i = 0; i < STREAMS; ++i) {
240+
services[i].threadId = i + 1;
241+
}
242+
243+
for (short i = 0; i < STREAMS; ++i) {
244+
// Refused registration is what a shared slot looks like from here: the
245+
// window fills and the next one has nowhere to go.
246+
REQUIRE_NOTHROW(registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &services[i], ServiceKind::Stream,
247+
ServiceRegistry::Salt{static_cast<short>(i + 1), 0}, "dummy-service",
248+
ServiceRegistry::SpecIndex{0}));
249+
}
250+
251+
// Every stream must get its own instance back, not a neighbour's.
252+
for (short i = 0; i < STREAMS; ++i) {
253+
auto* found = reinterpret_cast<DummyService*>(
254+
registry.get({TypeIdHelpers::uniqueId<DummyService>()},
255+
ServiceRegistry::Salt{static_cast<short>(i + 1), 0}, ServiceKind::Stream));
256+
REQUIRE(found != nullptr);
257+
CHECK(found->threadId == i + 1);
258+
}
259+
}
260+
216261
TEST_CASE("TestServiceRegistryCtor")
217262
{
218263
using namespace o2::framework;

0 commit comments

Comments
 (0)