Project
Loading...
Searching...
No Matches
test_ForwardInputs.cxx
Go to the documentation of this file.
1// Copyright 2019-2025 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#include <catch_amalgamated.hpp>
13#include "Headers/DataHeader.h"
14#include "Framework/DataProcessingHeader.h"
15#include "Framework/DataProcessingHelpers.h"
16#include "Framework/SourceInfoHeader.h"
17#include "Framework/DomainInfoHeader.h"
18#include "Framework/Signpost.h"
19#include "Framework/MessageSet.h"
20#include "Framework/FairMQDeviceProxy.h"
21#include "Headers/Stack.h"
22#include "MemoryResources/MemoryResources.h"
23#include <fairmq/TransportFactory.h>
24#include <fairmq/Channel.h>
25#include <vector>
26
27O2_DECLARE_DYNAMIC_LOG(forwarding);
28using namespace o2::framework;
29
30TEST_CASE("ForwardInputsEmpty")
31{
32 o2::header::DataHeader dh;
33 dh.dataDescription = "CLUSTERS";
34 dh.dataOrigin = "TPC";
35 dh.subSpecification = 0;
36 dh.splitPayloadIndex = 0;
37 dh.splitPayloadParts = 1;
38
39 o2::framework::DataProcessingHeader dph{0, 1};
40 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
41
42 bool consume = true;
43 bool copyByDefault = true;
44 FairMQDeviceProxy proxy;
45
46 std::vector<MessageSet> currentSetOfInputs;
47
48 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
49 REQUIRE(result.empty());
50}
51
52TEST_CASE("ForwardInputsSingleMessageSingleRoute")
53{
54 o2::header::DataHeader dh;
55 dh.dataOrigin = "TST";
56 dh.dataDescription = "A";
57 dh.subSpecification = 0;
58 dh.splitPayloadIndex = 0;
59 dh.splitPayloadParts = 1;
60
61 o2::framework::DataProcessingHeader dph{0, 1};
62 std::vector<fair::mq::Channel> channels{
63 fair::mq::Channel("from_A_to_B")};
64
65 bool consume = true;
66 bool copyByDefault = true;
67 FairMQDeviceProxy proxy;
68 std::vector<ForwardRoute> routes{ForwardRoute{
69 .timeslice = 0,
70 .maxTimeslices = 1,
71 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
72 .channel = "from_A_to_B",
73 .policy = nullptr,
74 }};
75
76 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
77 for (auto& channel : channels) {
78 if (channel.GetName() == channelName) {
79 return channel;
80 }
81 }
82 throw std::runtime_error("Channel not found");
83 };
84
85 proxy.bind({}, {}, routes, findChannelByName, nullptr);
86
87 std::vector<MessageSet> currentSetOfInputs;
88 MessageSet messageSet;
89
90 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
91 fair::mq::MessagePtr payload(transport->CreateMessage());
92 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
93 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
94 messageSet.add(PartRef{std::move(header), std::move(payload)});
95 REQUIRE(messageSet.size() == 1);
96 currentSetOfInputs.emplace_back(std::move(messageSet));
97
98 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
99 REQUIRE(result.size() == 1); // One route
100 REQUIRE(result[0].Size() == 2); // Two messages for that route
101}
102
103TEST_CASE("ForwardInputsSingleMessageSingleRouteNoConsume")
104{
105 o2::header::DataHeader dh;
106 dh.dataOrigin = "TST";
107 dh.dataDescription = "A";
108 dh.subSpecification = 0;
109 dh.splitPayloadIndex = 0;
110 dh.splitPayloadParts = 1;
111
112 o2::framework::DataProcessingHeader dph{0, 1};
113 std::vector<fair::mq::Channel> channels{
114 fair::mq::Channel("from_A_to_B")};
115
116 bool copyByDefault = false;
117 FairMQDeviceProxy proxy;
118 std::vector<ForwardRoute> routes{ForwardRoute{
119 .timeslice = 0,
120 .maxTimeslices = 1,
121 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
122 .channel = "from_A_to_B",
123 .policy = nullptr,
124 }};
125
126 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
127 for (auto& channel : channels) {
128 if (channel.GetName() == channelName) {
129 return channel;
130 }
131 }
132 throw std::runtime_error("Channel not found");
133 };
134
135 proxy.bind({}, {}, routes, findChannelByName, nullptr);
136
137 std::vector<MessageSet> currentSetOfInputs;
138 MessageSet messageSet;
139
140 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
141 fair::mq::MessagePtr payload(nullptr);
142 REQUIRE(payload.get() == nullptr);
143 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
144 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
145 messageSet.add(PartRef{std::move(header), std::move(payload)});
146 REQUIRE(messageSet.size() == 1);
147 currentSetOfInputs.emplace_back(std::move(messageSet));
148
149 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, true);
150 REQUIRE(result.size() == 1);
151 REQUIRE(result[0].Size() == 0); // Because there is a nullptr, we do not forward this as it was already consumed.
152}
153
154TEST_CASE("ForwardInputsSingleMessageSingleRouteAtEOS")
155{
156 o2::header::DataHeader dh;
157 dh.dataOrigin = "TST";
158 dh.dataDescription = "A";
159 dh.subSpecification = 0;
160 dh.splitPayloadIndex = 0;
161 dh.splitPayloadParts = 1;
162
163 o2::framework::DataProcessingHeader dph{0, 1};
164
165 o2::framework::SourceInfoHeader sih{};
166
167 std::vector<fair::mq::Channel> channels{
168 fair::mq::Channel("from_A_to_B")};
169
170 bool consume = true;
171 bool copyByDefault = true;
172 FairMQDeviceProxy proxy;
173 std::vector<ForwardRoute> routes{ForwardRoute{
174 .timeslice = 0,
175 .maxTimeslices = 1,
176 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
177 .channel = "from_A_to_B",
178 .policy = nullptr,
179 }};
180
181 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
182 for (auto& channel : channels) {
183 if (channel.GetName() == channelName) {
184 return channel;
185 }
186 }
187 throw std::runtime_error("Channel not found");
188 };
189
190 proxy.bind({}, {}, routes, findChannelByName, nullptr);
191
192 std::vector<MessageSet> currentSetOfInputs;
193 MessageSet messageSet;
194
195 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
196 fair::mq::MessagePtr payload(transport->CreateMessage());
197 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
198 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph, sih});
199 REQUIRE(o2::header::get<SourceInfoHeader*>(header->GetData()));
200 messageSet.add(PartRef{std::move(header), std::move(payload)});
201 REQUIRE(messageSet.size() == 1);
202 currentSetOfInputs.emplace_back(std::move(messageSet));
203
204 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
205 REQUIRE(result.size() == 1); // One route
206 REQUIRE(result[0].Size() == 0); // FIXME: this is an actual error. It should be 2. However it cannot really happen.
207 // Correct behavior below:
208 // REQUIRE(result[0].Size() == 2);
209 // REQUIRE(o2::header::get<SourceInfoHeader*>(result[0].At(0)->GetData()) == nullptr);
210}
211
212TEST_CASE("ForwardInputsSingleMessageSingleRouteWithOldestPossible")
213{
214 o2::header::DataHeader dh;
215 dh.dataOrigin = "TST";
216 dh.dataDescription = "A";
217 dh.subSpecification = 0;
218 dh.splitPayloadIndex = 0;
219 dh.splitPayloadParts = 1;
220
221 o2::framework::DataProcessingHeader dph{0, 1};
222
223 o2::framework::DomainInfoHeader dih{};
224
225 std::vector<fair::mq::Channel> channels{
226 fair::mq::Channel("from_A_to_B")};
227
228 bool consume = true;
229 bool copyByDefault = true;
230 FairMQDeviceProxy proxy;
231 std::vector<ForwardRoute> routes{ForwardRoute{
232 .timeslice = 0,
233 .maxTimeslices = 1,
234 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
235 .channel = "from_A_to_B",
236 .policy = nullptr,
237 }};
238
239 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
240 for (auto& channel : channels) {
241 if (channel.GetName() == channelName) {
242 return channel;
243 }
244 }
245 throw std::runtime_error("Channel not found");
246 };
247
248 proxy.bind({}, {}, routes, findChannelByName, nullptr);
249
250 std::vector<MessageSet> currentSetOfInputs;
251 MessageSet messageSet;
252
253 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
254 fair::mq::MessagePtr payload(transport->CreateMessage());
255 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
256 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph, dih});
257 REQUIRE(o2::header::get<DomainInfoHeader*>(header->GetData()));
258 messageSet.add(PartRef{std::move(header), std::move(payload)});
259 REQUIRE(messageSet.size() == 1);
260 currentSetOfInputs.emplace_back(std::move(messageSet));
261
262 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
263 REQUIRE(result.size() == 1); // One route
264 REQUIRE(result[0].Size() == 0); // FIXME: this is actually wrong
265 // FIXME: actually correct behavior below
266 // REQUIRE(result[0].Size() == 2); // Two messages
267 // REQUIRE(o2::header::get<DomainInfoHeader*>(result[0].At(0)->GetData()) == nullptr); // it should not have the end of stream
268}
269
270TEST_CASE("ForwardInputsSingleMessageMultipleRoutes")
271{
272 o2::header::DataHeader dh;
273 dh.dataOrigin = "TST";
274 dh.dataDescription = "A";
275 dh.subSpecification = 0;
276 dh.splitPayloadIndex = 0;
277 dh.splitPayloadParts = 1;
278
279 o2::framework::DataProcessingHeader dph{0, 1};
280
281 std::vector<fair::mq::Channel> channels{
282 fair::mq::Channel("from_A_to_B"),
283 fair::mq::Channel("from_A_to_C"),
284 };
285
286 bool consume = true;
287 bool copyByDefault = true;
288 FairMQDeviceProxy proxy;
289 std::vector<ForwardRoute> routes{
290 ForwardRoute{
291 .timeslice = 0,
292 .maxTimeslices = 1,
293 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
294 .channel = "from_A_to_B",
295 .policy = nullptr,
296 },
297 ForwardRoute{
298 .timeslice = 0,
299 .maxTimeslices = 1,
300 .matcher = {"binding2", ConcreteDataMatcher{"TST", "A", 0}},
301 .channel = "from_A_to_C",
302 .policy = nullptr,
303 }};
304
305 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
306 for (auto& channel : channels) {
307 if (channel.GetName() == channelName) {
308 return channel;
309 }
310 }
311 throw std::runtime_error("Channel not found");
312 };
313
314 proxy.bind({}, {}, routes, findChannelByName, nullptr);
315
316 std::vector<MessageSet> currentSetOfInputs;
317 MessageSet messageSet;
318
319 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
320 fair::mq::MessagePtr payload(transport->CreateMessage());
321 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
322 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
323 messageSet.add(PartRef{std::move(header), std::move(payload)});
324 REQUIRE(messageSet.size() == 1);
325 currentSetOfInputs.emplace_back(std::move(messageSet));
326
327 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
328 REQUIRE(result.size() == 2); // Two routes
329 REQUIRE(result[0].Size() == 2); // Two messages per route
330 REQUIRE(result[1].Size() == 0); // Only the first DPL matched channel matters
331}
332
333TEST_CASE("ForwardInputsSingleMessageMultipleRoutesExternals")
334{
335 o2::header::DataHeader dh;
336 dh.dataOrigin = "TST";
337 dh.dataDescription = "A";
338 dh.subSpecification = 0;
339 dh.splitPayloadIndex = 0;
340 dh.splitPayloadParts = 1;
341
342 o2::framework::DataProcessingHeader dph{0, 1};
343
344 std::vector<fair::mq::Channel> channels{
345 fair::mq::Channel("external"),
346 fair::mq::Channel("from_A_to_C"),
347 };
348
349 bool consume = true;
350 bool copyByDefault = true;
351 FairMQDeviceProxy proxy;
352 std::vector<ForwardRoute> routes{
353 ForwardRoute{
354 .timeslice = 0,
355 .maxTimeslices = 1,
356 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
357 .channel = "external",
358 .policy = nullptr,
359 },
360 ForwardRoute{
361 .timeslice = 0,
362 .maxTimeslices = 1,
363 .matcher = {"binding2", ConcreteDataMatcher{"TST", "A", 0}},
364 .channel = "from_A_to_C",
365 .policy = nullptr,
366 }};
367
368 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
369 for (auto& channel : channels) {
370 if (channel.GetName() == channelName) {
371 return channel;
372 }
373 }
374 throw std::runtime_error("Channel not found");
375 };
376
377 proxy.bind({}, {}, routes, findChannelByName, nullptr);
378
379 std::vector<MessageSet> currentSetOfInputs;
380 MessageSet messageSet;
381
382 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
383 fair::mq::MessagePtr payload(transport->CreateMessage());
384 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
385 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
386 messageSet.add(PartRef{std::move(header), std::move(payload)});
387 REQUIRE(messageSet.size() == 1);
388 currentSetOfInputs.emplace_back(std::move(messageSet));
389
390 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
391 REQUIRE(result.size() == 2); // Two routes
392 REQUIRE(result[0].Size() == 2); // With external matching channels, we need to copy and then forward
393 REQUIRE(result[1].Size() == 2); //
394}
395
396TEST_CASE("ForwardInputsMultiMessageMultipleRoutes")
397{
398 o2::header::DataHeader dh1;
399 dh1.dataOrigin = "TST";
400 dh1.dataDescription = "A";
401 dh1.subSpecification = 0;
402 dh1.splitPayloadIndex = 0;
403 dh1.splitPayloadParts = 1;
404
405 o2::header::DataHeader dh2;
406 dh2.dataOrigin = "TST";
407 dh2.dataDescription = "B";
408 dh2.subSpecification = 0;
409 dh2.splitPayloadIndex = 0;
410 dh2.splitPayloadParts = 1;
411
412 o2::framework::DataProcessingHeader dph{0, 1};
413
414 std::vector<fair::mq::Channel> channels{
415 fair::mq::Channel("from_A_to_B"),
416 fair::mq::Channel("from_A_to_C"),
417 };
418
419 bool consume = true;
420 bool copyByDefault = true;
421 FairMQDeviceProxy proxy;
422 std::vector<ForwardRoute> routes{
423 ForwardRoute{
424 .timeslice = 0,
425 .maxTimeslices = 1,
426 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
427 .channel = "from_A_to_B",
428 .policy = nullptr,
429 },
430 ForwardRoute{
431 .timeslice = 0,
432 .maxTimeslices = 1,
433 .matcher = {"binding2", ConcreteDataMatcher{"TST", "B", 0}},
434 .channel = "from_A_to_C",
435 .policy = nullptr,
436 }};
437
438 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
439 for (auto& channel : channels) {
440 if (channel.GetName() == channelName) {
441 return channel;
442 }
443 }
444 throw std::runtime_error("Channel not found");
445 };
446
447 proxy.bind({}, {}, routes, findChannelByName, nullptr);
448
449 std::vector<MessageSet> currentSetOfInputs;
450
451 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
452 fair::mq::MessagePtr payload1(transport->CreateMessage());
453 fair::mq::MessagePtr payload2(transport->CreateMessage());
454 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
455 auto header1 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh1, dph});
456 MessageSet messageSet1;
457 messageSet1.add(PartRef{std::move(header1), std::move(payload1)});
458 REQUIRE(messageSet1.size() == 1);
459
460 auto header2 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh2, dph});
461 MessageSet messageSet2;
462 messageSet2.add(PartRef{std::move(header2), std::move(payload2)});
463 REQUIRE(messageSet2.size() == 1);
464 currentSetOfInputs.emplace_back(std::move(messageSet1));
465 currentSetOfInputs.emplace_back(std::move(messageSet2));
466 REQUIRE(currentSetOfInputs.size() == 2);
467
468 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
469 REQUIRE(result.size() == 2); // Two routes
470 REQUIRE(result[0].Size() == 2); //
471 REQUIRE(result[1].Size() == 2); //
472}
473
474TEST_CASE("ForwardInputsSingleMessageMultipleRoutesOnlyOneMatches")
475{
476 o2::header::DataHeader dh;
477 dh.dataOrigin = "TST";
478 dh.dataDescription = "A";
479 dh.subSpecification = 0;
480 dh.splitPayloadIndex = 0;
481 dh.splitPayloadParts = 1;
482
483 o2::framework::DataProcessingHeader dph{0, 1};
484
485 std::vector<fair::mq::Channel> channels{
486 fair::mq::Channel("from_A_to_B"),
487 fair::mq::Channel("from_A_to_C"),
488 };
489
490 bool consume = true;
491 bool copyByDefault = true;
492 FairMQDeviceProxy proxy;
493 std::vector<ForwardRoute> routes{
494 ForwardRoute{
495 .timeslice = 0,
496 .maxTimeslices = 1,
497 .matcher = {"binding", ConcreteDataMatcher{"TST", "B", 0}},
498 .channel = "from_A_to_B",
499 .policy = nullptr,
500 },
501 ForwardRoute{
502 .timeslice = 0,
503 .maxTimeslices = 1,
504 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
505 .channel = "from_A_to_C",
506 .policy = nullptr,
507 }};
508
509 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
510 for (auto& channel : channels) {
511 if (channel.GetName() == channelName) {
512 return channel;
513 }
514 }
515 throw std::runtime_error("Channel not found");
516 };
517
518 proxy.bind({}, {}, routes, findChannelByName, nullptr);
519
520 std::vector<MessageSet> currentSetOfInputs;
521 MessageSet messageSet;
522
523 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
524 fair::mq::MessagePtr payload(transport->CreateMessage());
525 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
526 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
527 messageSet.add(PartRef{std::move(header), std::move(payload)});
528 REQUIRE(messageSet.size() == 1);
529 currentSetOfInputs.emplace_back(std::move(messageSet));
530
531 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
532 REQUIRE(result.size() == 2); // Two routes
533 REQUIRE(result[0].Size() == 0); // Two messages per route
534 REQUIRE(result[1].Size() == 2); // Two messages per route
535}
536
537TEST_CASE("ForwardInputsSplitPayload")
538{
539 o2::header::DataHeader dh;
540 dh.dataOrigin = "TST";
541 dh.dataDescription = "A";
542 dh.subSpecification = 0;
543 dh.splitPayloadIndex = 2;
544 dh.splitPayloadParts = 2;
545
546 o2::header::DataHeader dh2;
547 dh2.dataOrigin = "TST";
548 dh2.dataDescription = "B";
549 dh2.subSpecification = 0;
550 dh2.splitPayloadIndex = 0;
551 dh2.splitPayloadParts = 1;
552
553 o2::framework::DataProcessingHeader dph{0, 1};
554
555 std::vector<fair::mq::Channel> channels{
556 fair::mq::Channel("from_A_to_B"),
557 fair::mq::Channel("from_A_to_C"),
558 };
559
560 bool consume = true;
561 bool copyByDefault = true;
562 FairMQDeviceProxy proxy;
563 std::vector<ForwardRoute> routes{
564 ForwardRoute{
565 .timeslice = 0,
566 .maxTimeslices = 1,
567 .matcher = {"binding", ConcreteDataMatcher{"TST", "B", 0}},
568 .channel = "from_A_to_B",
569 .policy = nullptr,
570 },
571 ForwardRoute{
572 .timeslice = 0,
573 .maxTimeslices = 1,
574 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
575 .channel = "from_A_to_C",
576 .policy = nullptr,
577 }};
578
579 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
580 for (auto& channel : channels) {
581 if (channel.GetName() == channelName) {
582 return channel;
583 }
584 }
585 throw std::runtime_error("Channel not found");
586 };
587
588 proxy.bind({}, {}, routes, findChannelByName, nullptr);
589
590 std::vector<MessageSet> currentSetOfInputs;
591 MessageSet messageSet;
592
593 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
594 fair::mq::MessagePtr payload1(transport->CreateMessage());
595 fair::mq::MessagePtr payload2(transport->CreateMessage());
596 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
597 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
598 std::vector<std::unique_ptr<fair::mq::Message>> messages;
599 messages.push_back(std::move(header));
600 messages.push_back(std::move(payload1));
601 messages.push_back(std::move(payload2));
602 auto fillMessages = [&messages](size_t t) -> fair::mq::MessagePtr {
603 return std::move(messages[t]);
604 };
605 messageSet.add(fillMessages, 3);
606 auto header2 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh2, dph});
607 PartRef part{std::move(header2), transport->CreateMessage()};
608 messageSet.add(std::move(part));
609
610 REQUIRE(messageSet.size() == 2);
611 currentSetOfInputs.emplace_back(std::move(messageSet));
612
613 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
614 REQUIRE(result.size() == 2); // Two routes
615 CHECK(result[0].Size() == 2); // No messages on this route
616 CHECK(result[1].Size() == 3);
617}
618
619TEST_CASE("ForwardInputsSplitPayloadNoMessageSet")
620{
621 o2::header::DataHeader dh;
622 dh.dataOrigin = "TST";
623 dh.dataDescription = "A";
624 dh.subSpecification = 0;
625 dh.splitPayloadIndex = 2;
626 dh.splitPayloadParts = 2;
627
628 o2::header::DataHeader dh2;
629 dh2.dataOrigin = "TST";
630 dh2.dataDescription = "B";
631 dh2.subSpecification = 0;
632 dh2.splitPayloadIndex = 0;
633 dh2.splitPayloadParts = 1;
634
635 o2::framework::DataProcessingHeader dph{0, 1};
636
637 std::vector<fair::mq::Channel> channels{
638 fair::mq::Channel("from_A_to_B"),
639 fair::mq::Channel("from_A_to_C"),
640 };
641
642 bool consume = true;
643 bool copyByDefault = true;
644 FairMQDeviceProxy proxy;
645 std::vector<ForwardRoute> routes{
646 ForwardRoute{
647 .timeslice = 0,
648 .maxTimeslices = 1,
649 .matcher = {"binding", ConcreteDataMatcher{"TST", "B", 0}},
650 .channel = "from_A_to_B",
651 .policy = nullptr,
652 },
653 ForwardRoute{
654 .timeslice = 0,
655 .maxTimeslices = 1,
656 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
657 .channel = "from_A_to_C",
658 .policy = nullptr,
659 }};
660
661 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
662 for (auto& channel : channels) {
663 if (channel.GetName() == channelName) {
664 return channel;
665 }
666 }
667 throw std::runtime_error("Channel not found");
668 };
669
670 proxy.bind({}, {}, routes, findChannelByName, nullptr);
671
672 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
673 fair::mq::MessagePtr payload1(transport->CreateMessage());
674 fair::mq::MessagePtr payload2(transport->CreateMessage());
675 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
676 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh, dph});
677 std::vector<std::unique_ptr<fair::mq::Message>> messages;
678 messages.push_back(std::move(header));
679 messages.push_back(std::move(payload1));
680 messages.push_back(std::move(payload2));
681 auto header2 = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dh2, dph});
682 messages.push_back(std::move(header2));
683 messages.push_back(transport->CreateMessage());
684
685 std::vector<fair::mq::Parts> result(2);
686 auto span = std::span(messages);
687 o2::framework::DataProcessingHelpers::routeForwardedMessages(proxy, span, result, copyByDefault, consume);
688 REQUIRE(result.size() == 2); // Two routes
689 CHECK(result[0].Size() == 2); // No messages on this route
690 CHECK(result[1].Size() == 3);
691}
692
693TEST_CASE("ForwardInputEOSSingleRoute")
694{
695 o2::framework::SourceInfoHeader sih{};
696
697 std::vector<fair::mq::Channel> channels{
698 fair::mq::Channel("from_A_to_B")};
699
700 bool consume = true;
701 bool copyByDefault = true;
702 FairMQDeviceProxy proxy;
703 std::vector<ForwardRoute> routes{ForwardRoute{
704 .timeslice = 0,
705 .maxTimeslices = 1,
706 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
707 .channel = "from_A_to_B",
708 .policy = nullptr,
709 }};
710
711 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
712 for (auto& channel : channels) {
713 if (channel.GetName() == channelName) {
714 return channel;
715 }
716 }
717 throw std::runtime_error("Channel not found");
718 };
719
720 proxy.bind({}, {}, routes, findChannelByName, nullptr);
721
722 std::vector<MessageSet> currentSetOfInputs;
723 MessageSet messageSet;
724
725 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
726 fair::mq::MessagePtr payload(transport->CreateMessage());
727 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
728 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, sih});
729 messageSet.add(PartRef{std::move(header), std::move(payload)});
730 REQUIRE(messageSet.size() == 1);
731 currentSetOfInputs.emplace_back(std::move(messageSet));
732
733 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
734 REQUIRE(result.size() == 1); // One route
735 REQUIRE(result[0].Size() == 0); // Oldest possible timeframe should not be forwarded
736}
737
738TEST_CASE("ForwardInputOldestPossibleSingleRoute")
739{
740 o2::framework::DomainInfoHeader dih{};
741
742 std::vector<fair::mq::Channel> channels{
743 fair::mq::Channel("from_A_to_B")};
744
745 bool consume = true;
746 bool copyByDefault = true;
747 FairMQDeviceProxy proxy;
748 std::vector<ForwardRoute> routes{ForwardRoute{
749 .timeslice = 0,
750 .maxTimeslices = 1,
751 .matcher = {"binding", ConcreteDataMatcher{"TST", "A", 0}},
752 .channel = "from_A_to_B",
753 .policy = nullptr,
754 }};
755
756 auto findChannelByName = [&channels](std::string const& channelName) -> fair::mq::Channel& {
757 for (auto& channel : channels) {
758 if (channel.GetName() == channelName) {
759 return channel;
760 }
761 }
762 throw std::runtime_error("Channel not found");
763 };
764
765 proxy.bind({}, {}, routes, findChannelByName, nullptr);
766
767 std::vector<MessageSet> currentSetOfInputs;
768 MessageSet messageSet;
769
770 auto transport = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
771 fair::mq::MessagePtr payload(transport->CreateMessage());
772 auto channelAlloc = o2::pmr::getTransportAllocator(transport.get());
773 auto header = o2::pmr::getMessage(o2::header::Stack{channelAlloc, dih});
774 messageSet.add(PartRef{std::move(header), std::move(payload)});
775 REQUIRE(messageSet.size() == 1);
776 currentSetOfInputs.emplace_back(std::move(messageSet));
777
778 auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
779 REQUIRE(result.size() == 1); // One route
780 REQUIRE(result[0].Size() == 0); // Oldest possible timeframe should not be forwarded
781}
routes
std::vector< OutputRoute > routes
Definition CCDBHelpers.cxx:58
O2_DECLARE_DYNAMIC_LOG
#define O2_DECLARE_DYNAMIC_LOG(name)
Definition Signpost.h:489
Measurement::Size
@ Size
o2::framework::FairMQDeviceProxy::bind
void bind(std::vector< OutputRoute > const &outputs, std::vector< InputRoute > const &inputs, std::vector< ForwardRoute > const &forwards, std::function< fair::mq::Channel &(std::string const &)> bindChannelByName, std::function< bool(void)> newStateRequestedCallback)
Definition FairMQDeviceProxy.cxx:231
CHECK
#define CHECK
Definition clusterProcessing.cxx:25
result
GLuint64EXT * result
Definition glcorearb.h:5662
o2::framework
Defining PrimaryVertex explicitly as messageable.
Definition BunchFilling.h:112
o2::framework::TEST_CASE
TEST_CASE("test_prepareArguments")
Definition test_DeviceSpecHelpers.cxx:102
o2::pmr::getMessage
fair::mq::MessagePtr getMessage(ContainerT &&container, FairMQMemoryResource *targetResource=nullptr)
Definition MemoryResources.h:53
o2::framework::DataProcessingHelpers::routeForwardedMessageSet
static std::vector< fair::mq::Parts > routeForwardedMessageSet(FairMQDeviceProxy &proxy, std::vector< MessageSet > &currentSetOfInputs, bool copy, bool consume)
Helper to route messages for forwarding.
Definition DataProcessingHelpers.cxx:395
o2::framework::DataProcessingHelpers::routeForwardedMessages
static void routeForwardedMessages(FairMQDeviceProxy &proxy, std::span< fair::mq::MessagePtr > &currentSetOfInputs, std::vector< fair::mq::Parts > &forwardedParts, bool copy, bool consume)
Helper to route messages for forwarding.
Definition DataProcessingHelpers.cxx:231
o2::framework::DomainInfoHeader
a BaseHeader with domain information from the source
Definition DomainInfoHeader.h:35
o2::framework::MessageSet::size
size_t size() const
get number of in-flight O2 messages
Definition MessageSet.h:86
o2::framework::MessageSet::add
void add(PartRef &&ref)
Definition MessageSet.h:123
o2::framework::PartRef
Reference to an inflight part.
Definition PartRef.h:24
o2::framework::SourceInfoHeader
a BaseHeader with state information from the source
Definition SourceInfoHeader.h:39
o2::header::DataHeader
the main header struct
Definition DataHeader.h:619
o2::header::DataHeader::splitPayloadParts
SplitPayloadPartsType splitPayloadParts
Definition DataHeader.h:647
o2::header::DataHeader::dataDescription
DataDescription dataDescription
Definition DataHeader.h:637
o2::header::DataHeader::subSpecification
SubSpecificationType subSpecification
Definition DataHeader.h:657
o2::header::DataHeader::splitPayloadIndex
SplitPayloadIndexType splitPayloadIndex
Definition DataHeader.h:662
o2::header::Stack
a move-only header stack with serialized headers This is the flat buffer where all the headers in a m...
Definition Stack.h:33
channels
std::vector< ChannelData > channels
Definition test_ctf_io_fdd.cxx:39