Skip to content

Commit b4570d8

Browse files
committed
DPL: add allocator to forward a fair::mq::Message payload by shallow copy
1 parent b4bf818 commit b4570d8

2 files changed

Lines changed: 25 additions & 0 deletions

File tree

Framework/Core/include/Framework/DataAllocator.h

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -455,6 +455,11 @@ class DataAllocator
455455
void snapshot(const Output& spec, const char* payload, size_t payloadSize,
456456
o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone);
457457

458+
/// create a shallow copy of the @a inputPayload and forward it to the output route of @a spec
459+
/// if the transport types of the input and output routes are different, a real copy is created as fallback
460+
void forwardPayload(const Output& spec, fair::mq::Message& inputPayload,
461+
o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone);
462+
458463
/// make an object of type T and route to output specified by OutputRef
459464
/// The object is owned by the framework, returned reference can be used to fill the object.
460465
///

Framework/Core/src/DataAllocator.cxx

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -342,6 +342,26 @@ void DataAllocator::snapshot(const Output& spec, const char* payload, size_t pay
342342
addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
343343
}
344344

345+
void DataAllocator::forwardPayload(const Output& spec, fair::mq::Message& inputPayload,
346+
o2::header::SerializationMethod serializationMethod)
347+
{
348+
auto& proxy = mRegistry.get<FairMQDeviceProxy>();
349+
auto& timingInfo = mRegistry.get<TimingInfo>();
350+
351+
RouteIndex routeIndex = matchDataHeader(spec, timingInfo.timeslice);
352+
auto* transport = proxy.getOutputTransport(routeIndex);
353+
354+
if (inputPayload.GetTransport() == transport) {
355+
auto payloadMessage = transport->CreateMessage();
356+
payloadMessage->Copy(inputPayload);
357+
addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
358+
} else {
359+
auto payloadMessage = transport->CreateMessage(inputPayload.GetSize(), fair::mq::Alignment{64});
360+
memcpy(payloadMessage->GetData(), inputPayload.GetData(), inputPayload.GetSize());
361+
addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
362+
}
363+
}
364+
345365
Output DataAllocator::getOutputByBind(OutputRef&& ref)
346366
{
347367
if (ref.label.empty()) {

0 commit comments

Comments
 (0)