Skip to content

TPC: make IDC/CMV distribute a pure forwarding device + minor improvements #15583

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 5 commits into
base: dev
Choose a base branch
from
Open

TPC: make IDC/CMV distribute a pure forwarding device + minor improvements #15583

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 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
e30a9ae
TPC: replace two-copy forwarding by shallow-copy forwarding in IDC/CM…
ehellbar Jul 6, 2026
3bcb8a4
TPC: adjust log severities and messages, add additional logs to IDC/C…
ehellbar Jul 6, 2026
cb23625
TPC CMV: add test workflow to create dummy CMV data
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 Detectors/TPC/workflow/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,11 @@ o2_add_executable(idc-test-ft
SOURCES test/test_ft_EPN_Aggregator.cxx
PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)

o2_add_executable(cmv-test-generator
COMPONENT_NAME tpc
SOURCES test/test_cmv_generator.cxx
PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)

o2_add_executable(miptrack-filter
COMPONENT_NAME tpc
SOURCES src/tpc-miptrack-filter.cxx
Expand Down
80 changes: 51 additions & 29 deletions Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
// check which buffer to use for current incoming data
const bool currentBuffer = (tf > mTFEnd[mBuffer]) ? !mBuffer : mBuffer;
if (mTFStart[currentBuffer] > tf) {
LOGP(detail, "All CRUs for current TF {} already received. Skipping this TF", tf);
LOGP(warning, "Current TF {} is older than start of currentBuffer {}. Skipping this TF", tf, mTFStart[currentBuffer]);
return;
}

Expand All @@ -166,6 +166,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
}

if (mProcessedCRU[currentBuffer][relTF] == mCRUs.size()) {
LOGP(warning, "All CRUs for current TF {} (relTF {}) already received. Skipping this TF", tf, relTF);
return;
}

Expand All @@ -182,25 +183,27 @@ class TPCDistributeCMVSpec : public o2::framework::Task

forwardOrbitInfo(pc, currentBuffer, relTF, currentOutLane);

for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mFilter)) {
auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
for (auto it = inputs.begin(); it != inputs.end(); ++it) {
auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
const unsigned int cru = tpcCRUHeader->subSpecification >> 7;

// check if cru is specified in input cru list
if (!(std::binary_search(mCRUs.begin(), mCRUs.end(), cru))) {
if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
LOGP(debug, "Received data from CRU: {} which was not specified as input. Skipping", cru);
continue;
}

if (mProcessedCRUs[currentBuffer][relTF][cru]) {
LOGP(warning, "CRU {} for current TF {} (relTF {}) already processed. Skipping ...", cru, tf, relTF);
continue;
}
// count total number of processed CRUs for given TF
++mProcessedCRU[currentBuffer][relTF];
// to keep track of processed CRUs
mProcessedCRUs[currentBuffer][relTF][cru] = true;

sendOutput(pc, currentOutLane, cru, pc.inputs().get<o2::pmr::vector<uint16_t>>(ref));
forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); });
}

LOGP(detail, "Number of received CRUs for current TF: {} Needed a total number of processed CRUs of: {} Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf);
Expand Down Expand Up @@ -247,7 +250,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
const unsigned int mTimeFrames{}; ///< number of TFs per aggregation interval
const int mNTFsBuffer{1}; ///< number of TFs for which the CMVs will be buffered (must match TPCFLPCMVSpec)
const unsigned int mOutLanes{}; ///< number of parallel aggregate pipelines this distributor feeds
std::array<unsigned int, 2> mProcessedTFs{{0, 0}}; ///< number of processed timeframes per buffer; triggers sendOutput when it reaches mTimeFrames
std::array<unsigned int, 2> mProcessedTFs{{0, 0}}; ///< number of processed timeframes per buffer; triggers finishInterval when it reaches mTimeFrames
std::array<std::vector<unsigned int>, 2> mProcessedCRU{}; ///< counter of received CRUs per (buffer, relTF); used to detect when a relTF is complete
std::array<std::vector<std::unordered_map<unsigned int, bool>>, 2> mProcessedCRUs{}; ///< per-CRU received flag ([buffer][relTF][CRU]); prevents double-counting when a CRU re-sends
std::array<long, 2> mTFStart{}; ///< absolute TF counter of the first TF in each buffer interval
Expand All @@ -274,14 +277,26 @@ class TPCDistributeCMVSpec : public o2::framework::Task
/// Returns the total number of real TFs per buffer interval (= mNTFsBuffer * mTimeFrames)
unsigned int getNRealTFs() const { return mNTFsBuffer * mTimeFrames; }

void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector<uint16_t> cmvs)
void sendEmptyCMVOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru)
{
pc.outputs().adoptContainer(o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, o2::pmr::vector<uint16_t>());
}

void sendEmptyOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane)
{
pc.outputs().adoptContainer(o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(cmvs));
pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[outLane], header::DataHeader::SubSpecificationType{outLane}}, static_cast<uint64_t>(0));
}

void sendOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane, const uint64_t orbitInfo)
template <typename EmptyFn>
void forwardData(o2::framework::ProcessingContext& pc, o2::framework::Output outSpec, const unsigned int cru, auto& it, EmptyFn&& sendEmptyData)
{
pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[outLane], header::DataHeader::SubSpecificationType{outLane}}, orbitInfo);
if (auto* payloadMsg = it.getPayload()) {
pc.outputs().forwardPayload(outSpec, *payloadMsg);
} else [[unlikely]] {
// this should never happen
LOGP(warning, "No payload for CRU {}; sending empty {} payload", cru, outSpec.description.as<std::string>());
sendEmptyData();
}
}

void forwardOrbitInfo(o2::framework::ProcessingContext& pc, const bool currentBuffer, const unsigned int relTF, const unsigned int currentOutLane)
Expand All @@ -290,14 +305,15 @@ class TPCDistributeCMVSpec : public o2::framework::Task
return;
}

for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter)) {
auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
const unsigned int cru = hdr->subSpecification >> 7;
auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter);
for (auto it = inputs.begin(); it != inputs.end(); ++it) {
auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
continue;
}

sendOrbitInfo(pc, currentOutLane, pc.inputs().get<uint64_t>(ref));
forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{currentOutLane}}, cru, it, [&] { sendEmptyOrbitInfo(pc, currentOutLane); });
mOrbitInfoForwarded[currentBuffer][relTF] = true;
break;
}
Expand All @@ -319,24 +335,30 @@ class TPCDistributeCMVSpec : public o2::framework::Task
}

if (!mOrbitInfoForwarded[mBuffer].empty()) {
for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter)) {
auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
const unsigned int cru = hdr->subSpecification >> 7;
auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter);
for (auto it = inputs.begin(); it != inputs.end(); ++it) {
auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
continue;
}
sendOrbitInfo(pc, currentOutLane, pc.inputs().get<uint64_t>(ref));

forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{currentOutLane}}, cru, it, [&] { sendEmptyOrbitInfo(pc, currentOutLane); });
break;
}
}

for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mFilter)) {
auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
const unsigned int cru = hdr->subSpecification >> 7;
auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
for (auto it = inputs.begin(); it != inputs.end(); ++it) {
auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
const unsigned int cru = tpcCRUHeader->subSpecification >> 7;

// check if cru is specified in input cru list
if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
continue;
}
sendOutput(pc, currentOutLane, cru, pc.inputs().get<o2::pmr::vector<uint16_t>>(ref));

forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); });
}
}

Expand Down Expand Up @@ -368,15 +390,15 @@ class TPCDistributeCMVSpec : public o2::framework::Task

// if the last buffer has a smaller time range than expected, flush its remaining uncompleted TFs
if ((mTFStart[currentBuffer] > mTFStart[!currentBuffer]) && (relTF > mNTFsDataDrop)) {
LOGP(warning, "Checking last buffer from {} to {}", mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size());
LOGP(warning, "Checking last buffer from relTF {} to {}", mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size());
const unsigned int lastLane = (currentOutLane == 0) ? (mOutLanes - 1) : (currentOutLane - 1);
checkMissingData(pc, !currentBuffer, mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size(), lastLane);
LOGP(detail, "All empty TFs for TF {} for current buffer filled with dummy and sent. Clearing buffer", tf);
LOGP(warning, "All empty TFs of last buffer [{}, {}] filled with dummy and sent, triggered by data from TF {} (relTF {}). Clearing buffer", mTFStart[!currentBuffer], mTFEnd[!currentBuffer], tf, relTF);
finishInterval(pc, lastLane, !currentBuffer, tf);
}

const int tfEndCheck = std::clamp(static_cast<int>(relTF) - mNTFsDataDrop, 0, static_cast<int>(mProcessedCRU[currentBuffer].size()));
LOGP(detail, "Checking current buffer from {} to {}", mStartNTFsDataDrop[currentBuffer], tfEndCheck);
LOGP(detail, "Checking current buffer from relTF {} to {}", mStartNTFsDataDrop[currentBuffer], tfEndCheck);
checkMissingData(pc, currentBuffer, mStartNTFsDataDrop[currentBuffer], tfEndCheck, currentOutLane);
mStartNTFsDataDrop[currentBuffer] = tfEndCheck;
}
Expand All @@ -386,21 +408,21 @@ class TPCDistributeCMVSpec : public o2::framework::Task
{
for (int iTF = startTF; iTF < endTF; ++iTF) {
if (mProcessedCRU[currentBuffer][iTF] != mCRUs.size()) {
LOGP(warning, "CRUs for lane {} rel. TF: {} curr TF {} are missing! Processed {} CRUs out of {}", outLane, iTF, mTFStart[currentBuffer] + static_cast<long>(iTF) * mNTFsBuffer, mProcessedCRU[currentBuffer][iTF], mCRUs.size());
LOGP(warning, "CRUs for lane {} rel. TF: {} curr TF {} are missing! Processed {} CRUs out of {}", outLane, iTF, mTFStart[currentBuffer] + static_cast<long>(iTF) * mNTFsBuffer + mNTFsBuffer - 1, mProcessedCRU[currentBuffer][iTF], mCRUs.size());
++mProcessedTFs[currentBuffer];
mProcessedCRU[currentBuffer][iTF] = mCRUs.size();

// send empty payloads for missing CRUs so the aggregate lane sees a complete set
for (auto& it : mProcessedCRUs[currentBuffer][iTF]) {
if (!it.second) {
it.second = true;
sendOutput(pc, outLane, it.first, o2::pmr::vector<uint16_t>());
sendEmptyCMVOutput(pc, outLane, it.first);
}
}

// send zero orbit placeholder for missing TF so the aggregate lane can still reconstruct timing
if (!mOrbitInfoForwarded[currentBuffer][iTF]) {
sendOrbitInfo(pc, outLane, 0);
sendEmptyOrbitInfo(pc, outLane);
mOrbitInfoForwarded[currentBuffer][iTF] = true;
}
}
Expand All @@ -420,7 +442,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
}
}

LOGP(detail, "All TFs {} for current buffer received. Clearing buffer", tf);
LOGP(info, "All TFs for buffer [{}, {}] received at data from TF {}. Clearing buffer", mTFStart[buffer], mTFEnd[buffer], tf);
clearBuffer(buffer);
mStartNTFsDataDrop[buffer] = 0;
mSendOutputStartInfo[buffer] = true;
Expand Down
Loading