Skip to content

DPL: pass forwarded inputs as spans #15672

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ktf merged 1 commit into from
Aug 10, 2026
Merged

DPL: pass forwarded inputs as spans #15672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Framework/Core/include/Framework/DataProcessingHelpers.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ struct DataProcessingHelpers {
/// starts the EoS timers and returns the new TransitionHandlingState in case as new state is requested
static TransitionHandlingState updateStateTransition(ServiceRegistryRef const& ref, ProcessingPolicies const& policies);
/// Helper to route messages for forwarding
static std::vector<fair::mq::Parts> routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector<std::vector<fair::mq::MessagePtr>>& currentSetOfInputs,
static std::vector<fair::mq::Parts> routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector<std::span<fair::mq::MessagePtr>>& currentSetOfInputs,
bool copy, bool consume);
/// Helper to route messages for forwarding
static void routeForwardedMessages(FairMQDeviceProxy& proxy, std::span<fair::mq::MessagePtr>& currentSetOfInputs, std::vector<fair::mq::Parts>& forwardedParts,
Expand Down
3 changes: 2 additions & 1 deletion Framework/Core/include/Framework/DataRelayer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

#include <cstddef>
#include <mutex>
#include <span>
#include <vector>
#include <functional>

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

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

// Callback for when some messages are about to be owned by the the DataRelayer
using OnInsertionCallback = std::function<void(ServiceRegistryRef&, std::span<fair::mq::MessagePtr>&)>;
Expand Down
32 changes: 19 additions & 13 deletions Framework/Core/src/DataProcessingDevice.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -587,7 +587,7 @@ auto decongestionCallbackLate = [](AsyncTask& task, size_t aid) -> void {
// the inputs which are shared between this device and others
// to the next one in the daisy chain.
// FIXME: do it in a smarter way than O(N^2)
static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<std::vector<fair::mq::MessagePtr>>& currentSetOfInputs,
static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector<std::span<fair::mq::MessagePtr>>& currentSetOfInputs,
TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume = true) {
auto& proxy = registry.get<FairMQDeviceProxy>();

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

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

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

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

//
auto getInputSpan = [ref, &currentSetOfInputs](TimesliceSlot slot, bool consume = true) {
auto getInputSpan = [ref, &currentSetOfInputs, &ownedInputs](TimesliceSlot slot, bool consume = true) {
auto& relayer = ref.get<DataRelayer>();
if (consume) {
currentSetOfInputs = relayer.consumeAllInputsForTimeslice(slot);
ownedInputs = relayer.consumeAllInputsForTimeslice(slot);
} else {
currentSetOfInputs = relayer.consumeExistingInputsForTimeslice(slot);
ownedInputs = relayer.consumeExistingInputsForTimeslice(slot);
}
currentSetOfInputs.resize(ownedInputs.size());
for (size_t i = 0; i < ownedInputs.size(); ++i) {
currentSetOfInputs[i] = std::span(ownedInputs[i]);
}
// Convert raw message indices directly to a DataRef in O(1).
// Used both by the sequential PartIterator and as the fallback for positional access.
Expand Down Expand Up @@ -2252,7 +2256,7 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
// to avoid double counting them.
// This was actually the easiest solution we could find for
// O2-646.
auto cleanTimers = [&currentSetOfInputs](TimesliceSlot slot, InputRecord& record) {
auto cleanTimers = [&currentSetOfInputs, &ownedInputs](TimesliceSlot slot, InputRecord& record) {
assert(record.size() == currentSetOfInputs.size());
for (size_t ii = 0, ie = record.size(); ii < ie; ++ii) {
// assuming that for timer inputs we do have exactly one PartRef object
Expand All @@ -2265,8 +2269,10 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
if (input.header == nullptr) {
continue;
}
// This will hopefully delete the message.
currentSetOfInputs[ii].clear();
// For the consume=false (Process) path, ownedInputs holds the actual
// message vectors and the span points into them.
ownedInputs[ii].clear();
currentSetOfInputs[ii] = {};
}
};

Expand Down
5 changes: 2 additions & 3 deletions Framework/Core/src/DataProcessingHelpers.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,15 +393,14 @@ void DataProcessingHelpers::cleanForwardedMessages(std::span<fair::mq::MessagePt
}

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

for (size_t ii = 0, ie = currentSetOfInputs.size(); ii < ie; ++ii) {
auto span = std::span<fair::mq::MessagePtr>(currentSetOfInputs[ii]);
routeForwardedMessages(proxy, span, forwardedParts, copyByDefault, consume);
routeForwardedMessages(proxy, currentSetOfInputs[ii], forwardedParts, copyByDefault, consume);
}
return forwardedParts;
};
Expand Down
6 changes: 5 additions & 1 deletion Framework/Core/src/DataRelayer.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,7 +434,11 @@ void DataRelayer::pruneCache(TimesliceSlot slot, OnDropCallback onDrop)
if (anyDropped) {
O2_SIGNPOST_ID_GENERATE(aid, data_relayer);
O2_SIGNPOST_EVENT_EMIT(data_relayer, aid, "pruneCache", "Dropping stuff from slot %zu with timeslice %zu", slot.index, oldestPossibleTimeslice.timeslice.value);
onDrop(slot, dropped, oldestPossibleTimeslice);
std::vector<std::span<fair::mq::MessagePtr>> droppedSpans(dropped.size());
for (size_t ai = 0, ae = dropped.size(); ai != ae; ++ai) {
droppedSpans[ai] = dropped[ai];
}
onDrop(slot, droppedSpans, oldestPossibleTimeslice);
}
}
assert(cache.empty() == false);
Expand Down
48 changes: 36 additions & 12 deletions Framework/Core/test/test_ForwardInputs.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,18 @@
O2_DECLARE_DYNAMIC_LOG(forwarding);
using namespace o2::framework;

// Build a vector of spans over an existing vector-of-vectors for tests that
// construct currentSetOfInputs locally (rather than via consumeAllInputsForTimeslice).
static std::vector<std::span<fair::mq::MessagePtr>> asSpans(std::vector<std::vector<fair::mq::MessagePtr>>& vecs)
{
std::vector<std::span<fair::mq::MessagePtr>> spans;
spans.reserve(vecs.size());
for (auto& v : vecs) {
spans.emplace_back(v);
}
return spans;
}

TEST_CASE("ForwardInputsEmpty")
{
o2::header::DataHeader dh;
Expand All @@ -45,7 +57,8 @@ TEST_CASE("ForwardInputsEmpty")

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

auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, currentSetOfInputs, copyByDefault, consume);
auto spans = asSpans(currentSetOfInputs);
auto result = o2::framework::DataProcessingHelpers::routeForwardedMessageSet(proxy, spans, copyByDefault, consume);
REQUIRE(result.empty());
}

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

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

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

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

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

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

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

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

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

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

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

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