test_DataRelayer.cxx

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
 
#include <catch_amalgamated.hpp>
#include "Framework/DeviceState.h"
#include "Headers/DataHeader.h"
#include "Headers/Stack.h"
#include "MemoryResources/MemoryResources.h"
#include "Framework/CompletionPolicyHelpers.h"
#include "Framework/DataRelayer.h"
#include "Framework/DataModelViews.h"
#include "Framework/DataProcessingStats.h"
#include "Framework/DataProcessingStates.h"
#include "Framework/DriverConfig.h"
#include "Framework/TimingHelpers.h"
#include "../src/DataRelayerHelpers.h"
#include "Framework/DataProcessingHeader.h"
#include "Framework/ServiceRegistryHelpers.h"
#include "Framework/WorkflowSpec.h"
#include <Monitoring/Monitoring.h>
#include <fairmq/TransportFactory.h>
#include <fairmq/Channel.h>
#include "Framework/FairMQDeviceProxy.h"
#include "Framework/ExpirationHandler.h"
#include "Framework/LifetimeHelpers.h"
#include <array>
#include <cstring>
#include <new>
#include <cstdlib>
#include <atomic>
#include <vector>
#include <uv.h>
 
using Monitoring = o2::monitoring::Monitoring;
using namespace o2::framework;
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<bool> gCountAllocations{false};
std::atomic<size_t> 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;
  ServiceRegistryRef ref{registry};
  Monitoring monitoring;
  const DriverConfig driverConfig{
    .batch = false,
  };
  DataProcessingStates states(
    TimingHelpers::defaultRealtimeBaseConfigurator(0, uv_default_loop()),
    TimingHelpers::defaultCPUTimeConfigurator(uv_default_loop()));
  DataProcessingStats stats(
    TimingHelpers::defaultRealtimeBaseConfigurator(0, uv_default_loop()),
    TimingHelpers::defaultCPUTimeConfigurator(uv_default_loop()), {});
  int quickUpdateInterval = 1;
  using MetricSpec = DataProcessingStats::MetricSpec;
  std::vector<MetricSpec> specs{
    MetricSpec{.name = "malformed_inputs", .metricId = static_cast<short>(ProcessingStatsId::MALFORMED_INPUTS), .minPublishInterval = quickUpdateInterval},
    MetricSpec{.name = "dropped_computations", .metricId = static_cast<short>(ProcessingStatsId::DROPPED_COMPUTATIONS), .minPublishInterval = quickUpdateInterval},
    MetricSpec{.name = "dropped_incoming_messages", .metricId = static_cast<short>(ProcessingStatsId::DROPPED_INCOMING_MESSAGES), .minPublishInterval = quickUpdateInterval},
    MetricSpec{.name = "relayed_messages", .metricId = static_cast<short>(ProcessingStatsId::RELAYED_MESSAGES), .minPublishInterval = quickUpdateInterval}};
 
  for (auto& spec : specs) {
    stats.registerMetric(spec);
  }
 
  DeviceState state;
  ref.registerService(ServiceRegistryHelpers::handleForService<Monitoring>(&monitoring));
  ref.registerService(ServiceRegistryHelpers::handleForService<DataProcessingStats>(&stats));
  ref.registerService(ServiceRegistryHelpers::handleForService<DataProcessingStates>(&states));
  ref.registerService(ServiceRegistryHelpers::handleForService<DriverConfig const>(&driverConfig));
  ref.registerService(ServiceRegistryHelpers::handleForService<DeviceState>(&state));
  // A simple test where an input is provided
  // and the subsequent InputRecord is immediately requested.
  SECTION("TestNoWait")
  {
    InputSpec spec{"clusters", "TPC", "CLUSTERS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec, 0, "Fake", 0}};
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh;
    dh.dataDescription = "CLUSTERS";
    dh.dataOrigin = "TPC";
    dh.subSpecification = 0;
    dh.splitPayloadIndex = 0;
    dh.splitPayloadParts = 1;
 
    DataProcessingHeader dph{0, 1};
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    std::array<fair::mq::MessagePtr, 2> messages;
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
    messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, dph});
    messages[1] = transport->CreateMessage(1000);
    fair::mq::MessagePtr& header = messages[0];
    fair::mq::MessagePtr& payload = messages[1];
    DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].slot.index == 0);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    REQUIRE(header.get() == nullptr);
    REQUIRE(payload.get() == nullptr);
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    // one MessageSet with one PartRef with header and payload
    REQUIRE((result | count_inputs{}) == 1);
    REQUIRE((result[0] | count_parts{}) == 1);
  }
 
  //
  SECTION("TestNoWaitMatcher")
  {
    Monitoring metrics;
    auto specs = o2::framework::select("clusters:TPC/CLUSTERS");
 
    std::vector<InputRoute> inputs = {
      InputRoute{specs[0], 0, "Fake", 0}};
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh;
    dh.dataDescription = "CLUSTERS";
    dh.dataOrigin = "TPC";
    dh.subSpecification = 0;
    dh.splitPayloadIndex = 0;
    dh.splitPayloadParts = 1;
 
    DataProcessingHeader dph{0, 1};
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    std::array<fair::mq::MessagePtr, 2> messages;
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
    messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, dph});
    messages[1] = transport->CreateMessage(1000);
    fair::mq::MessagePtr& header = messages[0];
    fair::mq::MessagePtr& payload = messages[1];
    DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].slot.index == 0);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    REQUIRE(header.get() == nullptr);
    REQUIRE(payload.get() == nullptr);
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    // one MessageSet with one PartRef with header and payload
    REQUIRE((result | count_inputs{}) == 1);
    REQUIRE((result[0] | count_parts{}) == 1);
  }
 
  // This test a more complicated set of inputs, and verifies that data is
  // correctly relayed before being processed.
  SECTION("TestRelay")
  {
    Monitoring metrics;
    InputSpec spec1{
      "clusters",
      "TPC",
      "CLUSTERS",
    };
    InputSpec spec2{
      "clusters_its",
      "ITS",
      "CLUSTERS",
    };
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
      InputRoute{spec2, 1, "Fake2", 0}};
 
    std::vector<ForwardRoute> forwards;
 
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAll();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
 
    auto createMessage = [&transport, &channelAlloc, &relayer](DataHeader& dh, size_t time) {
      std::array<fair::mq::MessagePtr, 2> messages;
      messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{time, 1}});
      messages[1] = transport->CreateMessage(1000);
      fair::mq::MessagePtr& header = messages[0];
      fair::mq::MessagePtr& payload = messages[1];
      DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
      relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
      REQUIRE(header.get() == nullptr);
      REQUIRE(payload.get() == nullptr);
    };
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh1;
    dh1.dataDescription = "CLUSTERS";
    dh1.dataOrigin = "TPC";
    dh1.subSpecification = 0;
    dh1.splitPayloadIndex = 0;
    dh1.splitPayloadParts = 1;
 
    // Let's create the second O2 Message:
    DataHeader dh2;
    dh2.dataDescription = "CLUSTERS";
    dh2.dataOrigin = "ITS";
    dh2.subSpecification = 0;
    dh2.splitPayloadIndex = 0;
    dh2.splitPayloadParts = 1;
 
    createMessage(dh1, 0);
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 0);
 
    createMessage(dh2, 0);
    ready.clear();
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].slot.index == 0);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
 
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    // two MessageSets, each with one PartRef
    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
  // correctly relayed before being processed.
  SECTION("TestRelayBug")
  {
    Monitoring metrics;
    InputSpec spec1{
      "clusters",
      "TPC",
      "CLUSTERS",
    };
    InputSpec spec2{
      "clusters_its",
      "ITS",
      "CLUSTERS",
    };
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
      InputRoute{spec2, 1, "Fake2", 0}};
 
    std::vector<ForwardRoute> forwards;
 
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAll();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(3);
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
 
    auto createMessage = [&transport, &channelAlloc, &relayer](DataHeader& dh, size_t time) {
      std::array<fair::mq::MessagePtr, 2> messages;
      messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{time, 1}});
      messages[1] = transport->CreateMessage(1000);
      fair::mq::MessagePtr& header = messages[0];
      fair::mq::MessagePtr& payload = messages[1];
      DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
      relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
      REQUIRE(header.get() == nullptr);
      REQUIRE(payload.get() == nullptr);
    };
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh1;
    dh1.dataDescription = "CLUSTERS";
    dh1.dataOrigin = "TPC";
    dh1.subSpecification = 0;
    dh1.splitPayloadIndex = 0;
    dh1.splitPayloadParts = 1;
 
    // Let's create the second O2 Message:
    DataHeader dh2;
    dh2.dataDescription = "CLUSTERS";
    dh2.dataOrigin = "ITS";
    dh2.subSpecification = 0;
    dh2.splitPayloadIndex = 0;
    dh2.splitPayloadParts = 1;
 
    // Let's create the second O2 Message:
    DataHeader dh3;
    dh3.dataDescription = "CLUSTERS";
    dh3.dataOrigin = "FOO";
    dh3.subSpecification = 0;
    dh3.splitPayloadIndex = 0;
    dh3.splitPayloadParts = 1;
 
    createMessage(dh1, 0);
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 0);
    createMessage(dh1, 1);
    ready.clear();
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 0);
    createMessage(dh2, 0);
    ready.clear();
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].slot.index == 0);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    createMessage(dh2, 1);
    ready.clear();
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].slot.index == 1);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
  }
 
  // This tests a simple cache pruning, where a single input is shifted out of
  // the cache.
  SECTION("TestCache")
  {
    Monitoring metrics;
    InputSpec spec{"clusters", "TPC", "CLUSTERS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec, 0, "Fake", 0}};
    std::vector<ForwardRoute> forwards;
 
    auto policy = CompletionPolicyHelpers::consumeWhenAll();
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    // Only two messages to fill the cache.
    relayer.setPipelineLength(2);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh;
    dh.dataDescription = "CLUSTERS";
    dh.dataOrigin = "TPC";
    dh.subSpecification = 0;
    dh.splitPayloadIndex = 0;
    dh.splitPayloadParts = 1;
 
    DataProcessingHeader dph{0, 1};
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
    auto createMessage = [&transport, &channelAlloc, &relayer, &dh](auto const& h) {
      std::array<fair::mq::MessagePtr, 2> messages;
      messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, h});
      messages[1] = transport->CreateMessage(1000);
      fair::mq::MessagePtr& header = messages[0];
      fair::mq::MessagePtr& payload = messages[1];
      DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
      auto res = relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
      REQUIRE((res.type != DataRelayer::RelayChoice::Type::WillRelay || header.get() == nullptr));
      REQUIRE((res.type != DataRelayer::RelayChoice::Type::WillRelay || payload.get() == nullptr));
      REQUIRE((res.type != DataRelayer::RelayChoice::Type::Backpressured || header.get() != nullptr));
      REQUIRE((res.type != DataRelayer::RelayChoice::Type::Backpressured || payload.get() != nullptr));
    };
 
    // This fills the cache, and then empties it.
    createMessage(DataProcessingHeader{0, 1});
    createMessage(DataProcessingHeader{1, 1});
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 2);
    REQUIRE(ready[0].slot.index == 1);
    REQUIRE(ready[1].slot.index == 0);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    REQUIRE(ready[1].op == CompletionPolicy::CompletionOp::Consume);
    for (size_t i = 0; i < ready.size(); ++i) {
      auto result = relayer.consumeAllInputsForTimeslice(ready[i].slot);
    }
 
    // This fills the cache and makes 2 obsolete.
    createMessage(DataProcessingHeader{2, 1});
    createMessage(DataProcessingHeader{3, 1});
    createMessage(DataProcessingHeader{4, 1});
    ready.clear();
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 2);
 
    auto result1 = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    auto result2 = relayer.consumeAllInputsForTimeslice(ready[1].slot);
    // One for the header, one for the payload
    REQUIRE((result1 | count_inputs{}) == 1);
    REQUIRE((result2 | count_inputs{}) == 1);
  }
 
  // This the any policy. Even when there are two inputs, given the any policy
  // it will run immediately.
  SECTION("TestPolicies")
  {
    Monitoring metrics;
    InputSpec spec1{"clusters", "TPC", "CLUSTERS"};
    InputSpec spec2{"tracks", "TPC", "TRACKS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
      InputRoute{spec2, 1, "Fake2", 0},
    };
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::processWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    // Only two messages to fill the cache.
    relayer.setPipelineLength(2);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh1;
    dh1.dataDescription = "CLUSTERS";
    dh1.dataOrigin = "TPC";
    dh1.subSpecification = 0;
    dh1.splitPayloadIndex = 0;
    dh1.splitPayloadParts = 1;
 
    DataHeader dh2;
    dh2.dataDescription = "TRACKS";
    dh2.dataOrigin = "TPC";
    dh2.subSpecification = 0;
    dh2.splitPayloadIndex = 0;
    dh2.splitPayloadParts = 1;
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
    auto createMessage = [&transport, &channelAlloc, &relayer](auto const& dh, auto const& h) {
      std::array<fair::mq::MessagePtr, 2> messages;
      messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, h});
      messages[1] = transport->CreateMessage(1000);
      fair::mq::MessagePtr& header = messages[0];
      DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
      return relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
    };
 
    // This fills the cache, and then empties it.
    createMessage(dh1, DataProcessingHeader{0, 1});
    std::vector<RecordAction> ready1;
    relayer.getReadyToProcess(ready1);
    REQUIRE(ready1.size() == 1);
    REQUIRE(ready1[0].slot.index == 0);
    REQUIRE(ready1[0].op == CompletionPolicy::CompletionOp::Process);
 
    createMessage(dh1, DataProcessingHeader{1, 1});
    std::vector<RecordAction> ready2;
    relayer.getReadyToProcess(ready2);
    REQUIRE(ready2.size() == 1);
    REQUIRE(ready2[0].slot.index == 1);
    REQUIRE(ready2[0].op == CompletionPolicy::CompletionOp::Process);
 
    createMessage(dh2, DataProcessingHeader{1, 1});
    std::vector<RecordAction> ready3;
    relayer.getReadyToProcess(ready3);
    REQUIRE(ready3.size() == 1);
    REQUIRE(ready3[0].slot.index == 1);
    REQUIRE(ready3[0].op == CompletionPolicy::CompletionOp::Consume);
  }
 
  SECTION("TestClear")
  {
    Monitoring metrics;
    InputSpec spec1{"clusters", "TPC", "CLUSTERS"};
    InputSpec spec2{"tracks", "TPC", "TRACKS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
      InputRoute{spec2, 1, "Fake2", 0},
    };
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::processWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    // Only two messages to fill the cache.
    relayer.setPipelineLength(3);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh1;
    dh1.dataDescription = "CLUSTERS";
    dh1.dataOrigin = "TPC";
    dh1.subSpecification = 0;
    dh1.splitPayloadIndex = 0;
    dh1.splitPayloadParts = 1;
 
    DataHeader dh2;
    dh2.dataDescription = "TRACKS";
    dh2.dataOrigin = "TPC";
    dh2.subSpecification = 0;
    dh2.splitPayloadIndex = 0;
    dh2.splitPayloadParts = 1;
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
    auto createMessage = [&transport, &channelAlloc, &relayer](auto const& dh, auto const& h) {
      std::array<fair::mq::MessagePtr, 2> messages;
      messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh, h});
      messages[1] = transport->CreateMessage(1000);
      fair::mq::MessagePtr& header = messages[0];
      DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
      return relayer.relay(header->GetData(), messages.data(), fakeInfo, messages.size());
    };
 
    // This fills the cache, and then empties it.
    createMessage(dh1, DataProcessingHeader{0, 1});
    createMessage(dh1, DataProcessingHeader{1, 1});
    createMessage(dh2, DataProcessingHeader{1, 1});
    relayer.clear();
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 0);
  }
 
  SECTION("TestTooMany")
  {
    Monitoring metrics;
    InputSpec spec1{"clusters", "TPC", "CLUSTERS"};
    InputSpec spec2{"tracks", "TPC", "TRACKS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
      InputRoute{spec2, 1, "Fake2", 0},
    };
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::processWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    // Only two messages to fill the cache.
    relayer.setPipelineLength(1);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh1;
    dh1.dataDescription = "CLUSTERS";
    dh1.dataOrigin = "TPC";
    dh1.subSpecification = 0;
    dh1.splitPayloadIndex = 0;
    dh1.splitPayloadParts = 1;
 
    DataHeader dh2;
    dh2.dataDescription = "TRACKS";
    dh2.dataOrigin = "TPC";
    dh2.subSpecification = 0;
    dh2.splitPayloadIndex = 0;
    dh2.splitPayloadParts = 1;
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
 
    std::array<fair::mq::MessagePtr, 4> messages;
    messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh1, DataProcessingHeader{0, 1}});
    messages[1] = transport->CreateMessage(1000);
    fair::mq::MessagePtr& header = messages[0];
    fair::mq::MessagePtr& payload = messages[1];
    DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    relayer.relay(header->GetData(), &messages[0], fakeInfo, 2);
    REQUIRE(header.get() == nullptr);
    REQUIRE(payload.get() == nullptr);
    // This fills the cache, and then waits.
    messages[2] = o2::pmr::getMessage(Stack{channelAlloc, dh1, DataProcessingHeader{1, 1}});
    messages[3] = transport->CreateMessage(1000);
    fair::mq::MessagePtr& header2 = messages[2];
    fair::mq::MessagePtr& payload2 = messages[3];
    DataRelayer::InputInfo fakeInfo2{2, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    auto action = relayer.relay(header2->GetData(), &messages[2], fakeInfo2, 2);
    REQUIRE(action.type == DataRelayer::RelayChoice::Type::Backpressured);
    REQUIRE(header2.get() != nullptr);
    REQUIRE(payload2.get() != nullptr);
  }
 
  SECTION("SplitParts")
  {
    Monitoring metrics;
    InputSpec spec1{"clusters", "TPC", "CLUSTERS"};
    InputSpec spec2{"its", "ITS", "CLUSTERS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
      InputRoute{spec2, 0, "Fake2", 0},
    };
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::processWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    // Only two messages to fill the cache.
    relayer.setPipelineLength(1);
 
    // Let's create a dummy O2 Message with two headers in the stack:
    // - DataHeader matching the one provided in the input
    DataHeader dh1;
    dh1.dataDescription = "CLUSTERS";
    dh1.dataOrigin = "TPC";
    dh1.subSpecification = 0;
    dh1.splitPayloadIndex = 0;
    dh1.splitPayloadParts = 1;
 
    DataHeader dh2;
    dh2.dataDescription = "TRACKS";
    dh2.dataOrigin = "TPC";
    dh2.subSpecification = 0;
    dh2.splitPayloadIndex = 0;
    dh2.splitPayloadParts = 1;
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
 
    std::array<fair::mq::MessagePtr, 6> messages;
    messages[0] = o2::pmr::getMessage(Stack{channelAlloc, dh1, DataProcessingHeader{0, 1}});
    messages[1] = transport->CreateMessage(1000);
    fair::mq::MessagePtr& header = messages[0];
    fair::mq::MessagePtr& payload = messages[1];
    DataRelayer::InputInfo fakeInfo{0, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    relayer.relay(header->GetData(), &messages[0], fakeInfo, 2);
    REQUIRE(header.get() == nullptr);
    REQUIRE(payload.get() == nullptr);
    // This fills the cache, and then waits.
    messages[2] = o2::pmr::getMessage(Stack{channelAlloc, dh1, DataProcessingHeader{1, 1}});
    messages[3] = transport->CreateMessage(1000);
    fair::mq::MessagePtr& header2 = messages[2];
    fair::mq::MessagePtr& payload2 = messages[3];
    DataRelayer::InputInfo fakeInfo2{2, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    auto action = relayer.relay(header2->GetData(), &messages[2], fakeInfo, 2);
    REQUIRE(action.type == DataRelayer::RelayChoice::Type::Backpressured);
    CHECK(action.timeslice.value == 1);
    REQUIRE(header2.get() != nullptr);
    REQUIRE(payload2.get() != nullptr);
    // This fills the cache, and then waits.
    messages[4] = o2::pmr::getMessage(Stack{channelAlloc, dh1, DataProcessingHeader{1, 1}});
    messages[5] = transport->CreateMessage(1000);
    DataRelayer::InputInfo fakeInfo3{4, 2, DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    relayer.relay(header2->GetData(), &messages[4], fakeInfo3, 2);
    REQUIRE(action.type == DataRelayer::RelayChoice::Type::Backpressured);
    CHECK(action.timeslice.value == 1);
    REQUIRE(header2.get() != nullptr);
    REQUIRE(payload2.get() != nullptr);
  }
 
  SECTION("SplitPayloadPairs")
  {
    Monitoring metrics;
    InputSpec spec1{"clusters", "TPC", "CLUSTERS"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
    };
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    DataHeader dh{"CLUSTERS", "TPC", 0};
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
    size_t timeslice = 0;
 
    const int nSplitParts = 100;
    std::vector<std::unique_ptr<fair::mq::Message>> splitParts;
    splitParts.reserve(2 * nSplitParts);
 
    for (size_t i = 0; i < nSplitParts; ++i) {
      dh.splitPayloadIndex = i;
      dh.splitPayloadParts = nSplitParts;
 
      fair::mq::MessagePtr header = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{timeslice, 1}});
      fair::mq::MessagePtr payload = transport->CreateMessage(100);
 
      splitParts.emplace_back(std::move(header));
      splitParts.emplace_back(std::move(payload));
    }
    REQUIRE(splitParts.size() == 2 * nSplitParts);
 
    DataRelayer::InputInfo fakeInfo{0, splitParts.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
    relayer.relay(splitParts[0]->GetData(), splitParts.data(), fakeInfo, splitParts.size());
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    auto messageSet = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    // we have one input route and thus one message set containing pairs for all
    // payloads
    REQUIRE((messageSet | count_inputs{}) == 1);
    REQUIRE((messageSet[0] | count_parts{}) == nSplitParts);
    REQUIRE((messageSet[0] | get_num_payloads{0}) == 1);
  }
 
  SECTION("SplitPayloadSequence")
  {
    Monitoring metrics;
    InputSpec spec1{"clusters", "TST", "COUNTER"};
 
    std::vector<InputRoute> inputs = {
      InputRoute{spec1, 0, "Fake1", 0},
    };
 
    std::vector<ForwardRoute> forwards;
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    size_t timeslice = 0;
 
    std::vector<size_t> sequenceSize;
    size_t nTotalPayloads = 0;
 
    auto createSequence = [&nTotalPayloads, &timeslice, &sequenceSize, &transport, &relayer](size_t nPayloads) -> void {
      auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
      std::vector<std::unique_ptr<fair::mq::Message>> messages;
      messages.reserve(nPayloads + 1);
      DataHeader dh{"COUNTER", "TST", 0};
 
      // one header with index set to the number of split parts indicates sequence
      // of payloads without additional headers
      dh.splitPayloadIndex = nPayloads;
      dh.splitPayloadParts = nPayloads;
      fair::mq::MessagePtr header = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{timeslice, 1}});
      messages.emplace_back(std::move(header));
 
      for (size_t i = 0; i < nPayloads; ++i) {
        messages.emplace_back(transport->CreateMessage(100));
        *(reinterpret_cast<size_t*>(messages.back()->GetData())) = nTotalPayloads;
        ++nTotalPayloads;
      }
      REQUIRE(messages.size() == nPayloads + 1);
      DataRelayer::InputInfo fakeInfo{0, messages.size(), DataRelayer::InputType::Data, {ChannelIndex::INVALID}};
      relayer.relay(messages[0]->GetData(), messages.data(), fakeInfo, messages.size(), nPayloads);
      sequenceSize.emplace_back(nPayloads);
    };
    createSequence(100);
    createSequence(1);
    createSequence(42);
 
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
    auto messageSet = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    // we have one input route
    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;
    for (size_t seqid = 0; seqid < sequenceSize.size(); ++seqid) {
      REQUIRE((messageSet[0] | get_num_payloads{seqid}) == sequenceSize[seqid]);
      for (size_t pi = 0; pi < (messageSet[0] | get_num_payloads{seqid}); ++pi) {
        REQUIRE((messageSet[0] | get_payload{seqid, pi}));
        auto const* data = (messageSet[0] | get_payload{seqid, pi})->GetData();
        REQUIRE(*(reinterpret_cast<size_t const*>(data)) == counter);
        ++counter;
      }
    }
  }
 
  SECTION("ProcessDanglingInputs")
  {
    InputSpec spec{"condition", "TST", "COND"};
    std::vector<InputRoute> inputs = {
      InputRoute{spec, 0, "from_source_to_self", 0}};
 
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    // Bind a fake input channel so FairMQDeviceProxy::getInputChannelIndex works
    FairMQDeviceProxy proxy;
    std::vector<fair::mq::Channel> 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<FairMQDeviceProxy>(&proxy));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
 
    DataHeader dh{"COND", "TST", 0};
    dh.splitPayloadParts = 1;
    dh.splitPayloadIndex = 0;
    DataProcessingHeader dph{0, 1};
 
    ExpirationHandler handler;
    handler.name = "test-condition";
    handler.routeIndex = RouteIndex{0};
    handler.lifetime = Lifetime::Condition;
 
    // Creator: claim an empty slot and assign timeslice 0 to it
    handler.creator = [](ServiceRegistryRef services, ChannelIndex channelIndex) -> TimesliceSlot {
      auto& index = services.get<TimesliceIndex>();
      for (size_t si = 0; si < index.size(); si++) {
        TimesliceSlot slot{si};
        if (!index.isValid(slot)) {
          index.associate(TimesliceId{0}, slot);
          (void)index.setOldestPossibleInput({1}, channelIndex);
          return slot;
        }
      }
      return TimesliceSlot{TimesliceSlot::INVALID};
    };
 
    // Checker: always trigger expiration
    handler.checker = LifetimeHelpers::expireAlways();
 
    // Handler: materialise a dummy header+payload into the PartRef
    handler.handler = [&transport, &channelAlloc, &dh, &dph](ServiceRegistryRef, PartRef& ref, data_matcher::VariableContext&) {
      ref.header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
      ref.payload = transport->CreateMessage(4);
    };
 
    std::vector<ExpirationHandler> handlers{handler};
    auto activity = relayer.processDanglingInputs(handlers, {registry}, true);
 
    REQUIRE(activity.newSlots == 1);
    REQUIRE(activity.expiredSlots == 1);
 
    // The materialised data should now be ready to consume
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
 
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    REQUIRE((result | count_inputs{}) == 1);
    REQUIRE((result[0] | count_parts{}) == 1);
  }
 
  SECTION("ProcessDanglingInputsSkipsWhenDataPresent")
  {
    // processDanglingInputs must not overwrite a slot that already has data.
    // This is guarded by the (part.messages | get_header{0}) != nullptr check.
    InputSpec spec{"condition", "TST", "COND"};
    std::vector<InputRoute> inputs = {
      InputRoute{spec, 0, "from_source_to_self", 0}};
 
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    FairMQDeviceProxy proxy;
    std::vector<fair::mq::Channel> 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<FairMQDeviceProxy>(&proxy));
 
    auto policy = CompletionPolicyHelpers::consumeWhenAny();
    DataRelayer relayer(policy, inputs, index, {registry}, -1);
    relayer.setPipelineLength(4);
 
    auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
 
    DataHeader dh{"COND", "TST", 0};
    dh.splitPayloadParts = 1;
    dh.splitPayloadIndex = 0;
    DataProcessingHeader dph{0, 1};
 
    // Build an expiration handler that always tries to expire
    ExpirationHandler handler;
    handler.name = "test-condition";
    handler.routeIndex = RouteIndex{0};
    handler.lifetime = Lifetime::Condition;
    handler.creator = [](ServiceRegistryRef services, ChannelIndex channelIndex) -> TimesliceSlot {
      auto& index = services.get<TimesliceIndex>();
      for (size_t si = 0; si < index.size(); si++) {
        TimesliceSlot slot{si};
        if (!index.isValid(slot)) {
          index.associate(TimesliceId{0}, slot);
          (void)index.setOldestPossibleInput({1}, channelIndex);
          return slot;
        }
      }
      return TimesliceSlot{TimesliceSlot::INVALID};
    };
    handler.checker = LifetimeHelpers::expireAlways();
    int handlerCallCount = 0;
    handler.handler = [&transport, &channelAlloc, &dh, &dph, &handlerCallCount](ServiceRegistryRef, PartRef& ref, data_matcher::VariableContext&) {
      ref.header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
      ref.payload = transport->CreateMessage(4);
      handlerCallCount++;
    };
    std::vector<ExpirationHandler> handlers{handler};
 
    // First call: slot is empty, so the handler fires and materialises data
    auto activity1 = relayer.processDanglingInputs(handlers, {registry}, true);
    REQUIRE(activity1.expiredSlots == 1);
    REQUIRE(handlerCallCount == 1);
 
    // Second call: slot already has data — the handler must NOT fire again
    auto activity2 = relayer.processDanglingInputs(handlers, {registry}, false);
    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<InputRoute> inputs = {
      InputRoute{spec0, 0, "Fake0", 0},
      InputRoute{spec1, 1, "Fake1", 0},
      InputRoute{spec2, 2, "Fake2", 0},
    };
 
    std::vector<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&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<DataHeader, 3> 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<uint32_t>(input + 1) + static_cast<uint32_t>(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<fair::mq::MessagePtr, 2> 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<std::pair<size_t, size_t>, 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<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
 
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    REQUIRE((result | count_inputs{}) == 3);
 
    std::array<size_t, 3> 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<InputRoute> 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<InputChannelInfo> infos{1};
    TimesliceIndex index{1, infos};
    ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&index));
 
    FairMQDeviceProxy proxy;
    std::vector<fair::mq::Channel> 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<FairMQDeviceProxy>(&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<uint32_t>(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<fair::mq::MessagePtr, 2> 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<ExpirationHandler> handlers{handler};
    auto activity = relayer.processDanglingInputs(handlers, {registry}, true);
    REQUIRE(activity.expiredSlots == 1);
 
    std::vector<RecordAction> ready;
    relayer.getReadyToProcess(ready);
    REQUIRE(ready.size() == 1);
    REQUIRE(ready[0].op == CompletionPolicy::CompletionOp::Consume);
 
    auto result = relayer.consumeAllInputsForTimeslice(ready[0].slot);
    REQUIRE((result | count_inputs{}) == 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));
 
      // 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<InputSpec> specs;
        std::vector<InputRoute> inputs;
        std::vector<DataHeader> prototypes;
        std::array<char const*, kInputs> 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<InputChannelInfo> infos{1};
        TimesliceIndex index{1, infos};
        ref.registerService(ServiceRegistryHelpers::handleForService<TimesliceIndex>(&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<std::array<fair::mq::MessagePtr, 2>> 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<std::array<fair::mq::MessagePtr, 2>>& 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<RecordAction> 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);
      }
    }
  }
}