Extended test of payload sequences

Author: matthiasrichter

Framework/Core/test/test_DataRelayer.cxx

@@ -666,7 +666,7 @@ BOOST_AUTO_TEST_CASE(SplitPayloadPairs)
 BOOST_AUTO_TEST_CASE(SplitPayloadSequence)
 {
   Monitoring metrics;
-  InputSpec spec1{"clusters", "TPC", "CLUSTERS"};
+  InputSpec spec1{"clusters", "TST", "COUNTER"};
 
   std::vector<InputRoute> inputs = {
     InputRoute{spec1, 0, "Fake1", 0},
@@ -679,37 +679,55 @@ BOOST_AUTO_TEST_CASE(SplitPayloadSequence)
   DataRelayer relayer(policy, inputs, metrics, index);
   relayer.setPipelineLength(4);
 
-  DataHeader dh{"CLUSTERS", "TPC", 0};
-
   auto transport = FairMQTransportFactory::CreateTransportFactory("zeromq");
-  auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
   size_t timeslice = 0;
 
-  const int nSplitParts = 100;
-  std::vector<std::unique_ptr<FairMQMessage>> splitParts;
-  splitParts.reserve(nSplitParts + 1);
+  std::vector<size_t> sequenceSize;
+  size_t nTotalPayloads = 0;
 
-  // one header with index set to the number of split parts indicates sequence
-  // of payloads without additional headers
-  dh.splitPayloadIndex = nSplitParts;
-  dh.splitPayloadParts = nSplitParts;
-  FairMQMessagePtr header = o2::pmr::getMessage(Stack{channelAlloc, dh, DataProcessingHeader{timeslice, 1}});
-  splitParts.emplace_back(std::move(header));
+  auto createSequence = [&nTotalPayloads, &timeslice, &sequenceSize, &transport, &relayer](size_t nPayloads) -> void {
+    auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
+    std::vector<std::unique_ptr<FairMQMessage>> messages;
+    messages.reserve(nPayloads + 1);
+    DataHeader dh{"COUNTER", "TST", 0};
 
-  for (size_t i = 0; i < nSplitParts; ++i) {
-    splitParts.emplace_back(transport->CreateMessage(100));
-    *(reinterpret_cast<size_t*>(splitParts.back()->GetData())) = i;
-  }
-  BOOST_REQUIRE_EQUAL(splitParts.size(), nSplitParts + 1);
+    // one header with index set to the number of split parts indicates sequence
+    // of payloads without additional headers
+    dh.splitPayloadIndex = nPayloads;
+    dh.splitPayloadParts = nPayloads;
+    FairMQMessagePtr 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;
+    }
+    BOOST_CHECK_EQUAL(messages.size(), nPayloads + 1);
+    relayer.relay(messages[0]->GetData(), messages.data(), messages.size(), nPayloads);
+    sequenceSize.emplace_back(nPayloads);
+  };
+  createSequence(100);
+  createSequence(1);
+  createSequence(42);
 
-  relayer.relay(splitParts[0]->GetData(), splitParts.data(), splitParts.size(), nSplitParts);
   std::vector<RecordAction> ready;
   relayer.getReadyToProcess(ready);
   BOOST_REQUIRE_EQUAL(ready.size(), 1);
   BOOST_REQUIRE_EQUAL(ready[0].op, CompletionPolicy::CompletionOp::Consume);
   auto messageSet = relayer.getInputsForTimeslice(ready[0].slot);
-  // we have one input route and thus one message set containing one sequence of messages
+  // we have one input route
   BOOST_REQUIRE_EQUAL(messageSet.size(), 1);
-  BOOST_REQUIRE_EQUAL(messageSet[0].size(), 1);
-  BOOST_CHECK_EQUAL(messageSet[0].getNumberOfPayloads(0), 100);
+  // one message set containing number of added sequences of messages
+  BOOST_REQUIRE_EQUAL(messageSet[0].size(), sequenceSize.size());
+  size_t counter = 0;
+  for (auto seqid = 0; seqid < sequenceSize.size(); ++seqid) {
+    BOOST_CHECK_EQUAL(messageSet[0].getNumberOfPayloads(seqid), sequenceSize[seqid]);
+    for (auto pi = 0; pi < messageSet[0].getNumberOfPayloads(seqid); ++pi) {
+      BOOST_REQUIRE(messageSet[0].payload(seqid, pi));
+      auto const* data = messageSet[0].payload(seqid, pi)->GetData();
+      BOOST_CHECK_EQUAL(*(reinterpret_cast<size_t const*>(data)), counter);
+      ++counter;
+    }
+  }
 }