From 68780d76be3286b7eeab5777c3ade2ae73afc71b Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:23:22 +0200 Subject: [PATCH 1/4] DPL: pin per-input message identity in the relayer tests The existing sections assert part counts and pointer nullness, never payload contents, so any change to how a slot's messages are stored can shuffle them between inputs without a single test noticing. Stamp every payload and check it comes back on the right input, in the order it was relayed. Two arrangements that a shared per-slot buffer makes interesting: arrivals interleaved across three inputs, so a cell is no longer the last one written when its second part shows up; and an expiring input materialised into a slot the other inputs already occupy, which leaves the cells out of input order. Contents are also re-checked after the slot has been refilled, which pins that consuming really does hand the messages over. --- Framework/Core/test/test_DataRelayer.cxx | 205 +++++++++++++++++++++++ 1 file changed, 205 insertions(+) diff --git a/Framework/Core/test/test_DataRelayer.cxx b/Framework/Core/test/test_DataRelayer.cxx index 271b7829a9525..7d05454922e18 100644 --- a/Framework/Core/test/test_DataRelayer.cxx +++ b/Framework/Core/test/test_DataRelayer.cxx @@ -32,6 +32,7 @@ #include "Framework/ExpirationHandler.h" #include "Framework/LifetimeHelpers.h" #include +#include #include #include @@ -968,4 +969,208 @@ TEST_CASE("DataRelayer") REQUIRE(activity2.expiredSlots == 0); REQUIRE(handlerCallCount == 1); // handler was not called a second time } + + // Once the DataRelayer keeps a slot's messages in one shared buffer, every + // input's parts live next to each other, so a slip in the offset bookkeeping + // corrupts a *different* input's cell while leaving all the part counts + // intact. Counting parts therefore cannot catch it: stamp each payload and + // check identity. The arrival order below is interleaved on purpose -- after + // step 2 input 0 is no longer the last cell, so step 3 has to relocate it, + // and likewise input 1 at step 5. + SECTION("InterleavedPartsKeepIdentity") + { + InputSpec spec0{"clusters", "TPC", "CLUSTERS"}; + InputSpec spec1{"its", "ITS", "CLUSTERS"}; + InputSpec spec2{"tracks", "TPC", "TRACKS"}; + + std::vector inputs = { + InputRoute{spec0, 0, "Fake0", 0}, + InputRoute{spec1, 1, "Fake1", 0}, + InputRoute{spec2, 2, "Fake2", 0}, + }; + + std::vector infos{1}; + TimesliceIndex index{1, infos}; + ref.registerService(ServiceRegistryHelpers::handleForService(&index)); + + auto policy = CompletionPolicyHelpers::consumeWhenAll(); + DataRelayer relayer(policy, inputs, index, {registry}, -1); + relayer.setPipelineLength(1); + + auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq"); + auto channelAlloc = o2::pmr::getTransportAllocator(transport.get()); + + std::array prototypes; + prototypes[0].dataOrigin = "TPC"; + prototypes[0].dataDescription = "CLUSTERS"; + prototypes[1].dataOrigin = "ITS"; + prototypes[1].dataDescription = "CLUSTERS"; + prototypes[2].dataOrigin = "TPC"; + prototypes[2].dataDescription = "TRACKS"; + + auto stampOf = [](size_t input, size_t part) -> uint32_t { + return 1000u * static_cast(input + 1) + static_cast(part); + }; + + auto relayOne = [&](size_t input, size_t part, size_t timeslice) { + DataHeader dh = prototypes[input]; + dh.subSpecification = 0; + dh.splitPayloadIndex = 0; + dh.splitPayloadParts = 1; + dh.payloadSize = sizeof(uint32_t); + + std::array msgs; + msgs[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{timeslice, 1}}); + msgs[1] = transport->CreateMessage(sizeof(uint32_t)); + uint32_t const stamp = stampOf(input, part); + memcpy(msgs[1]->GetData(), &stamp, sizeof(stamp)); + DataRelayer::InputInfo info{0, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}}; + relayer.relay(msgs[0]->GetData(), msgs.data(), info, 2); + REQUIRE(msgs[0].get() == nullptr); + REQUIRE(msgs[1].get() == nullptr); + }; + + std::array, 5> const arrivals = {{{0, 0}, {1, 0}, {0, 1}, {2, 0}, {1, 1}}}; + for (auto const& [input, part] : arrivals) { + relayOne(input, part, 0); + } + + std::vector ready; + relayer.getReadyToProcess(ready); + REQUIRE(ready.size() == 1); + REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume); + + auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); + REQUIRE(result.size() == 3); + + std::array const expectedParts = {2, 2, 1}; + auto checkContents = [&]() { + for (size_t i = 0; i < 3; ++i) { + REQUIRE((result[i] | count_parts{}) == expectedParts[i]); + for (size_t p = 0; p < expectedParts[i]; ++p) { + auto& header = result[i] | get_header{p}; + auto& payload = result[i] | get_payload{p, 0}; + REQUIRE(header.get() != nullptr); + REQUIRE(payload.get() != nullptr); + uint32_t seen = 0; + memcpy(&seen, payload->GetData(), sizeof(seen)); + REQUIRE(seen == stampOf(i, p)); + } + } + }; + checkContents(); + + // The consumed messages belong to the caller now. Refilling the very same + // slot must not disturb them, whether the relayer handed over vectors or an + // arena it has since reused. + relayOne(0, 0, 1); + checkContents(); + } + + // An expiring input is materialised straight into the slot, so with one + // shared buffer per slot it lands *after* whatever the other inputs already + // hold -- the cells are then no longer in input order. Check that the data + // which was already there survives the expiry untouched. + SECTION("ExpiryDoesNotDisturbNeighbours") + { + InputSpec dataSpec0{"clusters", "TPC", "CLUSTERS"}; + InputSpec condSpec{"condition", "TST", "COND"}; + InputSpec dataSpec2{"tracks", "TPC", "TRACKS"}; + + std::vector inputs = { + InputRoute{dataSpec0, 0, "from_source_to_self", 0}, + InputRoute{condSpec, 1, "from_source_to_self", 0}, + InputRoute{dataSpec2, 2, "from_source_to_self", 0}, + }; + + std::vector infos{1}; + TimesliceIndex index{1, infos}; + ref.registerService(ServiceRegistryHelpers::handleForService(&index)); + + FairMQDeviceProxy proxy; + std::vector channels{fair::mq::Channel("from_source_to_self")}; + auto findChannel = [&channels](std::string const& name) -> fair::mq::Channel& { + for (auto& ch : channels) { + if (ch.GetName() == name) { + return ch; + } + } + throw std::runtime_error("Channel not found: " + name); + }; + proxy.bind({}, inputs, {}, findChannel, [] { return false; }); + ref.registerService(ServiceRegistryHelpers::handleForService(&proxy)); + + auto policy = CompletionPolicyHelpers::consumeWhenAll(); + DataRelayer relayer(policy, inputs, index, {registry}, -1); + relayer.setPipelineLength(1); + + auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq"); + auto channelAlloc = o2::pmr::getTransportAllocator(transport.get()); + + auto stampOf = [](size_t input) -> uint32_t { return 7000u + static_cast(input); }; + + auto relayData = [&](size_t input, char const* origin, char const* description) { + DataHeader dh; + dh.dataOrigin.runtimeInit(origin); + dh.dataDescription.runtimeInit(description); + dh.subSpecification = 0; + dh.splitPayloadIndex = 0; + dh.splitPayloadParts = 1; + dh.payloadSize = sizeof(uint32_t); + std::array msgs; + msgs[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{0, 1}}); + msgs[1] = transport->CreateMessage(sizeof(uint32_t)); + uint32_t const stamp = stampOf(input); + memcpy(msgs[1]->GetData(), &stamp, sizeof(stamp)); + DataRelayer::InputInfo info{0, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}}; + relayer.relay(msgs[0]->GetData(), msgs.data(), info, 2); + REQUIRE(msgs[0].get() == nullptr); + }; + + // The two data inputs arrive first, so the slot is already occupied when + // the condition expires into it. + relayData(0, "TPC", "CLUSTERS"); + relayData(2, "TPC", "TRACKS"); + + DataHeader condDh{"COND", "TST", 0}; + condDh.splitPayloadParts = 1; + condDh.splitPayloadIndex = 0; + DataProcessingHeader condDph{0, 1}; + + ExpirationHandler handler; + handler.name = "test-condition"; + handler.routeIndex = RouteIndex{1}; + handler.lifetime = Lifetime::Condition; + // Deliberately *not* a fresh slot: return the one the data is already in, + // which is what puts the materialised cell out of input order. + handler.creator = [](ServiceRegistryRef, ChannelIndex) -> TimesliceSlot { + return TimesliceSlot{0}; + }; + handler.checker = LifetimeHelpers::expireAlways(); + handler.handler = [&transport, &channelAlloc, &condDh, &condDph](ServiceRegistryRef, PartRef& part, data_matcher::VariableContext&) { + part.header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, condDh, condDph}); + part.payload = transport->CreateMessage(4); + }; + + std::vector handlers{handler}; + auto activity = relayer.processDanglingInputs(handlers, {registry}, true); + REQUIRE(activity.expiredSlots == 1); + + std::vector ready; + relayer.getReadyToProcess(ready); + REQUIRE(ready.size() == 1); + REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume); + + auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); + REQUIRE(result.size() == 3); + REQUIRE((result[1] | count_parts{}) == 1); + for (size_t i : {0u, 2u}) { + REQUIRE((result[i] | count_parts{}) == 1); + auto& payload = result[i] | get_payload{0, 0}; + REQUIRE(payload.get() != nullptr); + uint32_t seen = 0; + memcpy(&seen, payload->GetData(), sizeof(seen)); + REQUIRE(seen == stampOf(i)); + } + } } From 3b58c28caeb49db181e40b2aadfc212d2a9f9885 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:23:50 +0200 Subject: [PATCH 2/4] DPL: assert an allocation budget for a relay/consume cycle A timing benchmark cannot tell a storage-layout regression from a busy machine. Count allocations instead: with eight inputs, relaying every input plus the consume costs 18 allocations, and that number must not grow when the way a slot holds its messages changes. The messages are built before the counter is armed, so what is measured is the relayer rather than fair::mq. The global operator new replacement only counts while a test arms it, so the rest of the binary is unaffected. --- Framework/Core/test/test_DataRelayer.cxx | 123 +++++++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/Framework/Core/test/test_DataRelayer.cxx b/Framework/Core/test/test_DataRelayer.cxx index 7d05454922e18..bcaa4b531c83a 100644 --- a/Framework/Core/test/test_DataRelayer.cxx +++ b/Framework/Core/test/test_DataRelayer.cxx @@ -33,6 +33,9 @@ #include "Framework/LifetimeHelpers.h" #include #include +#include +#include +#include #include #include @@ -42,6 +45,41 @@ using DataHeader = o2::header::DataHeader; using Stack = o2::header::Stack; using RecordAction = o2::framework::DataRelayer::RecordAction; +// Replacing the global allocation functions lets a test assert an allocation +// *budget* rather than a wall-clock time: the DataRelayer's storage layout is +// supposed to cost a bounded number of allocations per timeslice, and that is a +// deterministic property, unlike a benchmark on a shared machine. Counting is +// off unless a test arms it, so nothing else in the binary is affected. +namespace +{ +std::atomic gCountAllocations{false}; +std::atomic gAllocations{0}; + +struct AllocationCounter { + AllocationCounter() + { + gAllocations.store(0, std::memory_order_relaxed); + gCountAllocations.store(true, std::memory_order_relaxed); + } + ~AllocationCounter() { gCountAllocations.store(false, std::memory_order_relaxed); } + static size_t count() { return gAllocations.load(std::memory_order_relaxed); } +}; +} // namespace + +void* operator new(std::size_t size) +{ + if (gCountAllocations.load(std::memory_order_relaxed)) { + gAllocations.fetch_add(1, std::memory_order_relaxed); + } + if (void* p = std::malloc(size ? size : 1)) { + return p; + } + throw std::bad_alloc(); +} + +void operator delete(void* p) noexcept { std::free(p); } +void operator delete(void* p, std::size_t) noexcept { std::free(p); } + TEST_CASE("DataRelayer") { ServiceRegistry registry; @@ -1171,6 +1209,91 @@ TEST_CASE("DataRelayer") uint32_t seen = 0; memcpy(&seen, payload->GetData(), sizeof(seen)); REQUIRE(seen == stampOf(i)); + + // A storage-layout change is supposed to cost a bounded number of allocations + // per timeslice regardless of how many inputs there are. Assert that budget + // directly: it is deterministic, unlike timing it on a machine that is also + // compiling. The bound below is what upstream costs; if a change makes the + // relayer allocate more per timeslice, this fails without anyone having to + // read a benchmark table. + SECTION("RelayAllocationBudget") + { + constexpr size_t kInputs = 8; + std::vector specs; + std::vector inputs; + std::vector prototypes; + std::array const descriptions = { + "CLUSTERS", "TRACKS", "DIGITS", "VERTICES", "ERRORS", "CALIB", "RAWDATA", "MCLABELS"}; + for (size_t i = 0; i < kInputs; ++i) { + o2::header::DataDescription desc; + desc.runtimeInit(descriptions[i]); + specs.emplace_back(InputSpec{"in", "TST", desc}); + } + for (size_t i = 0; i < kInputs; ++i) { + inputs.emplace_back(InputRoute{specs[i], i, "Fake", 0}); + DataHeader dh; + dh.dataOrigin = "TST"; + dh.dataDescription.runtimeInit(descriptions[i]); + dh.subSpecification = 0; + dh.splitPayloadIndex = 0; + dh.splitPayloadParts = 1; + dh.payloadSize = 8; + prototypes.push_back(dh); + } + + std::vector infos{1}; + TimesliceIndex index{1, infos}; + ref.registerService(ServiceRegistryHelpers::handleForService(&index)); + + auto policy = CompletionPolicyHelpers::consumeWhenAll(); + DataRelayer relayer(policy, inputs, index, {registry}, -1); + relayer.setPipelineLength(1); + + auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq"); + auto channelAlloc = o2::pmr::getTransportAllocator(transport.get()); + + // Build the messages first: creating them allocates, and that cost has + // nothing to do with how the relayer stores them. Only the relay + consume + // is measured. + auto makeMessages = [&](size_t timeslice) { + std::vector> msgs(kInputs); + for (size_t i = 0; i < kInputs; ++i) { + msgs[i][0] = o2::pmr::getMessage(Stack{channelAlloc, prototypes[i], DataProcessingHeader{timeslice, 1}}); + msgs[i][1] = transport->CreateMessage(8); + } + return msgs; + }; + + auto cycle = [&](std::vector>& msgs) { + for (size_t i = 0; i < kInputs; ++i) { + DataRelayer::InputInfo info{0, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}}; + relayer.relay(msgs[i][0]->GetData(), msgs[i].data(), info, 2); + } + std::vector ready; + relayer.getReadyToProcess(ready); + REQUIRE(ready.size() == 1); + return relayer.consumeAllInputsForTimeslice(ready[0].slot); + }; + + // Warm up, so the measured cycle is the recurring cost rather than the + // first-time growth of every internal buffer. + for (size_t t = 0; t < 4; ++t) { + auto msgs = makeMessages(t); + auto warm = cycle(msgs); + } + + auto msgs = makeMessages(4); + size_t allocations = 0; + { + AllocationCounter counting; + auto result = cycle(msgs); + allocations = AllocationCounter::count(); + } + // With one vector per input this measures 18 for eight inputs. The exact + // figure matters less than the fact that it must not grow when the way a + // slot's messages are stored changes; tighten the bound if it drops. + REQUIRE(allocations <= 18); + } } } } From 03e80c5c3d9c6c0683a2c6306a5fc43afda15c9a Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:27:18 +0200 Subject: [PATCH 3/4] DPL: ask a record how many inputs it holds, rather than the container The relayer tests and benchmarks reach into the consumed record with .size() and .at(), which pins them to the record being a vector. Both operator[] and a count_inputs pipe work on anything the relayer might hand back -- a vector of per-input sets, or an arena holding them in one buffer -- so a change of storage leaves this code untouched instead of rewriting thirty call sites. --- .../Core/include/Framework/DataModelViews.h | 17 +++++++++++ Framework/Core/test/benchmark_DataRelayer.cxx | 14 ++++----- Framework/Core/test/test_DataRelayer.cxx | 30 +++++++++---------- 3 files changed, 39 insertions(+), 22 deletions(-) diff --git a/Framework/Core/include/Framework/DataModelViews.h b/Framework/Core/include/Framework/DataModelViews.h index dd8d65ea16459..e50d62d9eb6ec 100644 --- a/Framework/Core/include/Framework/DataModelViews.h +++ b/Framework/Core/include/Framework/DataModelViews.h @@ -49,6 +49,23 @@ struct count_payloads { } }; +// How many inputs a consumed record holds. A record is either a vector of +// per-input message sets or an arena keeping them in one buffer; both answer +// this, but they spell it differently, so ask through here and callers stay put +// when the storage underneath them changes. +struct count_inputs { + // ends the pipeline, returns the number of inputs + template + friend size_t operator|(R&& r, count_inputs self) + { + if constexpr (requires { r.numInputs(); }) { + return r.numInputs(); + } else { + return r.size(); + } + } +}; + struct count_parts { // ends the pipeline, returns the number of parts template diff --git a/Framework/Core/test/benchmark_DataRelayer.cxx b/Framework/Core/test/benchmark_DataRelayer.cxx index ca47b63193c1e..f7adfcb0e4b6e 100644 --- a/Framework/Core/test/benchmark_DataRelayer.cxx +++ b/Framework/Core/test/benchmark_DataRelayer.cxx @@ -139,8 +139,8 @@ static void BM_RelaySingleSlot(benchmark::State& state) assert(ready[0].slot.index == 0); assert(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); - assert(result.size() == 1); - assert((result.at(0) | count_parts{}) == 1); + assert((result | count_inputs{}) == 1); + assert((result[0] | count_parts{}) == 1); inflightMessages.assign(std::make_move_iterator(result[0].begin()), std::make_move_iterator(result[0].end())); } @@ -196,8 +196,8 @@ static void BM_RelayMultipleSlots(benchmark::State& state) assert(ready.size() == 1); assert(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); - assert(result.size() == 1); - assert((result.at(0) | count_parts{}) == 1); + assert((result | count_inputs{}) == 1); + assert((result[0] | count_parts{}) == 1); inflightMessages.assign(std::make_move_iterator(result[0].begin()), std::make_move_iterator(result[0].end())); } @@ -271,9 +271,9 @@ static void BM_RelayMultipleRoutes(benchmark::State& state) assert(ready.size() == 1); assert(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); - assert(result.size() == 2); - assert((result.at(0) | count_parts{}) == 1); - assert((result.at(1) | count_parts{}) == 1); + assert((result | count_inputs{}) == 2); + assert((result[0] | count_parts{}) == 1); + assert((result[1] | count_parts{}) == 1); inflightMessages.assign(std::make_move_iterator(result[0].begin()), std::make_move_iterator(result[0].end())); inflightMessages.insert(inflightMessages.end(), diff --git a/Framework/Core/test/test_DataRelayer.cxx b/Framework/Core/test/test_DataRelayer.cxx index bcaa4b531c83a..3a5181897892f 100644 --- a/Framework/Core/test/test_DataRelayer.cxx +++ b/Framework/Core/test/test_DataRelayer.cxx @@ -158,8 +158,8 @@ TEST_CASE("DataRelayer") REQUIRE(payload.get() == nullptr); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); // one MessageSet with one PartRef with header and payload - REQUIRE(result.size() == 1); - REQUIRE((result.at(0) | count_parts{}) == 1); + REQUIRE((result | count_inputs{}) == 1); + REQUIRE((result[0] | count_parts{}) == 1); } // @@ -208,8 +208,8 @@ TEST_CASE("DataRelayer") REQUIRE(payload.get() == nullptr); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); // one MessageSet with one PartRef with header and payload - REQUIRE(result.size() == 1); - REQUIRE((result.at(0) | count_parts{}) == 1); + REQUIRE((result | count_inputs{}) == 1); + REQUIRE((result[0] | count_parts{}) == 1); } // This test a more complicated set of inputs, and verifies that data is @@ -288,9 +288,9 @@ TEST_CASE("DataRelayer") auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); // two MessageSets, each with one PartRef - REQUIRE(result.size() == 2); - REQUIRE((result.at(0) | count_parts{}) == 1); - REQUIRE((result.at(1) | count_parts{}) == 1); + REQUIRE((result | count_inputs{}) == 2); + REQUIRE((result[0] | count_parts{}) == 1); + REQUIRE((result[1] | count_parts{}) == 1); } // This test a more complicated set of inputs, and verifies that data is @@ -458,8 +458,8 @@ TEST_CASE("DataRelayer") auto result1 = relayer.consumeAllInputsForTimeslice(ready[0].slot); auto result2 = relayer.consumeAllInputsForTimeslice(ready[1].slot); // One for the header, one for the payload - REQUIRE(result1.size() == 1); - REQUIRE(result2.size() == 1); + REQUIRE((result1 | count_inputs{}) == 1); + REQUIRE((result2 | count_inputs{}) == 1); } // This the any policy. Even when there are two inputs, given the any policy @@ -776,7 +776,7 @@ TEST_CASE("DataRelayer") auto messageSet = relayer.consumeAllInputsForTimeslice(ready[0].slot); // we have one input route and thus one message set containing pairs for all // payloads - REQUIRE(messageSet.size() == 1); + REQUIRE((messageSet | count_inputs{}) == 1); REQUIRE((messageSet[0] | count_parts{}) == nSplitParts); REQUIRE((messageSet[0] | get_num_payloads{0}) == 1); } @@ -838,7 +838,7 @@ TEST_CASE("DataRelayer") REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto messageSet = relayer.consumeAllInputsForTimeslice(ready[0].slot); // we have one input route - REQUIRE(messageSet.size() == 1); + REQUIRE((messageSet | count_inputs{}) == 1); // one message set containing number of added sequences of messages REQUIRE((messageSet[0] | count_parts{}) == sequenceSize.size()); size_t counter = 0; @@ -930,8 +930,8 @@ TEST_CASE("DataRelayer") REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); - REQUIRE(result.size() == 1); - REQUIRE((result.at(0) | count_parts{}) == 1); + REQUIRE((result | count_inputs{}) == 1); + REQUIRE((result[0] | count_parts{}) == 1); } SECTION("ProcessDanglingInputsSkipsWhenDataPresent") @@ -1079,7 +1079,7 @@ TEST_CASE("DataRelayer") REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); - REQUIRE(result.size() == 3); + REQUIRE((result | count_inputs{}) == 3); std::array const expectedParts = {2, 2, 1}; auto checkContents = [&]() { @@ -1200,7 +1200,7 @@ TEST_CASE("DataRelayer") REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume); auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); - REQUIRE(result.size() == 3); + REQUIRE((result | count_inputs{}) == 3); REQUIRE((result[1] | count_parts{}) == 1); for (size_t i : {0u, 2u}) { REQUIRE((result[i] | count_parts{}) == 1); From 026cb248f5dd8cf174069099e58992ff144b585c Mon Sep 17 00:00:00 2001 From: Giulio Eulisse <10544+ktf@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:34:19 +0200 Subject: [PATCH 4/4] DPL: benchmark a relay/consume cycle across many inputs Every existing benchmark here uses one or two inputs, which is exactly the regime where per-input storage costs nothing to speak of, so none of them can see a change to how a slot holds its messages. Sweep 1/8/32/128 instead. Note this is the only benchmark using consumeWhenAll, which looks the TimesliceIndex up in the service registry (CompletionPolicyHelpers). The others use consumeWhenAny and never do, which is why BenchmarkServices does not register it and why it has to be registered here -- without it the benchmark throws at every input count, including one. --- Framework/Core/test/benchmark_DataRelayer.cxx | 79 +++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/Framework/Core/test/benchmark_DataRelayer.cxx b/Framework/Core/test/benchmark_DataRelayer.cxx index f7adfcb0e4b6e..d2ba05d770759 100644 --- a/Framework/Core/test/benchmark_DataRelayer.cxx +++ b/Framework/Core/test/benchmark_DataRelayer.cxx @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -402,4 +403,82 @@ static void BM_RelayMultiplePayloads(benchmark::State& state) BENCHMARK(BM_RelayMultiplePayloads)->Arg(10)->Arg(100)->Arg(1000); +// Every benchmark above uses one or two inputs, which is exactly the regime +// where per-input storage costs nothing to speak of. Sweep the number of inputs +// so a change to how a slot holds its messages is visible where it matters. +// +// Note this is the only benchmark here using consumeWhenAll, which needs the +// TimesliceIndex from the registry (CompletionPolicyHelpers.cxx). The others use +// consumeWhenAny and never look it up, which is why BenchmarkServices does not +// register it and why it has to be registered here. +static void BM_RelayManyInputs(benchmark::State& state) +{ + BenchmarkServices services; + size_t const nInputs = state.range(0); + + std::vector specs; + std::vector inputs; + std::vector prototypes; + specs.reserve(nInputs); + for (size_t i = 0; i < nInputs; ++i) { + char description[16]; + snprintf(description, sizeof(description), "DATA%03zu", i); + o2::header::DataDescription desc; + desc.runtimeInit(description); + specs.emplace_back(InputSpec{"in", "TST", desc}); + DataHeader dh; + dh.dataOrigin = "TST"; + dh.dataDescription = desc; + dh.subSpecification = 0; + dh.splitPayloadIndex = 0; + dh.splitPayloadParts = 1; + dh.payloadSize = 100; + prototypes.push_back(dh); + } + for (size_t i = 0; i < nInputs; ++i) { + inputs.emplace_back(InputRoute{specs[i], i, "Fake", 0}); + } + + std::vector infos{1}; + TimesliceIndex index{1, infos}; + auto ref = services.ref(); + ref.registerService(ServiceRegistryHelpers::handleForService(&index)); + + auto policy = CompletionPolicyHelpers::consumeWhenAll(); + DataRelayer relayer(policy, inputs, index, ref, -1); + relayer.setPipelineLength(1); + + auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq"); + + // One message pair per input, recycled through the relayer every iteration. + std::vector inflight; + inflight.reserve(2 * nInputs); + for (size_t i = 0; i < nInputs; ++i) { + Stack stack{prototypes[i], DataProcessingHeader{0, 1}}; + fair::mq::MessagePtr header = transport->CreateMessage(stack.size()); + memcpy(header->GetData(), stack.data(), stack.size()); + inflight.emplace_back(std::move(header)); + inflight.emplace_back(transport->CreateMessage(prototypes[i].payloadSize)); + } + + for (auto _ : state) { + for (size_t i = 0; i < nInputs; ++i) { + DataRelayer::InputInfo info{0, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}}; + relayer.relay(inflight[2 * i]->GetData(), &inflight[2 * i], info, 2); + } + std::vector ready; + relayer.getReadyToProcess(ready); + assert(ready.size() == 1); + auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot); + inflight.clear(); + for (size_t i = 0; i < nInputs; ++i) { + for (auto& msg : result[i]) { + inflight.emplace_back(std::move(msg)); + } + } + } +} + +BENCHMARK(BM_RelayManyInputs)->Arg(1)->Arg(8)->Arg(32)->Arg(128); + BENCHMARK_MAIN();