Skip to content

Commit 909c5f9

Browse files
ktfktf
authored
DPL: propaedeutic to navigate a MessageSet without caching pairs (#15187)
All this should be fairly straight forward changes while still preserving the old API. If something fails at this level it means that even the counting of dataset changes with this included, which it should not be.
1 parent d17d862 commit 909c5f9

File tree

3 files changed

+115
-82
lines changed

3 files changed

+115
-82
lines changed

‎Framework/Core/include/Framework/DataModelViews.h‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -206,15 +206,10 @@ struct get_num_payloads {
206206

207207
struct MessageSet;
208208

209-
struct MessageStore {
210-
std::span<MessageSet> sets;
211-
size_t inputsPerSlot = 0;
212-
};
213-
214209
struct inputs_for_slot {
215210
TimesliceSlot slot;
216211
template <typename R>
217-
requires requires(R r) { std::ranges::random_access_range<decltype(r.sets)>; }
212+
requires requires(R r) { requires std::ranges::random_access_range<decltype(r.sets)>; }
218213
friend std::span<o2::framework::MessageSet> operator|(R&& r, inputs_for_slot self)
219214
{
220215
return std::span(r.sets[self.slot.index * r.inputsPerSlot]);

‎Framework/Core/include/Framework/MessageSet.h‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,13 @@
1212
#define FRAMEWORK_MESSAGESET_H
1313

1414
#include "Framework/PartRef.h"
15+
#include <fairmq/Message.h>
16+
#include "Framework/DataModelViews.h"
1517
#include <memory>
1618
#include <vector>
1719
#include <cassert>
1820

19-
namespace o2
20-
{
21-
namespace framework
21+
namespace o2::framework
2222
{
2323

2424
/// A set of inflight messages.
@@ -83,21 +83,21 @@ struct MessageSet {
8383
}
8484

8585
/// get number of in-flight O2 messages
86-
size_t size() const
86+
[[nodiscard]] size_t size() const
8787
{
88-
return messageMap.size();
88+
return messages | count_parts{};
8989
}
9090

9191
/// get number of header-payload pairs
92-
size_t getNumberOfPairs() const
92+
[[nodiscard]] size_t getNumberOfPairs() const
9393
{
94-
return pairMap.size();
94+
return messages | count_payloads{};
9595
}
9696

9797
/// get number of payloads for an in-flight message
98-
size_t getNumberOfPayloads(size_t mi) const
98+
[[nodiscard]] size_t getNumberOfPayloads(size_t mi) const
9999
{
100-
return messageMap[mi].size;
100+
return messages | get_num_payloads{mi};
101101
}
102102

103103
/// clear the set
@@ -179,6 +179,6 @@ struct MessageSet {
179179
}
180180
};
181181

182-
} // namespace framework
183-
} // namespace o2
182+
} // namespace o2::framework
183+
184184
#endif // FRAMEWORK_MESSAGESET_H

‎Framework/Core/test/test_MessageSet.cxx‎

Lines changed: 103 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -10,126 +10,164 @@
1010
// or submit itself to any jurisdiction.
1111

1212
#include <fairmq/Message.h>
13+
#include <fairmq/TransportFactory.h>
1314
#include "Framework/MessageSet.h"
15+
#include "Framework/DataProcessingHeader.h"
16+
#include "Headers/Stack.h"
17+
#include "MemoryResources/MemoryResources.h"
1418
#include <catch_amalgamated.hpp>
1519

1620
using namespace o2::framework;
1721

18-
TEST_CASE("MessageSet") {
22+
TEST_CASE("MessageSet")
23+
{
1924
o2::framework::MessageSet msgSet;
20-
std::vector<fair::mq::MessagePtr> ptrs;
21-
std::unique_ptr<fair::mq::Message> msg(nullptr);
25+
o2::header::DataHeader dh{};
26+
dh.splitPayloadParts = 0;
27+
dh.splitPayloadIndex = 0;
28+
o2::framework::DataProcessingHeader dph{0, 1};
29+
auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
30+
fair::mq::MessagePtr payload(transport->CreateMessage());
31+
auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
32+
fair::mq::MessagePtr header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
2233
std::unique_ptr<fair::mq::Message> msg2(nullptr);
23-
ptrs.emplace_back(std::move(msg));
34+
std::vector<fair::mq::MessagePtr> ptrs;
35+
ptrs.emplace_back(std::move(header));
2436
ptrs.emplace_back(std::move(msg2));
2537
msgSet.add([&ptrs](size_t i) -> fair::mq::MessagePtr& { return ptrs[i]; }, 2);
2638

2739
REQUIRE(msgSet.messages.size() == 2);
28-
REQUIRE(msgSet.messageMap.size() == 1);
29-
REQUIRE(msgSet.pairMap.size() == 1);
30-
REQUIRE(msgSet.messageMap[0].position == 0);
31-
REQUIRE(msgSet.messageMap[0].size == 1);
32-
33-
REQUIRE(msgSet.pairMap[0].partIndex == 0);
34-
REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
40+
REQUIRE((msgSet.messages | count_payloads{}) == 1);
41+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).headerIdx == 0);
42+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).payloadIdx == 1);
43+
REQUIRE((msgSet.messages | get_pair{0}).headerIdx == 0);
44+
REQUIRE((msgSet.messages | get_pair{0}).payloadIdx == 1);
45+
CHECK_THROWS((msgSet.messages | get_pair{1}));
3546
}
3647

37-
TEST_CASE("MessageSetWithFunction") {
48+
TEST_CASE("MessageSetWithFunction")
49+
{
3850
std::vector<fair::mq::MessagePtr> ptrs;
39-
std::unique_ptr<fair::mq::Message> msg(nullptr);
51+
o2::header::DataHeader dh{};
52+
dh.splitPayloadParts = 0;
53+
dh.splitPayloadIndex = 0;
54+
o2::framework::DataProcessingHeader dph{0, 1};
55+
auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
56+
fair::mq::MessagePtr payload(transport->CreateMessage());
57+
auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
58+
fair::mq::MessagePtr header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
4059
std::unique_ptr<fair::mq::Message> msg2(nullptr);
41-
ptrs.emplace_back(std::move(msg));
60+
ptrs.emplace_back(std::move(header));
4261
ptrs.emplace_back(std::move(msg2));
4362
o2::framework::MessageSet msgSet([&ptrs](size_t i) -> fair::mq::MessagePtr& { return ptrs[i]; }, 2);
4463

4564
REQUIRE(msgSet.messages.size() == 2);
46-
REQUIRE(msgSet.messageMap.size() == 1);
47-
REQUIRE(msgSet.pairMap.size() == 1);
48-
REQUIRE(msgSet.messageMap[0].position == 0);
49-
REQUIRE(msgSet.messageMap[0].size == 1);
50-
51-
REQUIRE(msgSet.pairMap[0].partIndex == 0);
52-
REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
65+
REQUIRE((msgSet.messages | count_payloads{}) == 1);
66+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).headerIdx == 0);
67+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).payloadIdx == 1);
68+
REQUIRE((msgSet.messages | get_pair{0}).headerIdx == 0);
69+
REQUIRE((msgSet.messages | get_pair{0}).payloadIdx == 1);
70+
CHECK_THROWS((msgSet.messages | get_pair{1}));
5371
}
5472

55-
TEST_CASE("MessageSetWithMultipart") {
73+
TEST_CASE("MessageSetWithMultipart")
74+
{
5675
std::vector<fair::mq::MessagePtr> ptrs;
57-
std::unique_ptr<fair::mq::Message> msg(nullptr);
76+
o2::header::DataHeader dh{};
77+
dh.splitPayloadParts = 2;
78+
dh.splitPayloadIndex = 2;
79+
o2::framework::DataProcessingHeader dph{0, 1};
80+
auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
81+
fair::mq::MessagePtr payload(transport->CreateMessage());
82+
auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
83+
fair::mq::MessagePtr header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
5884
std::unique_ptr<fair::mq::Message> msg2(nullptr);
5985
std::unique_ptr<fair::mq::Message> msg3(nullptr);
60-
ptrs.emplace_back(std::move(msg));
86+
ptrs.emplace_back(std::move(header));
6187
ptrs.emplace_back(std::move(msg2));
6288
ptrs.emplace_back(std::move(msg3));
6389
o2::framework::MessageSet msgSet([&ptrs](size_t i) -> fair::mq::MessagePtr& { return ptrs[i]; }, 3);
6490

6591
REQUIRE(msgSet.messages.size() == 3);
66-
REQUIRE(msgSet.messageMap.size() == 1);
67-
REQUIRE(msgSet.pairMap.size() == 2);
68-
REQUIRE(msgSet.messageMap[0].position == 0);
69-
REQUIRE(msgSet.messageMap[0].size == 2);
70-
71-
REQUIRE(msgSet.pairMap[0].partIndex == 0);
72-
REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
73-
REQUIRE(msgSet.pairMap[1].partIndex == 0);
74-
REQUIRE(msgSet.pairMap[1].payloadIndex == 1);
92+
REQUIRE((msgSet.messages | count_payloads{}) == 2);
93+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).headerIdx == 0);
94+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).payloadIdx == 1);
95+
REQUIRE((msgSet.messages | get_dataref_indices{0, 1}).headerIdx == 0);
96+
REQUIRE((msgSet.messages | get_dataref_indices{0, 1}).payloadIdx == 2);
97+
REQUIRE((msgSet.messages | get_pair{0}).headerIdx == 0);
98+
REQUIRE((msgSet.messages | get_pair{0}).payloadIdx == 1);
99+
REQUIRE((msgSet.messages | get_pair{1}).headerIdx == 0);
100+
REQUIRE((msgSet.messages | get_pair{1}).payloadIdx == 2);
101+
CHECK_THROWS((msgSet.messages | get_pair{2}));
75102
}
76103

77-
TEST_CASE("MessageSetAddPartRef") {
104+
TEST_CASE("MessageSetAddPartRef")
105+
{
78106
std::vector<fair::mq::MessagePtr> ptrs;
79107
std::unique_ptr<fair::mq::Message> msg(nullptr);
80108
std::unique_ptr<fair::mq::Message> msg2(nullptr);
81109
ptrs.emplace_back(std::move(msg));
82110
ptrs.emplace_back(std::move(msg2));
83-
PartRef ref {std::move(msg), std::move(msg2)};
111+
PartRef ref{std::move(msg), std::move(msg2)};
84112
o2::framework::MessageSet msgSet;
85113
msgSet.add(std::move(ref));
86114

87115
REQUIRE(msgSet.messages.size() == 2);
88-
REQUIRE(msgSet.messageMap.size() == 1);
89-
REQUIRE(msgSet.pairMap.size() == 1);
90-
REQUIRE(msgSet.messageMap[0].position == 0);
91-
REQUIRE(msgSet.messageMap[0].size == 1);
92-
93-
REQUIRE(msgSet.pairMap[0].partIndex == 0);
94-
REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
95116
}
96117

97118
TEST_CASE("MessageSetAddMultiple")
98119
{
99120
std::vector<fair::mq::MessagePtr> ptrs;
100-
std::unique_ptr<fair::mq::Message> msg(nullptr);
121+
o2::header::DataHeader dh1{};
122+
dh1.splitPayloadParts = 0;
123+
dh1.splitPayloadIndex = 0;
124+
o2::header::DataHeader dh2{};
125+
dh2.splitPayloadParts = 1;
126+
dh2.splitPayloadIndex = 0;
127+
o2::header::DataHeader dh3{};
128+
dh3.splitPayloadParts = 2;
129+
dh3.splitPayloadIndex = 2;
130+
o2::framework::DataProcessingHeader dph{0, 1};
131+
auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
132+
fair::mq::MessagePtr payload(transport->CreateMessage());
133+
auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
134+
fair::mq::MessagePtr header1 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh1, dph});
135+
fair::mq::MessagePtr header2 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh2, dph});
136+
fair::mq::MessagePtr header3 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh3, dph});
137+
101138
std::unique_ptr<fair::mq::Message> msg2(nullptr);
102-
ptrs.emplace_back(std::move(msg));
103-
ptrs.emplace_back(std::move(msg2));
104-
PartRef ref{std::move(msg), std::move(msg2)};
139+
std::unique_ptr<fair::mq::Message> msg3(nullptr);
140+
PartRef ref{std::move(header1), std::move(msg2)};
105141
o2::framework::MessageSet msgSet;
106142
msgSet.add(std::move(ref));
107-
PartRef ref2{std::move(msg), std::move(msg2)};
143+
PartRef ref2{std::move(header2), std::move(msg2)};
108144
msgSet.add(std::move(ref2));
109145
std::vector<fair::mq::MessagePtr> msgs;
110-
msgs.push_back(std::unique_ptr<fair::mq::Message>(nullptr));
146+
msgs.push_back(std::move(header3));
111147
msgs.push_back(std::unique_ptr<fair::mq::Message>(nullptr));
112148
msgs.push_back(std::unique_ptr<fair::mq::Message>(nullptr));
113149
msgSet.add([&msgs](size_t i) {
114150
return std::move(msgs[i]);
115-
}, 3);
151+
},
152+
3);
116153

117154
REQUIRE(msgSet.messages.size() == 7);
118-
REQUIRE(msgSet.messageMap.size() == 3);
119-
REQUIRE(msgSet.pairMap.size() == 4);
120-
REQUIRE(msgSet.messageMap[0].position == 0);
121-
REQUIRE(msgSet.messageMap[0].size == 1);
122-
REQUIRE(msgSet.messageMap[1].position == 2);
123-
REQUIRE(msgSet.messageMap[1].size == 1);
124-
REQUIRE(msgSet.messageMap[2].position == 4);
125-
REQUIRE(msgSet.messageMap[2].size == 2);
126155

127-
REQUIRE(msgSet.pairMap[0].partIndex == 0);
128-
REQUIRE(msgSet.pairMap[0].payloadIndex == 0);
129-
REQUIRE(msgSet.pairMap[1].partIndex == 1);
130-
REQUIRE(msgSet.pairMap[1].payloadIndex == 0);
131-
REQUIRE(msgSet.pairMap[2].partIndex == 2);
132-
REQUIRE(msgSet.pairMap[2].payloadIndex == 0);
133-
REQUIRE(msgSet.pairMap[3].partIndex == 2);
134-
REQUIRE(msgSet.pairMap[3].payloadIndex == 1);
156+
REQUIRE((msgSet.messages | count_payloads{}) == 4);
157+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).headerIdx == 0);
158+
REQUIRE((msgSet.messages | get_dataref_indices{0, 0}).payloadIdx == 1);
159+
REQUIRE((msgSet.messages | get_dataref_indices{1, 0}).headerIdx == 2);
160+
REQUIRE((msgSet.messages | get_dataref_indices{1, 0}).payloadIdx == 3);
161+
REQUIRE((msgSet.messages | get_dataref_indices{2, 0}).headerIdx == 4);
162+
REQUIRE((msgSet.messages | get_dataref_indices{2, 0}).payloadIdx == 5);
163+
REQUIRE((msgSet.messages | get_dataref_indices{2, 1}).headerIdx == 4);
164+
REQUIRE((msgSet.messages | get_dataref_indices{2, 1}).payloadIdx == 6);
165+
REQUIRE((msgSet.messages | get_pair{0}).headerIdx == 0);
166+
REQUIRE((msgSet.messages | get_pair{0}).payloadIdx == 1);
167+
REQUIRE((msgSet.messages | get_pair{1}).headerIdx == 2);
168+
REQUIRE((msgSet.messages | get_pair{1}).payloadIdx == 3);
169+
REQUIRE((msgSet.messages | get_pair{2}).headerIdx == 4);
170+
REQUIRE((msgSet.messages | get_pair{2}).payloadIdx == 5);
171+
REQUIRE((msgSet.messages | get_pair{3}).headerIdx == 4);
172+
REQUIRE((msgSet.messages | get_pair{3}).payloadIdx == 6);
135173
}

0 commit comments

Comments
 (0)