From 2618d2ef1e03d4aa969be28daf4779e8edcb77cb Mon Sep 17 00:00:00 2001 From: Matthias Richter Date: Tue, 9 Nov 2021 11:28:26 +0100 Subject: [PATCH] DPL: Adjusting the InputSpan backend InputRecord is providing all payloads as header payload pairs, even if there are multiple payloads per header. The InputSpan backend is using a specific API of messageSet to access all payloads in header-payload pairs This is following the original O2 data model. As an extension, the API allows to access also the payloads of a sequence with multiple payloads per header in header-payload pairs. --- Framework/Core/include/Framework/MessageSet.h | 88 ++++++++++++++----- Framework/Core/src/DataProcessingDevice.cxx | 24 +++-- 2 files changed, 84 insertions(+), 28 deletions(-) diff --git a/Framework/Core/include/Framework/MessageSet.h b/Framework/Core/include/Framework/MessageSet.h index c41bb46bfaefb..7c28181d6d9bb 100644 --- a/Framework/Core/include/Framework/MessageSet.h +++ b/Framework/Core/include/Framework/MessageSet.h @@ -21,30 +21,44 @@ namespace o2 namespace framework { -/// A set of associated inflight messages. +/// A set of inflight messages. +/// The messages are stored in a linear vector. Originally, an O2 message was +/// comprised of a header-payload pair which makes indexing of pairs in the +/// storage simple. To support O2 messages with multiple payloads in a future +/// update of the data model, a message index is needed to store position in the +/// linear storage and number of messages. +/// DPL InputRecord API is providing refs of header-payload pairs, the original +/// O2 message model. For this purpose, also the pair index is filled and can +/// be used to access header and payload associated with a pair struct MessageSet { struct Index { Index(size_t p, size_t s) : position(p), size(s) {} size_t position = 0; size_t size = 0; }; + // linear storage of messages std::vector messages; - std::vector index; + // message map descibes O2 messages consisting of a header message and + // payload message(s), index descibes position in the linear storage + std::vector messageMap; + // pair map describes all messages in header-payload pairs and where + // in the message index the an associated header and payload can be found + std::vector> pairMap; MessageSet() - : messages(), index() + : messages(), messageMap(), pairMap() { } template MessageSet(F getter, size_t size) - : messages(), index() + : messages(), messageMap(), pairMap() { add(std::forward(getter), size); } MessageSet(MessageSet&& other) - : messages(std::move(other.messages)), index(std::move(other.index)) + : messages(std::move(other.messages)), messageMap(std::move(other.messageMap)), pairMap(std::move(other.pairMap)) { other.clear(); } @@ -55,72 +69,106 @@ struct MessageSet { return *this; } messages = std::move(other.messages); - index = std::move(other.index); + messageMap = std::move(other.messageMap); + pairMap = std::move(other.pairMap); other.clear(); return *this; } + /// get number of in-flight O2 messages size_t size() const { - return index.size(); + return messageMap.size(); } - size_t getNumberOfPayloads(size_t part) const + /// get number of header-payload pairs + size_t getNumberOfPairs() const { - return index[part].size; + return pairMap.size(); } + /// get number of payloads for an in-flight message + size_t getNumberOfPayloads(size_t mi) const + { + return messageMap[mi].size; + } + + /// clear the set void clear() { messages.clear(); - index.clear(); + messageMap.clear(); + pairMap.clear(); } // this is more or less legacy + // PartRef has been earlier used to store fixed header-payload pairs + // reset the set and store content of the part ref void reset(PartRef&& ref) { clear(); add(std::move(ref)); } + // this is more or less legacy + // PartRef has been earlier used to store fixed header-payload pairs + // add content of the part ref void add(PartRef&& ref) { - index.emplace_back(messages.size(), 1); + pairMap.emplace_back(messageMap.size(), 0); + messageMap.emplace_back(messages.size(), 1); messages.emplace_back(std::move(ref.header)); messages.emplace_back(std::move(ref.payload)); } + /// add an O2 message template void add(F getter, size_t size) { - index.emplace_back(messages.size(), size - 1); + auto partid = messageMap.size(); + messageMap.emplace_back(messages.size(), size - 1); for (size_t i = 0; i < size; ++i) { + if (i > 0) { + pairMap.emplace_back(partid, i - 1); + } messages.emplace_back(std::move(getter(i))); } } FairMQMessagePtr& header(size_t partIndex) { - return messages[index[partIndex].position]; + return messages[messageMap[partIndex].position]; } FairMQMessagePtr& payload(size_t partIndex, size_t payloadIndex = 0) { - assert(partIndex < index.size()); - assert(index[partIndex].position + payloadIndex + 1 < messages.size()); - return messages[index[partIndex].position + payloadIndex + 1]; + assert(partIndex < messageMap.size()); + assert(messageMap[partIndex].position + payloadIndex + 1 < messages.size()); + return messages[messageMap[partIndex].position + payloadIndex + 1]; } FairMQMessagePtr const& header(size_t partIndex) const { - return messages[index[partIndex].position]; + return messages[messageMap[partIndex].position]; } FairMQMessagePtr const& payload(size_t partIndex, size_t payloadIndex = 0) const { - assert(partIndex < index.size()); - assert(index[partIndex].position + payloadIndex + 1 < messages.size()); - return messages[index[partIndex].position + payloadIndex + 1]; + assert(partIndex < messageMap.size()); + assert(messageMap[partIndex].position + payloadIndex + 1 < messages.size()); + return messages[messageMap[partIndex].position + payloadIndex + 1]; + } + + FairMQMessagePtr const& associatedHeader(size_t pos) const + { + return messages[messageMap[pairMap[pos].first].position]; + } + + FairMQMessagePtr const& associatedPayload(size_t pos) const + { + auto partIndex = pairMap[pos].first; + auto payloadIndex = pairMap[pos].second; + return messages[messageMap[partIndex].position + payloadIndex + 1]; } }; diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index 4df50f4500da0..771207636d813 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -1163,18 +1163,26 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context, currentSetOfInputs = relayer->consumeExistingInputsForTimeslice(slot); } auto getter = [¤tSetOfInputs](size_t i, size_t partindex) -> DataRef { - if (currentSetOfInputs[i].size() > partindex) { - auto header = currentSetOfInputs[i].header(partindex).get(); - auto payload = currentSetOfInputs[i].payload(partindex).get(); - return DataRef{nullptr, - static_cast(header->GetData()), - static_cast(payload ? payload->GetData() : nullptr), - payload ? payload->GetSize() : 0}; + if (currentSetOfInputs[i].getNumberOfPairs() > partindex) { + const char* headerptr = nullptr; + const char* payloadptr = nullptr; + size_t payloadSize = 0; + // - each input can have multiple parts + // - "part" denotes a sequence of messages belonging together, the first message of the + // sequence is the header message + // - each part has one or more payload messages + // - InputRecord provides all payloads as header-payload pairs + auto const& headerMsg = currentSetOfInputs[i].associatedHeader(partindex); + auto const& payloadMsg = currentSetOfInputs[i].associatedPayload(partindex); + headerptr = static_cast(headerMsg->GetData()); + payloadptr = payloadMsg ? static_cast(payloadMsg->GetData()) : nullptr; + payloadSize = payloadMsg ? payloadMsg->GetSize() : 0; + return DataRef{nullptr, headerptr, payloadptr, payloadSize}; } return DataRef{}; }; auto nofPartsGetter = [¤tSetOfInputs](size_t i) -> size_t { - return currentSetOfInputs[i].size(); + return currentSetOfInputs[i].getNumberOfPairs(); }; return InputSpan{getter, nofPartsGetter, currentSetOfInputs.size()}; };