Skip to content

DPL: add allocator to forward a fair::mq::Message payload by shallow copy #15577

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

Open
ehellbar wants to merge 2 commits into
base: dev
Choose a base branch
from
+92 −6
Open

DPL: add allocator to forward a fair::mq::Message payload by shallow copy #15577

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b4bf818
DPL: add fairmq message pointer getters to InputRecord
ehellbar Jul 6, 2026
b4570d8
DPL: add allocator to forward a fair::mq::Message payload by shallow …
ehellbar Jul 6, 2026
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
5 changes: 5 additions & 0 deletions Framework/Core/include/Framework/DataAllocator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down
4 changes: 4 additions & 0 deletions Framework/Core/include/Framework/InputRecord.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,9 @@ class InputRecord
/// O(1) access to the part described by @a indices in slot @a pos.
[[nodiscard]] DataRef getAtIndices(int pos, DataRefIndices indices) const;

/// Return the payload as fair::mq::Message* for the part described by @a indices in slot @a slotIdx
fair::mq::Message* getPayloadAtIndices(size_t slotIdx, DataRefIndices indices) const;

/// O(1) advance from @a current to the next part's indices in slot @a pos.
[[nodiscard]] DataRefIndices nextIndices(int pos, DataRefIndices current) const
{
Expand Down Expand Up @@ -745,6 +748,7 @@ class InputRecord
[[nodiscard]] DataRefIndices initialIndices() const { return {0, 1}; }
[[nodiscard]] DataRefIndices endIndices() const { return {size_t(-1), size_t(-1)}; }
[[nodiscard]] DataRef getAtIndices(DataRefIndices idx) const { return record->getAtIndices((int)slot, idx); }
[[nodiscard]] fair::mq::Message* getPayloadAtIndices(DataRefIndices idx) const { return record->getPayloadAtIndices((int)slot, idx); }

@ktf ktf Jul 7, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

don't we already have an helper which does that?

@ehellbar ehellbar Jul 7, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't see it, at least not for what I am doing here.

I define the getter e.g. in https://github.com/AliceO2Group/AliceO2/pull/15577/changes#diff-5d24f98b5c6098b6ad0a1ab814450dec590c82b86147d85c5a34afa0521171a6R239, where I use the DataRefIndices directly since we anyway calculate them in the InputRecordIterator.

There is https://github.com/AliceO2Group/AliceO2/blob/dev/Framework/Core/include/Framework/DataModelViews.h#L209, but it uses some indices from which the vector indices need to be computed first, so it is more expensive than what I need to do. Or are you referring to something else?

[[nodiscard]] DataRefIndices nextIndices(DataRefIndices idx) const { return record->nextIndices((int)slot, idx); }
[[nodiscard]] size_t size() const { return record->getNofParts((int)slot); }

Expand Down
5 changes: 5 additions & 0 deletions Framework/Core/include/Framework/InputRecordWalker.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,11 @@ class InputRecordWalker
return not operator==(rh);
}

fair::mq::Message* getPayload() const
{
return mCurrentRange.getPayloadAtIndices(mCurrent.indices());
}

private:
bool next(bool isInitialPart = false)
{
Expand Down
17 changes: 17 additions & 0 deletions Framework/Core/include/Framework/InputSpan.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,11 @@

#include "Framework/DataRef.h"
#include <functional>
#include <fairmq/FwdDecls.h>

extern template class std::function<o2::framework::DataRef(size_t, o2::framework::DataRefIndices)>;
extern template class std::function<o2::framework::DataRefIndices(size_t, o2::framework::DataRefIndices)>;
extern template class std::function<fair::mq::Message*(size_t, o2::framework::DataRefIndices)>;

namespace o2::framework
{
Expand All @@ -38,6 +40,7 @@ class InputSpan
std::function<int(size_t)> refCountGetter,
std::function<DataRef(size_t, DataRefIndices)> indicesGetter,
std::function<DataRefIndices(size_t, DataRefIndices)> nextIndicesGetter,
std::function<fair::mq::Message*(size_t, DataRefIndices)> payloadGetter,
size_t size);

/// @a i-th element of the InputSpan (O(partidx) sequential scan via indices protocol)
Expand All @@ -56,6 +59,12 @@ class InputSpan
return mIndicesGetter(slotIdx, indices);
}

/// Return the payload as fair::mq::Message* for the part described by @a indices in slot @a slotIdx
[[nodiscard]] fair::mq::Message* getPayloadAtIndices(size_t slotIdx, DataRefIndices indices) const
{
return mPayloadGetter(slotIdx, indices);
}

/// Advance from @a current to the indices of the next part in slot @a slotIdx in O(1).
[[nodiscard]] DataRefIndices nextIndices(size_t slotIdx, DataRefIndices current) const
{
Expand Down Expand Up @@ -179,6 +188,12 @@ class InputSpan
return mCurrentIndices.headerIdx;
}

// return current indices
[[nodiscard]] DataRefIndices indices() const
{
return mCurrentIndices;
}

// return an iterable range over all parts in the current slot
// only available for slot-level iterators whose parent has parts(size_t)
[[nodiscard]] auto parts() const
Expand All @@ -201,6 +216,7 @@ class InputSpan
[[nodiscard]] DataRefIndices initialIndices() const { return {0, 1}; }
[[nodiscard]] DataRefIndices endIndices() const { return {size_t(-1), size_t(-1)}; }
[[nodiscard]] DataRef getAtIndices(DataRefIndices idx) const { return span->getAtIndices(slot, idx); }
[[nodiscard]] fair::mq::Message* getPayloadAtIndices(DataRefIndices idx) const { return span->getPayloadAtIndices(slot, idx); }
[[nodiscard]] DataRefIndices nextIndices(DataRefIndices idx) const { return span->nextIndices(slot, idx); }
[[nodiscard]] size_t size() const { return span->getNofParts(slot); }

Expand Down Expand Up @@ -230,6 +246,7 @@ class InputSpan
std::function<int(size_t)> mRefCountGetter;
std::function<DataRef(size_t, DataRefIndices)> mIndicesGetter;
std::function<DataRefIndices(size_t, DataRefIndices)> mNextIndicesGetter;
std::function<fair::mq::Message*(size_t, DataRefIndices)> mPayloadGetter;
size_t mSize;
};

Expand Down
20 changes: 20 additions & 0 deletions Framework/Core/src/DataAllocator.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<FairMQDeviceProxy>();
auto& timingInfo = mRegistry.get<TimingInfo>();

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()) {
Expand Down
9 changes: 8 additions & 1 deletion Framework/Core/src/DataProcessingDevice.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2203,7 +2203,14 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
auto next = currentSetOfInputs[i] | get_next_pair{current};
return next.headerIdx < currentSetOfInputs[i].size() ? next : DataRefIndices{size_t(-1), size_t(-1)};
};
return InputSpan{nofPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, currentSetOfInputs.size()};
auto payloadGetter = [&currentSetOfInputs](size_t i, DataRefIndices current) -> fair::mq::Message* {
auto const& msgs = currentSetOfInputs[i];
if (msgs.size() <= current.payloadIdx || !msgs[current.payloadIdx]) {
return nullptr;
}
return msgs[current.payloadIdx].get();
};
return InputSpan{nofPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, payloadGetter, currentSetOfInputs.size()};
};

auto markInputsAsDone = [ref](TimesliceSlot slot) -> void {
Expand Down
18 changes: 16 additions & 2 deletions Framework/Core/src/DataRelayer.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,14 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
auto next = partial[idx] | get_next_pair{current};
return next.headerIdx < partial[idx].size() ? next : DataRefIndices{size_t(-1), size_t(-1)};
};
InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, static_cast<size_t>(partial.size())};
auto payloadGetter = [&partial](size_t idx, DataRefIndices current) -> fair::mq::Message* {
auto const& msgs = partial[idx];
if (msgs.size() <= current.payloadIdx || !msgs[current.payloadIdx]) {
return nullptr;
}
return msgs[current.payloadIdx].get();
};
InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, payloadGetter, static_cast<size_t>(partial.size())};
// Setup the input span

if (expirator.checker(services, timestamp.value, span) == false) {
Expand Down Expand Up @@ -818,7 +825,14 @@ void DataRelayer::getReadyToProcess(std::vector<DataRelayer::RecordAction>& comp
auto next = partial[idx] | get_next_pair{current};
return next.headerIdx < partial[idx].size() ? next : DataRefIndices{size_t(-1), size_t(-1)};
};
InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, static_cast<size_t>(partial.size())};
auto payloadGetter = [&partial](size_t idx, DataRefIndices current) -> fair::mq::Message* {
auto const& msgs = partial[idx];
if (msgs.size() <= current.payloadIdx || !msgs[current.payloadIdx]) {
return nullptr;
}
return msgs[current.payloadIdx].get();
};
InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, payloadGetter, static_cast<size_t>(partial.size())};
CompletionPolicy::CompletionOp action = mCompletionPolicy.callbackFull(span, mInputs, mContext);

auto& variables = mTimesliceIndex.getVariablesForSlot(slot);
Expand Down
5 changes: 5 additions & 0 deletions Framework/Core/src/InputRecord.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,11 @@ DataRef InputRecord::getAtIndices(int pos, DataRefIndices indices) const
return ref;
}

fair::mq::Message* InputRecord::getPayloadAtIndices(size_t slotIdx, DataRefIndices indices) const
{
return mSpan.getPayloadAtIndices(slotIdx, indices);
}

size_t InputRecord::size() const
{
return mSpan.size();
Expand Down
4 changes: 3 additions & 1 deletion Framework/Core/src/InputSpan.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,17 @@

template class std::function<o2::framework::DataRef(size_t, o2::framework::DataRefIndices)>;
template class std::function<o2::framework::DataRefIndices(size_t, o2::framework::DataRefIndices)>;
template class std::function<fair::mq::Message*(size_t, o2::framework::DataRefIndices)>;

namespace o2::framework
{
InputSpan::InputSpan(std::function<size_t(size_t)> nofPartsGetter,
std::function<int(size_t)> refCountGetter,
std::function<DataRef(size_t, DataRefIndices)> indicesGetter,
std::function<DataRefIndices(size_t, DataRefIndices)> nextIndicesGetter,
std::function<fair::mq::Message*(size_t, DataRefIndices)> payloadGetter,
size_t size)
: mNofPartsGetter{nofPartsGetter}, mRefCountGetter(refCountGetter), mIndicesGetter{std::move(indicesGetter)}, mNextIndicesGetter{std::move(nextIndicesGetter)}, mSize{size}
: mNofPartsGetter{nofPartsGetter}, mRefCountGetter(refCountGetter), mIndicesGetter{std::move(indicesGetter)}, mNextIndicesGetter{std::move(nextIndicesGetter)}, mPayloadGetter{std::move(payloadGetter)}, mSize{size}
{
}

Expand Down
2 changes: 2 additions & 0 deletions Framework/Core/test/benchmark_InputRecord.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ static void BM_InputRecordGenericGetters(benchmark::State& state)
nullptr,
[](size_t, DataRefIndices) { return DataRef{nullptr, nullptr, nullptr}; },
[](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
nullptr,
0};
ServiceRegistry registry;
InputRecord emptyRecord(schema, span, registry);
Expand Down Expand Up @@ -92,6 +93,7 @@ static void BM_InputRecordGenericGetters(benchmark::State& state)
nullptr,
[&inputs](size_t i, DataRefIndices idx) { return DataRef{nullptr, static_cast<char const*>(inputs[2 * i + idx.headerIdx]), static_cast<char const*>(inputs[2 * i + idx.payloadIdx])}; },
[](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
nullptr,
inputs.size() / 2};
InputRecord record{schema, span2, registry};

Expand Down
1 change: 1 addition & 0 deletions Framework/Core/test/test_CompletionPolicy.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ TEST_CASE("TestCompletionPolicy_callback")
nullptr,
[&ref](size_t, DataRefIndices) -> DataRef { return ref; },
[](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
nullptr,
1};
std::vector<InputSpec> specs;
ServiceRegistryRef servicesRef{services};
Expand Down
2 changes: 2 additions & 0 deletions Framework/Core/test/test_InputRecord.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ TEST_CASE("TestInputRecord")
nullptr,
[](size_t, DataRefIndices) { return DataRef{nullptr, nullptr, nullptr}; },
[](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
nullptr,
0};
ServiceRegistry registry;
InputRecord emptyRecord(schema, span, registry);
Expand Down Expand Up @@ -99,6 +100,7 @@ TEST_CASE("TestInputRecord")
nullptr,
[&inputs](size_t i, DataRefIndices idx) { return DataRef{nullptr, static_cast<char const*>(inputs[2 * i + idx.headerIdx]), static_cast<char const*>(inputs[2 * i + idx.payloadIdx])}; },
[](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
nullptr,
inputs.size() / 2};
InputRecord record{schema, span2, registry};

Expand Down
2 changes: 1 addition & 1 deletion Framework/Core/test/test_InputRecordWalker.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ struct DataSet {
auto payload = static_cast<char const*>(this->messages[i].second.at(idx.payloadIdx)->data());
return DataRef{nullptr, header, payload}; }, [this](size_t i, DataRefIndices current) -> DataRefIndices {
size_t next = current.headerIdx + 2;
return next < this->messages[i].second.size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)}; }, this->messages.size()}, record{schema, span, registry}, values{std::move(v)}
return next < this->messages[i].second.size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)}; }, nullptr, this->messages.size()}, record{schema, span, registry}, values{std::move(v)}
{
REQUIRE(messages.size() == schema.size());
}
Expand Down
2 changes: 1 addition & 1 deletion Framework/Core/test/test_InputSpan.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ TEST_CASE("TestInputSpan")
return next < inputs[i].size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)};
};

InputSpan span{nPartsGetter, nullptr, indicesGetter, nextIndicesGetter, inputs.size()};
InputSpan span{nPartsGetter, nullptr, indicesGetter, nextIndicesGetter, nullptr, inputs.size()};
REQUIRE(span.size() == inputs.size());
routeNo = 0;
for (; routeNo < span.size(); ++routeNo) {
Expand Down
1 change: 1 addition & 0 deletions Framework/Utils/test/RawPageTestData.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ struct DataSet {
size_t next = current.headerIdx + 2;
return next < this->messages[i].size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)};
},
nullptr,
this->messages.size()},
record{schema, span, registry},
values{std::move(v)}
Expand Down
1 change: 1 addition & 0 deletions Framework/Utils/test/test_RootTreeWriter.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,7 @@ TEST_CASE("test_RootTreeWriter")
return DataRef{nullptr, static_cast<char const*>(store[2 * i + idx.headerIdx]->GetData()), static_cast<char const*>(store[2 * i + idx.payloadIdx]->GetData())};
},
[](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
nullptr,
store.size() / 2};
ServiceRegistry registry;
InputRecord inputs{
Expand Down