Skip to content

Commit 9e652fb

Browse files
ktfktf
committed
DPL: pass forwarded inputs as spans
Purely a change of view: the forwarding helpers, the drop callback and the device now take std::span over the message sets instead of owning vectors. The DataRelayer still owns and hands out vectors; the device keeps them in ownedInputs and passes spans down. No ownership or lifetime change.
1 parent 115d870 commit 9e652fb

6 files changed

Lines changed: 65 additions & 31 deletions

File tree

‎Framework/Core/include/Framework/DataProcessingHelpers.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ struct DataProcessingHelpers {
5454
/// starts the EoS timers and returns the new TransitionHandlingState in case as new state is requested
5555
static TransitionHandlingState updateStateTransition(ServiceRegistryRef const& ref, ProcessingPolicies const& policies);
5656
/// Helper to route messages for forwarding
57-
static std::vector<fair::mq::Parts> routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector<std::vector<fair::mq::MessagePtr>>& currentSetOfInputs,
57+
static std::vector<fair::mq::Parts> routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector<std::span<fair::mq::MessagePtr>>& currentSetOfInputs,
5858
bool copy, bool consume);
5959
/// Helper to route messages for forwarding
6060
static void routeForwardedMessages(FairMQDeviceProxy& proxy, std::span<fair::mq::MessagePtr>& currentSetOfInputs, std::vector<fair::mq::Parts>& forwardedParts,

‎Framework/Core/include/Framework/DataRelayer.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424

2525
#include <cstddef>
2626
#include <mutex>
27+
#include <span>
2728
#include <vector>
2829
#include <functional>
2930

@@ -113,7 +114,7 @@ class DataRelayer
113114
ActivityStats processDanglingInputs(std::vector<ExpirationHandler> const&,
114115
ServiceRegistryRef context, bool createNew);
115116

116-
using OnDropCallback = std::function<void(TimesliceSlot, std::vector<std::vector<fair::mq::MessagePtr>>&, TimesliceIndex::OldestOutputInfo info)>;
117+
using OnDropCallback = std::function<void(TimesliceSlot, std::vector<std::span<fair::mq::MessagePtr>>&, TimesliceIndex::OldestOutputInfo info)>;
117118

118119
// Callback for when some messages are about to be owned by the the DataRelayer
119120
using OnInsertionCallback = std::function<void(ServiceRegistryRef&, std::span<fair::mq::MessagePtr>&)>;

‎Framework/Core/src/DataProcessingDevice.cxx‎

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -587,7 +587,7 @@ auto decongestionCallbackLate = [](AsyncTask& task, size_t aid) -> void {
587587
// the inputs which are shared between this device and others
588588
// to the next one in the daisy chain.
589589
// FIXME: do it in a smarter way than O(N^2)
590-
static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<std::vector<fair::mq::MessagePtr>>& currentSetOfInputs,
590+
static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<std::span<fair::mq::MessagePtr>>& currentSetOfInputs,
591591
TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume = true) {
592592
auto& proxy = registry.get<FairMQDeviceProxy>();
593593

@@ -619,7 +619,7 @@ static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot,
619619
O2_SIGNPOST_END(forwarding, sid, "forwardInputs", "Forwarding done");
620620
};
621621

622-
static auto cleanEarlyForward = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<std::vector<fair::mq::MessagePtr>>& currentSetOfInputs,
622+
static auto cleanEarlyForward = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<std::span<fair::mq::MessagePtr>>& currentSetOfInputs,
623623
TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume = true) {
624624
auto& proxy = registry.get<FairMQDeviceProxy>();
625625

@@ -629,8 +629,7 @@ static auto cleanEarlyForward = [](ServiceRegistryRef registry, TimesliceSlot sl
629629
// Always copy them, because we do not want to actually send them.
630630
// We merely need the side effect of the consume, if applicable.
631631
for (size_t ii = 0, ie = currentSetOfInputs.size(); ii < ie; ++ii) {
632-
auto span = std::span<fair::mq::MessagePtr>(currentSetOfInputs[ii]);
633-
DataProcessingHelpers::cleanForwardedMessages(span, consume);
632+
DataProcessingHelpers::cleanForwardedMessages(currentSetOfInputs[ii], consume);
634633
}
635634

636635
O2_SIGNPOST_END(forwarding, sid, "forwardInputs", "Cleaning done");
@@ -1278,7 +1277,7 @@ void DataProcessingDevice::Run()
12781277
// - we can trigger further events from the queue
12791278
// - we can guarantee this is the last thing we do in the loop (
12801279
// assuming no one else is adding to the queue before this point).
1281-
auto onDrop = [&registry = mServiceRegistry, lid](TimesliceSlot slot, std::vector<std::vector<fair::mq::MessagePtr>>& dropped, TimesliceIndex::OldestOutputInfo oldestOutputInfo) {
1280+
auto onDrop = [&registry = mServiceRegistry, lid](TimesliceSlot slot, std::vector<std::span<fair::mq::MessagePtr>>& dropped, TimesliceIndex::OldestOutputInfo oldestOutputInfo) {
12821281
O2_SIGNPOST_START(device, lid, "run_loop", "Dropping message from slot %" PRIu64 ". Forwarding as needed.", (uint64_t)slot.index);
12831282
ServiceRegistryRef ref{registry};
12841283
ref.get<AsyncQueue>();
@@ -1985,7 +1984,7 @@ void DataProcessingDevice::handleData(ServiceRegistryRef ref, InputChannelInfo&
19851984
nPayloadsPerHeader = 1;
19861985
ii += (nMessages / 2) - 1;
19871986
}
1988-
auto onDrop = [ref](TimesliceSlot slot, std::vector<std::vector<fair::mq::MessagePtr>>& dropped, TimesliceIndex::OldestOutputInfo oldestOutputInfo) {
1987+
auto onDrop = [ref](TimesliceSlot slot, std::vector<std::span<fair::mq::MessagePtr>>& dropped, TimesliceIndex::OldestOutputInfo oldestOutputInfo) {
19891988
O2_SIGNPOST_ID_GENERATE(cid, async_queue);
19901989
O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "onDrop", "Dropping message from slot %zu. Forwarding as needed. Timeslice %zu",
19911990
slot.index, oldestOutputInfo.timeslice.value);
@@ -2163,15 +2162,20 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
21632162
// want to support multithreaded dispatching of operations, I can simply
21642163
// move these to some thread local store and the rest of the lambdas
21652164
// should work just fine.
2166-
std::vector<std::vector<fair::mq::MessagePtr>> currentSetOfInputs;
2165+
std::vector<std::span<fair::mq::MessagePtr>> currentSetOfInputs;
2166+
std::vector<std::vector<fair::mq::MessagePtr>> ownedInputs;
21672167

21682168
//
2169-
auto getInputSpan = [ref, &currentSetOfInputs](TimesliceSlot slot, bool consume = true) {
2169+
auto getInputSpan = [ref, &currentSetOfInputs, &ownedInputs](TimesliceSlot slot, bool consume = true) {
21702170
auto& relayer = ref.get<DataRelayer>();
21712171
if (consume) {
2172-
currentSetOfInputs = relayer.consumeAllInputsForTimeslice(slot);
2172+
ownedInputs = relayer.consumeAllInputsForTimeslice(slot);
21732173
} else {
2174-
currentSetOfInputs = relayer.consumeExistingInputsForTimeslice(slot);
2174+
ownedInputs = relayer.consumeExistingInputsForTimeslice(slot);
2175+
}
2176+
currentSetOfInputs.resize(ownedInputs.size());
2177+
for (size_t i = 0; i < ownedInputs.size(); ++i) {
2178+
currentSetOfInputs[i] = std::span(ownedInputs[i]);
21752179
}
21762180
// Convert raw message indices directly to a DataRef in O(1).
21772181
// Used both by the sequential PartIterator and as the fallback for positional access.
@@ -2252,7 +2256,7 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
22522256
// to avoid double counting them.
22532257
// This was actually the easiest solution we could find for
22542258
// O2-646.
2255-
auto cleanTimers = [&currentSetOfInputs](TimesliceSlot slot, InputRecord& record) {
2259+
auto cleanTimers = [&currentSetOfInputs, &ownedInputs](TimesliceSlot slot, InputRecord& record) {
22562260
assert(record.size() == currentSetOfInputs.size());
22572261
for (size_t ii = 0, ie = record.size(); ii < ie; ++ii) {
22582262
// assuming that for timer inputs we do have exactly one PartRef object
@@ -2265,8 +2269,10 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
22652269
if (input.header == nullptr) {
22662270
continue;
22672271
}
2268-
// This will hopefully delete the message.
2269-
currentSetOfInputs[ii].clear();
2272+
// For the consume=false (Process) path, ownedInputs holds the actual
2273+
// message vectors and the span points into them.
2274+
ownedInputs[ii].clear();
2275+
currentSetOfInputs[ii] = {};
22702276
}
22712277
};
22722278

‎Framework/Core/src/DataProcessingHelpers.cxx‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -393,15 +393,14 @@ void DataProcessingHelpers::cleanForwardedMessages(std::span<fair::mq::MessagePt
393393
}
394394

395395
auto DataProcessingHelpers::routeForwardedMessageSet(FairMQDeviceProxy& proxy,
396-
std::vector<std::vector<fair::mq::MessagePtr>>& currentSetOfInputs,
396+
std::vector<std::span<fair::mq::MessagePtr>>& currentSetOfInputs,
397397
const bool copyByDefault, bool consume) -> std::vector<fair::mq::Parts>
398398
{
399399
// we collect all messages per forward in a map and send them together
400400
std::vector<fair::mq::Parts> forwardedParts(proxy.getNumForwardChannels());
401401

402402
for (size_t ii = 0, ie = currentSetOfInputs.size(); ii < ie; ++ii) {
403-
auto span = std::span<fair::mq::MessagePtr>(currentSetOfInputs[ii]);
404-
routeForwardedMessages(proxy, span, forwardedParts, copyByDefault, consume);
403+
routeForwardedMessages(proxy, currentSetOfInputs[ii], forwardedParts, copyByDefault, consume);
405404
}
406405
return forwardedParts;
407406
};

‎Framework/Core/src/DataRelayer.cxx‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -434,7 +434,11 @@ void DataRelayer::pruneCache(TimesliceSlot slot, OnDropCallback onDrop)
434434
if (anyDropped) {
435435
O2_SIGNPOST_ID_GENERATE(aid, data_relayer);
436436
O2_SIGNPOST_EVENT_EMIT(data_relayer, aid, "pruneCache", "Dropping stuff from slot %zu with timeslice %zu", slot.index, oldestPossibleTimeslice.timeslice.value);
437-
onDrop(slot, dropped, oldestPossibleTimeslice);
437+
std::vector<std::span<fair::mq::MessagePtr>> droppedSpans(dropped.size());
438+
for (size_t ai = 0, ae = dropped.size(); ai != ae; ++ai) {
439+
droppedSpans[ai] = dropped[ai];
440+
}
441+
onDrop(slot, droppedSpans, oldestPossibleTimeslice);
438442
}
439443
}
440444
assert(cache.empty() == false);

‎Framework/Core/test/test_ForwardInputs.cxx‎

Lines changed: 36 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,18 @@
2727
O2_DECLARE_DYNAMIC_LOG(forwarding);
2828
using namespace o2::framework;
2929

30+
// Build a vector of spans over an existing vector-of-vectors for tests that
31+
// construct currentSetOfInputs locally (rather than via consumeAllInputsForTimeslice).
32+
static std::vector<std::span<fair::mq::MessagePtr>> asSpans(std::vector<std::vector<fair::mq::MessagePtr>>& vecs)
33+
{
34+
std::vector<std::span<fair::mq::MessagePtr>> spans;
35+
spans.reserve(vecs.size());
36+
for (auto& v : vecs) {
37+
spans.emplace_back(v);
38+
}
39+
return spans;
40+
}
41+
3042
TEST_CASE("ForwardInputsEmpty")
3143
{
3244
o2::header::DataHeader dh;
@@ -45,7 +57,8 @@ TEST_CASE("ForwardInputsEmpty")
4557

4658
std::vector<std::vector<fair::mq::MessagePtr>> currentSetOfInputs;
4759

48-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
60+
auto spans = asSpans(currentSetOfInputs);
61+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
4962
REQUIRE(result.empty());
5063
}
5164

@@ -96,7 +109,8 @@ TEST_CASE("ForwardInputsSingleMessageSingleRoute")
96109
REQUIRE((messageSet | count_parts{}) == 1);
97110
currentSetOfInputs.emplace_back(std::move(messageSet));
98111

99-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
112+
auto spans = asSpans(currentSetOfInputs);
113+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
100114
REQUIRE(result.size() == 1); // One route
101115
REQUIRE(result[0].Size() == 2); // Two messages for that route
102116
}
@@ -148,7 +162,8 @@ TEST_CASE("ForwardInputsSingleMessageSingleRouteNoConsume")
148162
REQUIRE((messageSet | count_parts{}) == 1);
149163
currentSetOfInputs.emplace_back(std::move(messageSet));
150164

151-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, true);
165+
auto spans = asSpans(currentSetOfInputs);
166+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, true);
152167
REQUIRE(result.size() == 1);
153168
REQUIRE(result[0].Size() == 0); // Because there is a nullptr, we do not forward this as it was already consumed.
154169
}
@@ -204,7 +219,8 @@ TEST_CASE("ForwardInputsSingleMessageSingleRouteAtEOS")
204219
REQUIRE((messageSet | count_parts{}) == 1);
205220
currentSetOfInputs.emplace_back(std::move(messageSet));
206221

207-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
222+
auto spans = asSpans(currentSetOfInputs);
223+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
208224
REQUIRE(result.size() == 1); // One route
209225
REQUIRE(result[0].Size() == 0); // FIXME: this is an actual error. It should be 2. However it cannot really happen.
210226
// Correct behavior below:
@@ -263,7 +279,8 @@ TEST_CASE("ForwardInputsSingleMessageSingleRouteWithOldestPossible")
263279
REQUIRE((messageSet | count_parts{}) == 1);
264280
currentSetOfInputs.emplace_back(std::move(messageSet));
265281

266-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
282+
auto spans = asSpans(currentSetOfInputs);
283+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
267284
REQUIRE(result.size() == 1); // One route
268285
REQUIRE(result[0].Size() == 0); // FIXME: this is actually wrong
269286
// FIXME: actually correct behavior below
@@ -329,7 +346,8 @@ TEST_CASE("ForwardInputsSingleMessageMultipleRoutes")
329346
REQUIRE((messageSet | count_parts{}) == 1);
330347
currentSetOfInputs.emplace_back(std::move(messageSet));
331348

332-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
349+
auto spans = asSpans(currentSetOfInputs);
350+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
333351
REQUIRE(result.size() == 2); // Two routes
334352
REQUIRE(result[0].Size() == 2); // Two messages per route
335353
REQUIRE(result[1].Size() == 0); // Only the first DPL matched channel matters
@@ -393,7 +411,8 @@ TEST_CASE("ForwardInputsSingleMessageMultipleRoutesExternals")
393411
REQUIRE((messageSet | count_parts{}) == 1);
394412
currentSetOfInputs.emplace_back(std::move(messageSet));
395413

396-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
414+
auto spans = asSpans(currentSetOfInputs);
415+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
397416
REQUIRE(result.size() == 2); // Two routes
398417
REQUIRE(result[0].Size() == 2); // With external matching channels, we need to copy and then forward
399418
REQUIRE(result[1].Size() == 2); //
@@ -473,7 +492,8 @@ TEST_CASE("ForwardInputsMultiMessageMultipleRoutes")
473492
currentSetOfInputs.emplace_back(std::move(messageSet2));
474493
REQUIRE(currentSetOfInputs.size() == 2);
475494

476-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
495+
auto spans = asSpans(currentSetOfInputs);
496+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
477497
REQUIRE(result.size() == 2); // Two routes
478498
REQUIRE(result[0].Size() == 2); //
479499
REQUIRE(result[1].Size() == 2); //
@@ -537,7 +557,8 @@ TEST_CASE("ForwardInputsSingleMessageMultipleRoutesOnlyOneMatches")
537557
REQUIRE((messageSet | count_parts{}) == 1);
538558
currentSetOfInputs.emplace_back(std::move(messageSet));
539559

540-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
560+
auto spans = asSpans(currentSetOfInputs);
561+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
541562
REQUIRE(result.size() == 2); // Two routes
542563
REQUIRE(result[0].Size() == 0); // Two messages per route
543564
REQUIRE(result[1].Size() == 2); // Two messages per route
@@ -621,7 +642,8 @@ TEST_CASE("ForwardInputsSplitPayload")
621642
REQUIRE((messageSet | count_parts{}) == 2);
622643
currentSetOfInputs.emplace_back(std::move(messageSet));
623644

624-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
645+
auto spans = asSpans(currentSetOfInputs);
646+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
625647
REQUIRE(result.size() == 2); // Two routes
626648
CHECK(result[0].Size() == 2); // No messages on this route
627649
CHECK(result[1].Size() == 3);
@@ -742,7 +764,8 @@ TEST_CASE("ForwardInputEOSSingleRoute")
742764
REQUIRE((messageSet | count_parts{}) == 1);
743765
currentSetOfInputs.emplace_back(std::move(messageSet));
744766

745-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
767+
auto spans = asSpans(currentSetOfInputs);
768+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
746769
REQUIRE(result.size() == 1); // One route
747770
REQUIRE(result[0].Size() == 0); // Oldest possible timeframe should not be forwarded
748771
}
@@ -788,7 +811,8 @@ TEST_CASE("ForwardInputOldestPossibleSingleRoute")
788811
REQUIRE((messageSet | count_parts{}) == 1);
789812
currentSetOfInputs.emplace_back(std::move(messageSet));
790813

791-
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
814+
auto spans = asSpans(currentSetOfInputs);
815+
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
792816
REQUIRE(result.size() == 1); // One route
793817
REQUIRE(result[0].Size() == 0); // Oldest possible timeframe should not be forwarded
794818
}

0 commit comments

Comments
 (0)