diff --git a/Framework/Core/include/Framework/DataProcessingHelpers.h b/Framework/Core/include/Framework/DataProcessingHelpers.h index f414e3aa4ae00..e19474447ed12 100644 --- a/Framework/Core/include/Framework/DataProcessingHelpers.h +++ b/Framework/Core/include/Framework/DataProcessingHelpers.h @@ -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 routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector>& currentSetOfInputs, + static std::vector routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector>& currentSetOfInputs, bool copy, bool consume); /// Helper to route messages for forwarding static void routeForwardedMessages(FairMQDeviceProxy& proxy, std::span& currentSetOfInputs, std::vector& forwardedParts, diff --git a/Framework/Core/include/Framework/DataRelayer.h b/Framework/Core/include/Framework/DataRelayer.h index b56a2cb59ff10..710653731cb3f 100644 --- a/Framework/Core/include/Framework/DataRelayer.h +++ b/Framework/Core/include/Framework/DataRelayer.h @@ -24,6 +24,7 @@ #include #include +#include #include #include @@ -113,7 +114,7 @@ class DataRelayer ActivityStats processDanglingInputs(std::vector const&, ServiceRegistryRef context, bool createNew); - using OnDropCallback = std::function>&, TimesliceIndex::OldestOutputInfo info)>; + using OnDropCallback = std::function>&, TimesliceIndex::OldestOutputInfo info)>; // Callback for when some messages are about to be owned by the the DataRelayer using OnInsertionCallback = std::function&)>; diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index ab9a66f4e45d3..bfd38b2948c77 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -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>& currentSetOfInputs, +static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector>& currentSetOfInputs, TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume = true) { auto& proxy = registry.get(); @@ -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>& currentSetOfInputs, +static auto cleanEarlyForward = [](ServiceRegistryRef registry, TimesliceSlot slot, std::vector>& currentSetOfInputs, TimesliceIndex::OldestOutputInfo oldestTimeslice, bool copy, bool consume = true) { auto& proxy = registry.get(); @@ -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(currentSetOfInputs[ii]); - DataProcessingHelpers::cleanForwardedMessages(span, consume); + DataProcessingHelpers::cleanForwardedMessages(currentSetOfInputs[ii], consume); } O2_SIGNPOST_END(forwarding, sid, "forwardInputs", "Cleaning done"); @@ -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 = [®istry = mServiceRegistry, lid](TimesliceSlot slot, std::vector>& dropped, TimesliceIndex::OldestOutputInfo oldestOutputInfo) { + auto onDrop = [®istry = mServiceRegistry, lid](TimesliceSlot slot, std::vector>& 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(); @@ -1985,7 +1984,7 @@ void DataProcessingDevice::handleData(ServiceRegistryRef ref, InputChannelInfo& nPayloadsPerHeader = 1; ii += (nMessages / 2) - 1; } - auto onDrop = [ref](TimesliceSlot slot, std::vector>& dropped, TimesliceIndex::OldestOutputInfo oldestOutputInfo) { + auto onDrop = [ref](TimesliceSlot slot, std::vector>& 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); @@ -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> currentSetOfInputs; + std::vector> currentSetOfInputs; + std::vector> ownedInputs; // - auto getInputSpan = [ref, ¤tSetOfInputs](TimesliceSlot slot, bool consume = true) { + auto getInputSpan = [ref, ¤tSetOfInputs, &ownedInputs](TimesliceSlot slot, bool consume = true) { auto& relayer = ref.get(); 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. @@ -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 = [¤tSetOfInputs](TimesliceSlot slot, InputRecord& record) { + auto cleanTimers = [¤tSetOfInputs, &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 @@ -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] = {}; } }; diff --git a/Framework/Core/src/DataProcessingHelpers.cxx b/Framework/Core/src/DataProcessingHelpers.cxx index b8399a4c591e7..a29e7b345c94c 100644 --- a/Framework/Core/src/DataProcessingHelpers.cxx +++ b/Framework/Core/src/DataProcessingHelpers.cxx @@ -393,15 +393,14 @@ void DataProcessingHelpers::cleanForwardedMessages(std::span>& currentSetOfInputs, + std::vector>& currentSetOfInputs, const bool copyByDefault, bool consume) -> std::vector { // we collect all messages per forward in a map and send them together std::vector forwardedParts(proxy.getNumForwardChannels()); for (size_t ii = 0, ie = currentSetOfInputs.size(); ii < ie; ++ii) { - auto span = std::span(currentSetOfInputs[ii]); - routeForwardedMessages(proxy, span, forwardedParts, copyByDefault, consume); + routeForwardedMessages(proxy, currentSetOfInputs[ii], forwardedParts, copyByDefault, consume); } return forwardedParts; }; diff --git a/Framework/Core/src/DataRelayer.cxx b/Framework/Core/src/DataRelayer.cxx index 67c382d413783..38b421e9bcdaf 100644 --- a/Framework/Core/src/DataRelayer.cxx +++ b/Framework/Core/src/DataRelayer.cxx @@ -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> 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); diff --git a/Framework/Core/test/test_ForwardInputs.cxx b/Framework/Core/test/test_ForwardInputs.cxx index 0263158ee0f9b..d0ca4f35d022e 100644 --- a/Framework/Core/test/test_ForwardInputs.cxx +++ b/Framework/Core/test/test_ForwardInputs.cxx @@ -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> asSpans(std::vector>& vecs) +{ + std::vector> spans; + spans.reserve(vecs.size()); + for (auto& v : vecs) { + spans.emplace_back(v); + } + return spans; +} + TEST_CASE("ForwardInputsEmpty") { o2::header::DataHeader dh; @@ -45,7 +57,8 @@ TEST_CASE("ForwardInputsEmpty") std::vector> 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()); } @@ -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 } @@ -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. } @@ -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: @@ -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 @@ -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 @@ -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); // @@ -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); // @@ -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 @@ -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); @@ -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 } @@ -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 }