diff --git a/Framework/Core/include/Framework/DataRelayer.h b/Framework/Core/include/Framework/DataRelayer.h index 012b909096317..1e010fc12f3d4 100644 --- a/Framework/Core/include/Framework/DataRelayer.h +++ b/Framework/Core/include/Framework/DataRelayer.h @@ -114,6 +114,9 @@ class DataRelayer 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&)>; + /// Prune all the pending entries in the cache. void prunePending(OnDropCallback); /// Prune the cache for a given slot @@ -135,6 +138,7 @@ class DataRelayer InputInfo const& info, size_t nMessages, size_t nPayloads = 1, + OnInsertionCallback onInsertion = nullptr, OnDropCallback onDrop = nullptr); /// This is to set the oldest possible @a timeslice this relayer can diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index 3925359b056b2..1bea6ec48922a 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -1854,11 +1854,56 @@ void DataProcessingDevice::handleData(ServiceRegistryRef ref, InputChannelInfo& VariableContextHelpers::getTimeslice(variables); forwardInputs(ref, slot, dropped, oldestOutputInfo, false, true); }; + + auto onInsertion = [](ServiceRegistryRef& ref, std::span& messages) { + O2_SIGNPOST_ID_GENERATE(sid, forwarding); + + auto& spec = ref.get(); + bool hasForwards = spec.forwards.empty() == false; + auto& context = ref.get(); + if (context.canForwardEarly && hasForwards) { + O2_SIGNPOST_EVENT_EMIT(device, sid, "device", "Early forwardinding before injecting data into relayer."); + auto& timesliceIndex = ref.get(); + auto oldestTimeslice = timesliceIndex.getOldestPossibleOutput(); + + auto& proxy = ref.get(); + + O2_SIGNPOST_ID_GENERATE(sid, forwarding); + O2_SIGNPOST_START(forwarding, sid, "forwardInputs", "Starting forwarding for incoming messages with oldestTimeslice %zu with copy", + oldestTimeslice.timeslice.value); + std::vector forwardedParts; + forwardedParts.resize(proxy.getNumForwards()); + DataProcessingHelpers::routeForwardedMessages(proxy, messages, forwardedParts, true, false); + + for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) { + if (forwardedParts[fi].Size() == 0) { + continue; + } + ForwardChannelInfo info = proxy.getForwardChannelInfo(ChannelIndex{fi}); + auto& parts = forwardedParts[fi]; + if (info.policy == nullptr) { + O2_SIGNPOST_EVENT_EMIT_ERROR(forwarding, sid, "forwardInputs", "Forwarding to %{public}s %d has no policy.", info.name.c_str(), fi); + continue; + } + O2_SIGNPOST_EVENT_EMIT(forwarding, sid, "forwardInputs", "Forwarding to %{public}s %d", info.name.c_str(), fi); + info.policy->forward(parts, ChannelIndex{fi}, ref); + } + auto& asyncQueue = ref.get(); + auto& decongestion = ref.get(); + O2_SIGNPOST_ID_GENERATE(aid, async_queue); + O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputs", "Queuing forwarding oldestPossible %zu", oldestTimeslice.timeslice.value); + AsyncQueueHelpers::post(asyncQueue, AsyncTask{.timeslice = oldestTimeslice.timeslice, .id = decongestion.oldestPossibleTimesliceTask, .debounce = -1, .callback = decongestionCallbackLate} + .user({.ref = ref, .oldestTimeslice = oldestTimeslice})); + O2_SIGNPOST_END(forwarding, sid, "forwardInputs", "Forwarding done"); + } + }; + auto relayed = relayer.relay(parts.At(headerIndex)->GetData(), &parts.At(headerIndex), input, nMessages, nPayloadsPerHeader, + onInsertion, onDrop); switch (relayed.type) { case DataRelayer::RelayChoice::Type::Backpressured: @@ -2273,9 +2318,13 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v bool consumeSomething = action.op == CompletionPolicy::CompletionOp::Consume || action.op == CompletionPolicy::CompletionOp::ConsumeExisting; if (context.canForwardEarly && hasForwards && consumeSomething) { - O2_SIGNPOST_EVENT_EMIT(device, aid, "device", "Early forwainding: %{public}s.", fmt::format("{}", action.op).c_str()); - auto& timesliceIndex = ref.get(); - forwardInputs(ref, action.slot, currentSetOfInputs, timesliceIndex.getOldestPossibleOutput(), true, action.op == CompletionPolicy::CompletionOp::Consume); + // We used to do fowarding here, however we now do it much earlier. + // We still need to clean the inputs which were already consumed + // via ConsumeExisting and which still have an header to hold the slot. + // FIXME: do we? This should really happen when we do the forwarding on + // insertion, because otherwise we lose the relevant information on how to + // navigate the set of headers. We could actually rely on the messageset index, + // is that the right thing to do though? } markInputsAsDone(action.slot); diff --git a/Framework/Core/src/DataRelayer.cxx b/Framework/Core/src/DataRelayer.cxx index 01e7a2b29fd35..ea2c4c0b73316 100644 --- a/Framework/Core/src/DataRelayer.cxx +++ b/Framework/Core/src/DataRelayer.cxx @@ -436,7 +436,8 @@ DataRelayer::RelayChoice InputInfo const& info, size_t nMessages, size_t nPayloads, - std::function&, TimesliceIndex::OldestOutputInfo)> onDrop) + OnInsertionCallback onInsertion, + OnDropCallback onDrop) { std::scoped_lock lock(mMutex); DataProcessingHeader const* dph = o2::header::get(rawHeader); @@ -482,6 +483,7 @@ DataRelayer::RelayChoice &messages, &nMessages, &nPayloads, + &onInsertion, &cache = mCache, &services = mContext, numInputTypes = mDistinctRoutesIndex.size()](TimesliceId timeslice, int input, TimesliceSlot slot, InputInfo const& info) -> size_t { @@ -512,7 +514,11 @@ DataRelayer::RelayChoice mi += nPayloads; continue; } - target.add([&messages, &mi](size_t i) -> fair::mq::MessagePtr& { return messages[mi + i]; }, nPayloads + 1); + auto span = std::span(messages + mi, messages + mi + nPayloads + 1); + if (onInsertion) { + onInsertion(services, span); + } + target.add([&span](size_t i) -> fair::mq::MessagePtr& { return span[i]; }, nPayloads + 1); mi += nPayloads; saved += nPayloads; }