Skip to content

Commit 2ca1b77

Browse files
matthiasrichtermatthiasrichter
committed
Adding DPL parser tool for raw page sequences provided by DPL input.
The DPLRawPageSequencer can be used to find sequences of consecutive raw pages with a similar property, e.g. the FEE ID. The actual check and callback to handle a sequence can be provided by lambda functions. The raw pages within one buffer/message are expected to have the full length except the last page which can be smaller. The fixed spacing allows fast scanning by binary search. Corresponding unit test is emulating variable-length sequences of raw pages. fixup! Adding DPL parser tool for raw page sequences provided by DPL input.
1 parent 56111e3 commit 2ca1b77

3 files changed

Lines changed: 282 additions & 0 deletions

File tree

‎Framework/Utils/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ foreach(t
7676
RootTreeWriter
7777
RawParser
7878
DPLRawParser
79+
DPLRawPageSequencer
7980
)
8081
o2_add_test(${t}
8182
SOURCES test/test_${t}.cxx

‎Framework/Utils/include/DPLUtils/DPLRawPageSequencer.h‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
#ifndef FRAMEWORK_UTILS_DPLRAWPAGESEQUENCER_H
12+
#define FRAMEWORK_UTILS_DPLRAWPAGESEQUENCER_H
13+
14+
/// @file DPLRawPageSequencer.h
15+
/// @author Matthias Richter
16+
/// @since 2021-07-09
17+
/// @brief A parser and sequencer utility for raw pages within DPL input
18+
19+
#include "DPLUtils/RawParser.h"
20+
#include "Framework/DataRef.h"
21+
#include "Framework/DataRefUtils.h"
22+
#include "Framework/Logger.h"
23+
#include "Framework/InputRecordWalker.h"
24+
#include "Headers/DataHeader.h"
25+
#include <utility> // std::declval
26+
27+
namespace o2::framework
28+
{
29+
class InputRecord;
30+
31+
/// @class DPLRawPageSequencer
32+
/// @brief This utility handles transparently the DPL inputs and triggers
33+
/// a customizable action on sequences of consecutive raw pages following
34+
/// similar RDH features, e.g. the same FEE ID.
35+
///
36+
/// A DPL processor will receive raw pages accumulated on three levels:
37+
/// 1) the DPL processor has one or more input route(s)
38+
/// 2) multiple parts per input route (split payloads or multiple input
39+
/// specs matching the same route spec
40+
/// 3) variable number of raw pages in one payload
41+
///
42+
/// The DPLRawPageSequencer loops transparently over all inputs matching
43+
/// the optional filter, and partitions input buffers into sequences of
44+
/// raw pages matching the provided predicate by binary search.
45+
///
46+
/// Note: binary search requires that all raw pages must have a fixed
47+
/// length, only the last page can be shorter.
48+
///
49+
/// Usage:
50+
/// auto isSameRdh = [](const char* left, const char* right) -> bool {
51+
/// // implement the condition here
52+
/// return left == right;
53+
/// };
54+
/// std::vector<std::pair<const char*, size_t>> pages;
55+
/// auto insertPages = [&pages](const char* ptr, size_t n) -> void {
56+
/// // as an example, the sequences are simply stored in a vector
57+
/// pages.emplace_back(ptr, n);
58+
/// };
59+
/// DPLRawPageSequencer(inputs)(isSameRdh, insertPages);
60+
///
61+
/// TODO:
62+
/// - support configurable page length
63+
class DPLRawPageSequencer
64+
{
65+
public:
66+
using rawparser_type = RawParser<8192>;
67+
using buffer_type = typename rawparser_type::buffer_type;
68+
69+
DPLRawPageSequencer() = delete;
70+
DPLRawPageSequencer(InputRecord& inputs, std::vector<InputSpec> filterSpecs = {}) : mInput(inputs, filterSpecs) {}
71+
72+
template <typename Predicate, typename Inserter>
73+
void operator()(Predicate&& pred, Inserter&& inserter)
74+
{
75+
return binary(std::forward<Predicate>(pred), std::forward<Inserter>(inserter));
76+
}
77+
78+
template <typename Predicate, typename Inserter>
79+
void binary(Predicate pred, Inserter inserter)
80+
{
81+
for (auto const& ref : mInput) {
82+
auto size = DataRefUtils::getPayloadSize(ref);
83+
auto const pageSize = rawparser_type::max_size;
84+
auto nPages = size / pageSize + (size % pageSize ? 1 : 0);
85+
if (nPages == 0) {
86+
continue;
87+
}
88+
// FIXME: automatic type from inserter/predicate?
89+
const char* iterator = ref.payload;
90+
91+
auto check = [&pred, &pageSize, payload = ref.payload](size_t left, size_t right) -> bool {
92+
return pred(payload + left * pageSize, payload + right * pageSize);
93+
};
94+
auto insert = [&inserter, &pageSize, payload = ref.payload](size_t pos, size_t n) -> void {
95+
inserter(payload + pos * pageSize, n);
96+
};
97+
// binary search the next different page based on the check predicate
98+
auto search = [&check](size_t first, size_t n) -> size_t {
99+
auto count = n;
100+
auto pos = first;
101+
while (count > 0) {
102+
auto step = count / 2;
103+
if (check(first, pos + step)) {
104+
// still the same
105+
pos += step;
106+
count = n - (pos - first);
107+
} else {
108+
if (step == 1) {
109+
pos += step;
110+
break;
111+
}
112+
count = step;
113+
}
114+
}
115+
return pos;
116+
};
117+
118+
size_t p = 0;
119+
do {
120+
// insert the full block if the last RDH matches the position
121+
if (check(p, nPages - 1)) {
122+
insert(p, nPages - p);
123+
break;
124+
}
125+
auto q = search(p, nPages - p);
126+
insert(p, q - p);
127+
p = q;
128+
} while (p < nPages);
129+
// if payloads are consecutive in memory we could apply this algorithm even over
130+
// O2 message boundaries
131+
}
132+
}
133+
134+
template <typename Predicate, typename Inserter>
135+
void forward(Predicate check, Inserter inserter)
136+
{
137+
for (auto const& ref : mInput) {
138+
auto size = DataRefUtils::getPayloadSize(ref);
139+
o2::framework::RawParser parser(ref.payload, size);
140+
const char* ptr = nullptr;
141+
int count = 0;
142+
for (auto it = parser.begin(); it != parser.end(); it++) {
143+
const char* current = reinterpret_cast<const char*>(it.raw());
144+
if (ptr == nullptr) {
145+
ptr = current;
146+
} else if (check(ptr, current) == false) {
147+
if (count) {
148+
inserter(ptr, count);
149+
}
150+
count = 0;
151+
ptr = current;
152+
}
153+
count++;
154+
}
155+
if (count) {
156+
inserter(ptr, count);
157+
}
158+
}
159+
}
160+
161+
private:
162+
InputRecordWalker mInput;
163+
};
164+
165+
} // namespace o2::framework
166+
167+
#endif //FRAMEWORK_UTILS_DPLRAWPAGESEQUENCER_H

‎Framework/Utils/test/test_DPLRawPageSequencer.cxx‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
12+
/// @file test_DPLRawPageSequencer.h
13+
/// @author Matthias Richter
14+
/// @since 2021-07-09
15+
/// @brief Unit test for the DPL raw page sequencer utility
16+
17+
#define BOOST_TEST_MODULE Test Framework Utils DPLRawPageSequencer
18+
#define BOOST_TEST_MAIN
19+
#define BOOST_TEST_DYN_LINK
20+
#include <boost/test/unit_test.hpp>
21+
#include "DPLUtils/DPLRawPageSequencer.h"
22+
#include "RawPageTestData.h"
23+
#include "Framework/InputRecord.h"
24+
#include "Headers/DataHeader.h"
25+
#include <vector>
26+
#include <memory>
27+
#include <iostream>
28+
#include <random>
29+
30+
using namespace o2::framework;
31+
using DataHeader = o2::header::DataHeader;
32+
auto const PAGESIZE = test::PAGESIZE;
33+
34+
BOOST_AUTO_TEST_CASE(test_DPLRawPageSequencer)
35+
{
36+
const int nPages = 64;
37+
const int nParts = 16;
38+
std::vector<InputSpec> inputspecs = {
39+
InputSpec{"tpc", "TPC", "RAWDATA", 0, Lifetime::Timeframe}};
40+
41+
std::vector<DataHeader> dataheaders;
42+
dataheaders.emplace_back("RAWDATA", "TPC", 0, nPages * PAGESIZE, 0, nParts);
43+
44+
std::random_device rd;
45+
std::uniform_int_distribution<> lengthDist(1, nPages);
46+
auto randlength = [&rd, &lengthDist]() {
47+
return lengthDist(rd);
48+
};
49+
50+
int rdhCount = 0;
51+
// whenever a new id is created, it is done from the current counter
52+
// position, so we also have the possibility to calculate the length
53+
std::vector<uint16_t> feeids;
54+
auto nextlength = randlength();
55+
auto createFEEID = [&rdhCount, &feeids, &nPages, &randlength, &nextlength]() {
56+
if (rdhCount % nPages == 0 || rdhCount - feeids.back() > nextlength) {
57+
feeids.emplace_back(rdhCount);
58+
nextlength = randlength();
59+
}
60+
return feeids.back();
61+
};
62+
auto amendRdh = [&rdhCount, createFEEID](test::RAWDataHeader& rdh) {
63+
rdh.feeId = createFEEID();
64+
rdhCount++;
65+
};
66+
67+
auto dataset = test::createData(inputspecs, dataheaders, amendRdh);
68+
InputRecord& inputs = dataset.record;
69+
BOOST_REQUIRE(dataset.messages.size() > 0);
70+
BOOST_REQUIRE(dataset.messages[0].at(0) != nullptr);
71+
BOOST_REQUIRE(inputs.size() > 0);
72+
BOOST_CHECK((*inputs.begin()).header == dataset.messages[0].at(0)->data());
73+
BOOST_REQUIRE(rdhCount == nPages * nParts);
74+
DPLRawPageSequencer parser(inputs);
75+
76+
auto isSameRdh = [](const char* left, const char* right) -> bool {
77+
if (left == right) {
78+
return true;
79+
}
80+
if (left == nullptr || right == nullptr) {
81+
return true;
82+
}
83+
84+
return reinterpret_cast<test::RAWDataHeader const*>(left)->feeId == reinterpret_cast<test::RAWDataHeader const*>(right)->feeId;
85+
};
86+
std::vector<std::pair<const char*, size_t>> pages;
87+
auto insertPages = [&pages](const char* ptr, size_t n) -> void {
88+
pages.emplace_back(ptr, n);
89+
};
90+
parser(isSameRdh, insertPages);
91+
92+
// a second parsing step based on forward search
93+
std::vector<std::pair<const char*, size_t>> pagesByForwardSearch;
94+
auto insertForwardPages = [&pagesByForwardSearch](const char* ptr, size_t n) -> void {
95+
pagesByForwardSearch.emplace_back(ptr, n);
96+
};
97+
DPLRawPageSequencer(inputs).forward(isSameRdh, insertForwardPages);
98+
99+
LOG(INFO) << "called RDH amend: " << rdhCount;
100+
LOG(INFO) << "created " << feeids.size() << " id(s), got " << pages.size() << " page(s)";
101+
BOOST_REQUIRE(pages.size() == feeids.size());
102+
BOOST_REQUIRE(pages.size() == pagesByForwardSearch.size());
103+
104+
feeids.emplace_back(rdhCount);
105+
auto lastId = feeids.front();
106+
for (auto i = 0; i < pages.size(); i++) {
107+
auto length = feeids[i + 1] - feeids[i];
108+
BOOST_CHECK_MESSAGE(pages[i].second == length, "sequence " << i << " at " << feeids[i] << " length " << length << ": got " << pages[i].second);
109+
BOOST_CHECK_MESSAGE(pages[i].first == pagesByForwardSearch[i].first && pages[i].second == pagesByForwardSearch[i].second,
110+
"mismatch with forward search at sequence " << i
111+
<< " [" << (void*)pages[i].first << "," << (void*)pagesByForwardSearch[i].first << "]"
112+
<< " [" << pages[i].second << "," << pagesByForwardSearch[i].second << "]");
113+
}
114+
}

0 commit comments

Comments
 (0)