From ebcc8b7d35b578cfa9854b01cf92bb3bb871a5ef Mon Sep 17 00:00:00 2001
From: Giulio Eulisse <10544+ktf@users.noreply.github.com>
Date: Fri, 14 Aug 2026 15:42:56 +0200
Subject: [PATCH] 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.
---
 .../Core/include/Framework/ServiceRegistry.h  | 13 +++++-
 Framework/Core/test/test_Services.cxx         | 45 +++++++++++++++++++
 2 files changed, 57 insertions(+), 1 deletion(-)

diff --git a/Framework/Core/include/Framework/ServiceRegistry.h b/Framework/Core/include/Framework/ServiceRegistry.h
index d6516e31be62d..44b75896331c6 100644
--- a/Framework/Core/include/Framework/ServiceRegistry.h
+++ b/Framework/Core/include/Framework/ServiceRegistry.h
@@ -177,7 +177,18 @@ struct ServiceRegistry {
 
   constexpr InstanceId instanceFromTypeSalt(ServiceTypeHash type, Salt salt) const
   {
-    return InstanceId{type.hash ^ valueFromSalt(salt)};
+    // Fold the whole salt down into the low bits. The slot is the low bits of
+    // this (see indexFromInstance) while streamId sits at bit 16 of
+    // valueFromSalt, so using that directly gives every stream the same slot
+    // for a given service: they pile into one probe window, and once it is
+    // MAX_DISTANCE deep the next registration is refused -- reported against
+    // whichever service happened to lose, not the one which filled it.
+    //
+    // Widening the table does not help on its own: the mask stays below bit 16
+    // until MAX_SERVICES passes 65536.
+    uint32_t mixed = static_cast<uint32_t>(static_cast<uint16_t>(salt.streamId)) * 0x9E3779B9u ^
+                     static_cast<uint32_t>(static_cast<uint16_t>(salt.dataProcessorId));
+    return InstanceId{type.hash ^ mixed};
   }
 
   constexpr Index indexFromInstance(InstanceId id) const
diff --git a/Framework/Core/test/test_Services.cxx b/Framework/Core/test/test_Services.cxx
index abac9eca5e9b0..2c746c99cd865 100644
--- a/Framework/Core/test/test_Services.cxx
+++ b/Framework/Core/test/test_Services.cxx
@@ -17,6 +17,7 @@
 #include <catch_amalgamated.hpp>
 #include <fairmq/ProgOptions.h>
 #include <memory>
+#include <vector>
 
 TEST_CASE("TestServiceRegistry")
 {
@@ -213,6 +214,50 @@ TEST_CASE("TestStreamServices")
   REQUIRE_THROWS_AS(registry.get({TypeIdHelpers::uniqueId<DummyService>()}, salt_1_1, ServiceKind::Stream), RuntimeErrorRef);
 }
 
+TEST_CASE("TestStreamServicesDoNotShareASlot")
+{
+  using namespace o2::framework;
+  ServiceRegistry registry;
+
+  ServiceSpec spec{.name = "dummy-service",
+                   .uniqueId = CommonServices::simpleServiceId<DummyService>(),
+                   .init = CommonServices::simpleServiceInit<DummyService, DummyService>(),
+                   .configure = CommonServices::noConfiguration(),
+                   .kind = ServiceKind::Stream};
+
+  DeviceState state;
+  fair::mq::ProgOptions options;
+  registry.declareService(spec, state, options, ServiceRegistry::globalDeviceSalt());
+
+  // One instance of the same service per stream, and more streams than a probe
+  // window is deep. The slot comes from the low bits of the type hash combined
+  // with the salt, so if the salt's streamId does not reach those bits every one
+  // of these lands on the same slot, and the first which does not fit in the
+  // window is refused outright.
+  constexpr short STREAMS = 32;
+  std::vector<DummyService> services(STREAMS);
+  for (short i = 0; i < STREAMS; ++i) {
+    services[i].threadId = i + 1;
+  }
+
+  for (short i = 0; i < STREAMS; ++i) {
+    // Refused registration is what a shared slot looks like from here: the
+    // window fills and the next one has nowhere to go.
+    REQUIRE_NOTHROW(registry.registerService({TypeIdHelpers::uniqueId<DummyService>()}, &services[i], ServiceKind::Stream,
+                                             ServiceRegistry::Salt{static_cast<short>(i + 1), 0}, "dummy-service",
+                                             ServiceRegistry::SpecIndex{0}));
+  }
+
+  // Every stream must get its own instance back, not a neighbour's.
+  for (short i = 0; i < STREAMS; ++i) {
+    auto* found = reinterpret_cast<DummyService*>(
+      registry.get({TypeIdHelpers::uniqueId<DummyService>()},
+                   ServiceRegistry::Salt{static_cast<short>(i + 1), 0}, ServiceKind::Stream));
+    REQUIRE(found != nullptr);
+    CHECK(found->threadId == i + 1);
+  }
+}
+
 TEST_CASE("TestServiceRegistryCtor")
 {
   using namespace o2::framework;