From 4b89ce93642fcd0ba2e10f9f56e9a8aa230ab411 Mon Sep 17 00:00:00 2001 From: Ernst Hellbar Date: Mon, 6 Jul 2026 09:01:39 +0200 Subject: [PATCH 1/2] DPL: add allocator to forward a fair::mq::Message payload by shallow copy --- .../Core/include/Framework/DataAllocator.h | 5 +++++ Framework/Core/src/DataAllocator.cxx | 20 +++++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/Framework/Core/include/Framework/DataAllocator.h b/Framework/Core/include/Framework/DataAllocator.h index 104e418cbcd19..89572fc86ca2c 100644 --- a/Framework/Core/include/Framework/DataAllocator.h +++ b/Framework/Core/include/Framework/DataAllocator.h @@ -455,6 +455,11 @@ class DataAllocator void snapshot(const Output& spec, const char* payload, size_t payloadSize, o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone); + /// create a shallow copy of the @a inputPayload and forward it to the output route of @a spec + /// if the transport types of the input and output routes are different, a real copy is created as fallback + void forwardPayload(const Output& spec, fair::mq::Message& inputPayload, + o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone); + /// make an object of type T and route to output specified by OutputRef /// The object is owned by the framework, returned reference can be used to fill the object. /// diff --git a/Framework/Core/src/DataAllocator.cxx b/Framework/Core/src/DataAllocator.cxx index 0802bb8300ae7..e899e5b364804 100644 --- a/Framework/Core/src/DataAllocator.cxx +++ b/Framework/Core/src/DataAllocator.cxx @@ -342,6 +342,26 @@ void DataAllocator::snapshot(const Output& spec, const char* payload, size_t pay addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod); } +void DataAllocator::forwardPayload(const Output& spec, fair::mq::Message& inputPayload, + o2::header::SerializationMethod serializationMethod) +{ + auto& proxy = mRegistry.get(); + auto& timingInfo = mRegistry.get(); + + RouteIndex routeIndex = matchDataHeader(spec, timingInfo.timeslice); + auto* transport = proxy.getOutputTransport(routeIndex); + + if (inputPayload.GetTransport() == transport) { + auto payloadMessage = transport->CreateMessage(); + payloadMessage->Copy(inputPayload); + addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod); + } else { + auto payloadMessage = transport->CreateMessage(inputPayload.GetSize(), fair::mq::Alignment{64}); + memcpy(payloadMessage->GetData(), inputPayload.GetData(), inputPayload.GetSize()); + addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod); + } +} + Output DataAllocator::getOutputByBind(OutputRef&& ref) { if (ref.label.empty()) { From 6ed485004ac1cda341a57f5d652a5fb75a98a4c7 Mon Sep 17 00:00:00 2001 From: ALICE Action Bot Date: Mon, 6 Jul 2026 09:17:56 +0000 Subject: [PATCH 2/2] Please consider the following formatting changes --- Framework/Core/src/DataAllocator.cxx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Framework/Core/src/DataAllocator.cxx b/Framework/Core/src/DataAllocator.cxx index e899e5b364804..92f7a7f8a6e8f 100644 --- a/Framework/Core/src/DataAllocator.cxx +++ b/Framework/Core/src/DataAllocator.cxx @@ -343,7 +343,7 @@ void DataAllocator::snapshot(const Output& spec, const char* payload, size_t pay } void DataAllocator::forwardPayload(const Output& spec, fair::mq::Message& inputPayload, - o2::header::SerializationMethod serializationMethod) + o2::header::SerializationMethod serializationMethod) { auto& proxy = mRegistry.get(); auto& timingInfo = mRegistry.get();