diff --git a/Detectors/TPC/workflow/CMakeLists.txt b/Detectors/TPC/workflow/CMakeLists.txt index 44bcfd23ebbc2..5429203d20453 100644 --- a/Detectors/TPC/workflow/CMakeLists.txt +++ b/Detectors/TPC/workflow/CMakeLists.txt @@ -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 diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h index af576b2f30a5b..f60506b411667 100644 --- a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h +++ b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h @@ -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; } @@ -158,7 +158,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task LOGP(debug, "Current TF: {}, relative TF: {}, current buffer: {}, current output lane: {}, mTFStart: {}", tf, relTF, currentBuffer, currentOutLane, mTFStart[currentBuffer]); if (relTF >= mProcessedCRU[currentBuffer].size()) { - LOGP(warning, "Skipping tf {}: relative tf {} is larger than size of buffer: {}", tf, relTF, mProcessedCRU[currentBuffer].size()); + LOGP(warning, "Skipping tf {} for lane {}: relative tf {} is larger than size of buffer [{}, {}]: {}", tf, currentOutLane, relTF, mTFStart[currentBuffer], mTFEnd[currentBuffer], mProcessedCRU[currentBuffer].size()); // check number of processed CRUs for previous TFs. If CRUs are missing for them, they are probably lost/not received mProcessedTotalData = mCheckEveryNData; checkIntervalsForMissingData(pc, currentBuffer, relTF, currentOutLane, tf); @@ -166,6 +166,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task } if (mProcessedCRU[currentBuffer][relTF] == mCRUs.size()) { + LOGP(warning, "All CRUs for current TF {} (relTF {}, lane {}) already received. Skipping this TF", tf, relTF, currentOutLane); return; } @@ -182,17 +183,19 @@ 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(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(*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 {}, lane {}) already processed. Skipping ...", cru, tf, relTF, currentOutLane); continue; } // count total number of processed CRUs for given TF @@ -200,7 +203,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task // to keep track of processed CRUs mProcessedCRUs[currentBuffer][relTF][cru] = true; - sendOutput(pc, currentOutLane, cru, pc.inputs().get>(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); @@ -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 mProcessedTFs{{0, 0}}; ///< number of processed timeframes per buffer; triggers sendOutput when it reaches mTimeFrames + std::array mProcessedTFs{{0, 0}}; ///< number of processed timeframes per buffer; triggers finishInterval when it reaches mTimeFrames std::array, 2> mProcessedCRU{}; ///< counter of received CRUs per (buffer, relTF); used to detect when a relTF is complete std::array>, 2> mProcessedCRUs{}; ///< per-CRU received flag ([buffer][relTF][CRU]); prevents double-counting when a CRU re-sends std::array mTFStart{}; ///< absolute TF counter of the first TF in each buffer interval @@ -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 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()); + } + + 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(0)); } - void sendOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane, const uint64_t orbitInfo) + template + 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()); + sendEmptyData(); + } } void forwardOrbitInfo(o2::framework::ProcessingContext& pc, const bool currentBuffer, const unsigned int relTF, const unsigned int currentOutLane) @@ -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(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(*it); + const unsigned int cru = tpcCRUHeader->subSpecification >> 7; if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) { continue; } - sendOrbitInfo(pc, currentOutLane, pc.inputs().get(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; } @@ -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(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(*it); + const unsigned int cru = tpcCRUHeader->subSpecification >> 7; if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) { continue; } - sendOrbitInfo(pc, currentOutLane, pc.inputs().get(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(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(*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>(ref)); + + forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); }); } } @@ -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(relTF) - mNTFsDataDrop, 0, static_cast(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; } @@ -386,7 +408,7 @@ 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(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(iTF) * mNTFsBuffer + mNTFsBuffer - 1, mProcessedCRU[currentBuffer][iTF], mCRUs.size()); ++mProcessedTFs[currentBuffer]; mProcessedCRU[currentBuffer][iTF] = mCRUs.size(); @@ -394,13 +416,13 @@ class TPCDistributeCMVSpec : public o2::framework::Task for (auto& it : mProcessedCRUs[currentBuffer][iTF]) { if (!it.second) { it.second = true; - sendOutput(pc, outLane, it.first, o2::pmr::vector()); + 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; } } @@ -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 [{}, {}] (lane {}) received at data from TF {}. Clearing buffer", mTFStart[buffer], mTFEnd[buffer], currentOutLane, tf); clearBuffer(buffer); mStartNTFsDataDrop[buffer] = 0; mSendOutputStartInfo[buffer] = true; diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h index 6e589cd6c4e8b..388c67ba59eef 100644 --- a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h +++ b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h @@ -136,7 +136,7 @@ class TPCDistributeIDCSpec : 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(info, "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; } @@ -145,7 +145,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task LOGP(debug, "current TF: {} relative TF: {} current buffer: {} current output lane: {} mTFStart: {}", tf, relTF, currentBuffer, currentOutLane, mTFStart[currentBuffer]); if (relTF >= mProcessedCRU[currentBuffer].size()) { - LOGP(warning, "Skipping tf {}: relative tf {} is larger than size of buffer: {}", tf, relTF, mProcessedCRU[currentBuffer].size()); + LOGP(warning, "Skipping tf {} for lane {}: relative tf {} is larger than size of buffer [{}, {}]: {}", tf, currentOutLane, relTF, mTFStart[currentBuffer], mTFEnd[currentBuffer], mProcessedCRU[currentBuffer].size()); // check number of processed CRUs for previous TFs. If CRUs are missing for them, they are probably lost/not received mProcessedTotalData = mCheckEveryNData; @@ -154,6 +154,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task } if (mProcessedCRU[currentBuffer][relTF] == mCRUs.size()) { + LOGP(warning, "All CRUs for current TF {} (relTF {}, lane {}) already received. Skipping this TF", tf, relTF, currentOutLane); return; } @@ -169,17 +170,19 @@ class TPCDistributeIDCSpec : public o2::framework::Task pc.outputs().snapshot(Output{gDataOriginTPC, getDataDescriptionIDCOrbitReset(), header::DataHeader::SubSpecificationType{currentOutLane}}, dataformats::Pair{o2::base::GRPGeomHelper::instance().getOrbitResetTimeMS(), o2::base::GRPGeomHelper::instance().getNHBFPerTF()}); } - for (auto& ref : InputRecordWalker(pc.inputs(), mFilter)) { - auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader(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(*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 {}, lane {}) already processed. Skipping ...", cru, tf, relTF, currentOutLane); continue; } else { // count total number of processed CRUs for given TF @@ -189,8 +192,14 @@ class TPCDistributeIDCSpec : public o2::framework::Task mProcessedCRUs[currentBuffer][relTF][cru] = true; } - // sending IDCs - sendOutput(pc, currentOutLane, cru, pc.inputs().get>(ref)); + // forward payload by shallow copy + if (auto* payloadMsg = it.getPayload()) { + pc.outputs().forwardPayload(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, *payloadMsg); + } else [[unlikely]] { + // this should never happen + LOGP(warning, "No IDCGROUP payload for CRU {} (TF {}, relTF {}); sending empty IDCAGG{} payload", cru, tf, relTF, currentOutLane); + sendEmptyIDCOutput(pc, currentOutLane, cru); + } } LOGP(info, "number of received CRUs for current TF: {} Needed a total number of processed CRUs of: {} Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf); @@ -247,9 +256,9 @@ class TPCDistributeIDCSpec : public o2::framework::Task std::vector mFilter{}; ///< filter for looping over input data std::vector mDataDescrOut{}; - void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector idcs) + void sendEmptyIDCOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru) { - pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(idcs)); + pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, pmr::vector()); } /// returns the output lane to which the data will be send @@ -284,19 +293,19 @@ class TPCDistributeIDCSpec : public o2::framework::Task void checkIntervalsForMissingData(o2::framework::ProcessingContext& pc, const bool currentBuffer, const long relTF, const unsigned int currentOutLane, const uint32_t tf) { if (!(mProcessedTotalData++ % mCheckEveryNData)) { - LOGP(info, "Checking for dropped packages..."); + LOGP(detail, "Checking for dropped packages..."); // if last buffer has smaller time range check the whole last buffer 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(info, "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(relTF) - mNTFsDataDrop, 0, static_cast(mProcessedCRU[currentBuffer].size())); - LOGP(info, "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; } @@ -306,7 +315,7 @@ class TPCDistributeIDCSpec : 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] + iTF, 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(iTF) * mNTFsBuffer + mNTFsBuffer - 1, mProcessedCRU[currentBuffer][iTF], mCRUs.size()); ++mProcessedTFs[currentBuffer]; mProcessedCRU[currentBuffer][iTF] = mCRUs.size(); @@ -314,7 +323,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task for (auto& it : mProcessedCRUs[currentBuffer][iTF]) { if (!it.second) { it.second = true; - sendOutput(pc, outLane, it.first, pmr::vector()); + sendEmptyIDCOutput(pc, outLane, it.first); } } } @@ -334,7 +343,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task } } - LOGP(info, "All TFs {} for current buffer received. Clearing buffer", tf); + LOGP(info, "All TFs for buffer [{}, {}] (lane {}) received at data from TF {}. Clearing buffer", mTFStart[buffer], mTFEnd[buffer], currentOutLane, tf); clearBuffer(buffer); mStartNTFsDataDrop[buffer] = 0; mSendOutputStartInfo[buffer] = true; diff --git a/Detectors/TPC/workflow/test/test_cmv_generator.cxx b/Detectors/TPC/workflow/test/test_cmv_generator.cxx new file mode 100644 index 0000000000000..0a670f8d4c0e1 --- /dev/null +++ b/Detectors/TPC/workflow/test/test_cmv_generator.cxx @@ -0,0 +1,426 @@ +// Copyright 2019-2020 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +/// \file test_cmv_generator.cxx +/// \brief DPL source workflow that generates dummy CMV data for testing the CMV FLP pipeline. +/// +/// Replaces o2-tpc-cmv-to-vector in tests; directly emits CMVVECTOR and CMVORBITS +/// messages per CRU per TF so the workflow can be piped straight into o2-tpc-cmv-flp: +/// +/// o2-tpc-cmv-test-generator --crus 0-359 --timeframes 100 \ +/// | o2-tpc-cmv-flp --crus 0-359 --n-TFs-buffer 10 \ +/// | o2-dpl-output-proxy --dataspec "downstream:TPC/CMVGROUP;downstream:TPC/CMVORBITINFO" ... +/// +/// \author Ernst Hellbar + +#include "Framework/DataProcessorSpec.h" +#include "Framework/Task.h" +#include "Framework/ControlService.h" +#include "Framework/ConfigParamRegistry.h" +#include "Framework/ConfigParamSpec.h" +#include "Framework/Logger.h" +#include "Headers/DataHeader.h" +#include "Algorithm/RangeTokenizer.h" +#include "TPCBase/CRU.h" +#include "DataFormatsTPC/CMV.h" +#include "TPCCalibration/CMVHelper.h" +#include "TPCCalibration/CMVContainer.h" +#include "TPCWorkflow/ProcessingHelpers.h" +#include "CommonUtils/TreeStreamRedirector.h" +#include "CommonUtils/ConfigurableParam.h" +#include "DetectorsRaw/HBFUtilsInitializer.h" +#include "DetectorsRaw/HBFUtils.h" +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace o2::framework; +using o2::header::gDataOriginTPC; + +// ───────────────────────────────────────────────────────────────────────────── +// workflow options +// ───────────────────────────────────────────────────────────────────────────── +void customize(std::vector& workflowOptions) +{ + const std::string cruDefault = "0-" + std::to_string(o2::tpc::CRU::MaxCRU - 1); + std::vector options{ + {"crus", VariantType::String, cruDefault.c_str(), {"List of CRUs, comma-separated ranges, e.g. 0-3,7,9-15"}}, + {"timeframes", VariantType::Int, 100, {"Number of TFs to generate; use -1 to run indefinitely"}}, + {"delay", VariantType::Bool, false, {"Add delay after sending all CRUs"}}, + {"delayTime", VariantType::Int, 1, {"Duration of the global per-TF delay in ms (requires --delay true)"}}, + {"delayEveryN", VariantType::Int, 1, {"Apply the global delay only on average once every N TFs, randomly chosen (1 = every TF, requires --delay true)"}}, + {"delayCRUs", VariantType::String, "", {"CRUs for which to add an extra per-CRU delay before sending, comma-separated ranges"}}, + {"delayTimeCRUs", VariantType::Int, 1, {"Duration of the per-CRU delay in ms (requires --delayCRUs)"}}, + {"dropTFsRandom", VariantType::Int, 0, {"Drop a whole TF randomly: on average one every N TFs (0 = disabled)"}}, + {"dropTFsRange", VariantType::String, "", {"Drop all TFs in this range, e.g. 10-12"}}, + {"tfLength", VariantType::Float, 0.f, {"Minimum wall-clock time between consecutively sent TFs in ms (rate limiter); the generator sleeps if a TF is produced faster than this (0 = disabled)"}}, + {"seed", VariantType::Int, 42, {"RNG seed for CMV value generation"}}, + {"amplitude", VariantType::Float, 5.0f, {"Amplitude of the sinusoidal CMV signal (ADC units); ignored when --input-file is set"}}, + {"noise", VariantType::Float, 1.0f, {"Gaussian noise std-dev added per time bin (ADC units); used as the smearing width in --input-file mode"}}, + {"input-file", VariantType::String, "", {"ROOT file with a CMV 'ccdb_object' tree; the template TF (see --input-entry) is decoded once and re-emitted, smeared per generated TF. Empty = synthetic sinusoidal signal"}}, + {"input-entry", VariantType::Int, 0, {"Tree entry (TF index) used as the template when --input-file is set"}}, + {"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings"}}}; + o2::raw::HBFUtilsInitializer::addConfigOption(options, "hbfutils"); + std::swap(workflowOptions, options); +} + +#include "Framework/runDataProcessing.h" + +// ───────────────────────────────────────────────────────────────────────────── +// generator device +// ───────────────────────────────────────────────────────────────────────────── +class CMVGeneratorDevice : public o2::framework::Task +{ + public: + static constexpr uint32_t sOrbitsPerPacket = 8; ///< each CMV packet covers 8 heartbeat orbits + + CMVGeneratorDevice(const std::vector& crus, + const std::unordered_set& delayCRUs, + unsigned int maxTFs, + bool delay, + int delayTime, + int delayEveryN, + int delayTimeCRUs, + int dropTFsRandom, + const std::vector& rangeTFsDrop, + float tfLength, + float amplitude, + float noise, + int seed, + const std::string& inputFile, + long long inputEntry) + : mCRUs(crus), mDelayCRUs(delayCRUs), mMaxTFs(maxTFs), mDelay(delay), mDelayTime(delayTime), mDelayEveryN(delayEveryN), mDelayTimeCRUs(delayTimeCRUs), mDropTFsRandom(dropTFsRandom), mRangeTFsDrop(rangeTFsDrop), mTFLength(tfLength), mAmplitude(amplitude), mNoise(noise), mRng(static_cast(seed)), mInputFileName(inputFile), mInputEntry(inputEntry) {} + + void init(o2::framework::InitContext& ic) final + { + mTimer100TFs = std::chrono::high_resolution_clock::now(); + mLastTFTime = std::chrono::high_resolution_clock::now(); + + if (!mCRUs.empty()) { + LOGP(info, "crus: {}", fmt::join(mCRUs, ", ")); + } + if (!mDelayCRUs.empty()) { + const std::vector delayCRUsSorted(mDelayCRUs.begin(), mDelayCRUs.end()); + LOGP(info, "delayCRUs: {}", fmt::join(delayCRUsSorted, ", ")); + } + + mWriteDebug = ic.options().get("write-debug"); + if (mWriteDebug) { + mDebugStreamFileName = ic.options().get("debug-file-name"); + LOGP(info, "Creating debug stream {}", mDebugStreamFileName); + mDebugStream = std::make_unique(mDebugStreamFileName.data(), "recreate"); + } + + if (!mInputFileName.empty()) { + o2::tpc::CMVFileHandle handle; + if (!handle.open(mInputFileName)) { + throw std::runtime_error("CMV generator: failed to open input file " + mInputFileName); + } + const auto nEntries = handle.tree->GetEntries(); + if (mInputEntry < 0 || mInputEntry >= nEntries) { + const auto msg = fmt::format("CMV generator: --input-entry {} out of range [0, {}) in {}", mInputEntry, nEntries, mInputFileName); + handle.close(); + throw std::runtime_error(msg); + } + const o2::tpc::CMVPerTF* tmpl = handle.getEntry(mInputEntry); + if (!tmpl) { + handle.close(); + throw std::runtime_error("CMV generator: failed to read/decode entry from " + mInputFileName); + } + // When noise is enabled we keep the per-CRU float template and re-encode it + // (template + noise) every TF. When noise is disabled the output is identical + // for every TF, so we encode it once here and just re-snapshot it in run(). + const bool addNoise = (mNoise > 0.f); + if (addNoise) { + mBaseCMVFloat.resize(mCRUs.size()); + for (size_t iCRU = 0; iCRU < mCRUs.size(); ++iCRU) { + const auto cru = mCRUs[iCRU]; + auto& base = mBaseCMVFloat[iCRU]; + base.resize(o2::tpc::cmv::NTimeBinsPerTF); + for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) { + base[tb] = tmpl->getCMVFloat(static_cast(cru), static_cast(tb)); + } + } + } else { + mBaseCMVEncoded.resize(mCRUs.size()); + for (size_t iCRU = 0; iCRU < mCRUs.size(); ++iCRU) { + const auto cru = mCRUs[iCRU]; + auto& enc = mBaseCMVEncoded[iCRU]; + enc.resize(o2::tpc::cmv::NTimeBinsPerTF); + for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) { + o2::tpc::cmv::Data d; + d.setCMVFloat(tmpl->getCMVFloat(static_cast(cru), static_cast(tb))); + enc[tb] = d.getCMV(); + } + } + } + handle.close(); + mUseInputFile = true; + LOGP(info, "Loaded CMV template from {} (entry {}): {} CRUs x {} bins, noise sigma {} ADC ({})", + mInputFileName, mInputEntry, mCRUs.size(), o2::tpc::cmv::NTimeBinsPerTF, mNoise, + addNoise ? "re-smeared per TF" : "encoded once, replayed verbatim"); + } + } + + void run(o2::framework::ProcessingContext& ctx) final + { + using timer = std::chrono::high_resolution_clock; + const auto tf = o2::tpc::processing_helpers::getCurrentTF(ctx); + + // ── TF dropping ────────────────────────────────────────────────────────── + // Note: RangeTokenizer guarantees sorted output, so front()/back() are min/max. + if (!mRangeTFsDrop.empty() && tf >= static_cast(mRangeTFsDrop.front()) && tf <= static_cast(mRangeTFsDrop.back())) { + LOGP(info, "Dropping TF {} (range drop)", tf); + return; + } + if (mDropTFsRandom > 0 && std::uniform_int_distribution{0, mDropTFsRandom - 1}(mRng) == 0) { + LOGP(info, "Dropping TF {} (random drop)", tf); + return; + } + + auto start = timer::now(); + + // ── CMV values ─────────────────────────────────────────────────────────── + // NTimeBinsPerTF = NPacketsPerTFPerCRU (4) * NTimeBinsPerPacket (3564) = 14256 + // - synthetic mode: shared cmvVec = sinusoidal signal + noise (same for all CRUs) + // - input-file mode: per-CRU template + the shared noise vector + const bool addNoise = (mNoise > 0.f); // skip all RNG when --noise 0 + std::normal_distribution noiseDist{0.f, mNoise}; + std::vector cmvVec(o2::tpc::cmv::NTimeBinsPerTF); + std::vector noiseVec; // only populated in input-file mode when noise is enabled + if (mUseInputFile) { + if (addNoise) { + noiseVec.resize(o2::tpc::cmv::NTimeBinsPerTF); + for (auto& n : noiseVec) { + n = noiseDist(mRng); + } + } + } else { + const float signal = -std::abs(mAmplitude * std::sin(tf * 0.05f)); + for (auto& v : cmvVec) { + o2::tpc::cmv::Data d; + d.setCMVFloat(addNoise ? (signal + noiseDist(mRng)) : signal); + v = d.getCMV(); + } + } + + // ── Orbit / BC info (same for all CRUs) ────────────────────────────────── + // One packed (orbit<<32|bc) entry per CMV packet (4 per TF). + // Each packet covers 8 heartbeat orbits (NTimeBinsPerPacket = 3564 = 8 LHC orbits), + // so the orbit advances by 8 per packet and by NPacketsPerTFPerCRU*8 = 32 per TF. + std::vector orbitBCVec(o2::tpc::cmv::NPacketsPerTFPerCRU); + for (uint32_t pkt = 0; pkt < o2::tpc::cmv::NPacketsPerTFPerCRU; ++pkt) { + const uint32_t orbit = static_cast(tf * o2::tpc::cmv::NPacketsPerTFPerCRU * sOrbitsPerPacket + pkt * sOrbitsPerPacket); + orbitBCVec[pkt] = uint64_t(orbit) << 32; // bc = 0 + } + + for (size_t iCRU = 0; iCRU < mCRUs.size(); ++iCRU) { + const auto cru = mCRUs[iCRU]; + const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7}; + + // ── per-CRU delay ──────────────────────────────────────────────────── + if (mDelayCRUs.count(cru)) { + LOGP(info, "Delaying CRU {} by {} ms (TF {})", cru, mDelayTimeCRUs, tf); + std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTimeCRUs)); + } + + // Select the vector to emit: the precomputed template (no noise) is sent + // verbatim; otherwise this CRU's template is smeared with the shared noise. + std::vector* out = &cmvVec; + if (mUseInputFile && !addNoise) { + out = &mBaseCMVEncoded[iCRU]; // encoded once in init(), reused every TF + } else if (mUseInputFile) { + const auto& base = mBaseCMVFloat[iCRU]; + for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) { + o2::tpc::cmv::Data d; + d.setCMVFloat(base[tb] + noiseVec[tb]); + cmvVec[tb] = d.getCMV(); + } + } + + ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVVECTOR", subSpec}, *out); + ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVORBITS", subSpec}, orbitBCVec); + + if (mWriteDebug) { + auto& stream = (*mDebugStream) << "cmvs"; + stream << "cru=" << cru + << "tfCounter=" << tf + << "nCMVs=" << out->size() + << "cmvs=" << *out + << "\n"; + } + } + + if (!(tf % 100)) { + const auto elapsed100 = std::chrono::duration_cast(timer::now() - mTimer100TFs).count(); + LOGP(info, "Generated CMV data for TF {} ({} ms for last 100 TFs)", tf, elapsed100); + mTimer100TFs = timer::now(); + } + + // ── global delay ───────────────────────────────────────────────────────── + if (mDelay && (mDelayEveryN <= 1 || std::uniform_int_distribution{0, mDelayEveryN - 1}(mRng) == 0)) { + auto elapsed = std::chrono::duration_cast(timer::now() - start).count(); + if (elapsed < mDelayTime) { + LOGP(info, "Delaying TF {} by {} ms", tf, mDelayTime - elapsed); + std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTime - elapsed)); + } + } + + // ── rate limiting ──────────────────────────────────────────────────────── + // Enforce a minimum wall-clock spacing between consecutively sent TFs. + if (mTFLength > 0.f) { + const auto elapsedSinceLast = std::chrono::duration_cast(timer::now() - mLastTFTime).count(); + const auto tfLengthUs = static_cast(mTFLength * 1000.f); + if (elapsedSinceLast < tfLengthUs) { + const auto waitUs = tfLengthUs - elapsedSinceLast; + LOGP(info, "Rate limiting TF {}: waiting {} us (tfLength={} ms)", tf, waitUs, mTFLength); + std::this_thread::sleep_for(std::chrono::microseconds(waitUs)); + } + mLastTFTime = timer::now(); + } + + // endOfStream() propagates the EoS signal to downstream devices (required for source devices). + if (mMaxTFs != std::numeric_limits::max() && tf >= mMaxTFs - 1) { + ctx.services().get().endOfStream(); + ctx.services().get().readyToQuit(QuitRequest::Me); + } + } + + void endOfStream(o2::framework::EndOfStreamContext&) final { closeFiles(); } + void stop() final { closeFiles(); } + + private: + void closeFiles() + { + if (mDebugStream) { + auto& stream = (*mDebugStream) << "cmvs"; + auto& tree = stream.getTree(); + tree.SetAlias("sector", "int(cru/10)"); + mDebugStream->Close(); + mDebugStream.reset(nullptr); + } + } + + const std::vector mCRUs{}; + const std::unordered_set mDelayCRUs{}; + const unsigned int mMaxTFs{}; + const bool mDelay{false}; + const int mDelayTime{1}; + const int mDelayEveryN{1}; + const int mDelayTimeCRUs{1}; + const int mDropTFsRandom{0}; + const std::vector mRangeTFsDrop{}; + const float mTFLength{0.f}; + const float mAmplitude{5.f}; + const float mNoise{1.f}; + std::mt19937 mRng{}; + const std::string mInputFileName{}; ///< CMV ROOT file to use as template ("" = synthetic mode) + const long long mInputEntry{0}; ///< tree entry (TF) used as template + bool mUseInputFile{false}; ///< true once a template has been loaded + std::vector> mBaseCMVFloat; ///< decoded template CMV values [iCRU][timeBin] (noise>0 path), aligned to mCRUs + std::vector> mBaseCMVEncoded; ///< pre-encoded template output [iCRU][timeBin] (noise==0 path), aligned to mCRUs + std::chrono::high_resolution_clock::time_point mTimer100TFs{}; + std::chrono::high_resolution_clock::time_point mLastTFTime{}; + bool mWriteDebug{false}; + std::string mDebugStreamFileName{}; + std::unique_ptr mDebugStream{}; +}; + +// ───────────────────────────────────────────────────────────────────────────── +DataProcessorSpec generateCMVsCRU(const std::vector& crus, + const std::unordered_set& delayCRUs, + unsigned int maxTFs, + bool delay, + int delayTime, + int delayEveryN, + int delayTimeCRUs, + int dropTFsRandom, + const std::vector& rangeTFsDrop, + float tfLength, + float amplitude, + float noise, + int seed, + const std::string& inputFile, + long long inputEntry) +{ + std::vector outputSpecs; + outputSpecs.reserve(crus.size() * 2); + for (const auto cru : crus) { + const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7}; + outputSpecs.emplace_back(gDataOriginTPC, "CMVVECTOR", subSpec, Lifetime::Timeframe); + outputSpecs.emplace_back(gDataOriginTPC, "CMVORBITS", subSpec, Lifetime::Timeframe); + } + + return DataProcessorSpec{ + "tpc-cmv-generator", + Inputs{}, + outputSpecs, + AlgorithmSpec{adaptFromTask(crus, delayCRUs, maxTFs, delay, delayTime, delayEveryN, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, tfLength, amplitude, noise, seed, inputFile, inputEntry)}, + Options{ + {"write-debug", VariantType::Bool, false, {"Write a debug output tree"}}, + {"debug-file-name", VariantType::String, "./cmv_generator_debug.root", {"Name of the debug output file"}}, + }}; +} + +// ───────────────────────────────────────────────────────────────────────────── +WorkflowSpec defineDataProcessing(ConfigContext const& config) +{ + const auto tpcCRUs = o2::RangeTokenizer::tokenize(config.options().get("crus")); + const std::vector crus(tpcCRUs.begin(), tpcCRUs.end()); + + const auto delayCRUsStr = config.options().get("delayCRUs"); + std::unordered_set delayCRUs; + if (!delayCRUsStr.empty()) { + for (const auto cru : o2::RangeTokenizer::tokenize(delayCRUsStr)) { + delayCRUs.insert(static_cast(cru)); + } + } + + const auto dropTFsRangeStr = config.options().get("dropTFsRange"); + const auto rangeTFsDrop = dropTFsRangeStr.empty() ? std::vector{} : o2::RangeTokenizer::tokenize(dropTFsRangeStr); + const int timeframesInt = config.options().get("timeframes"); + // -1 means run indefinitely; map to UINT_MAX so the termination check never fires. + const auto timeframes = (timeframesInt < 0) ? std::numeric_limits::max() : static_cast(timeframesInt); + const auto delay = config.options().get("delay"); + const auto delayTime = config.options().get("delayTime"); + const auto delayEveryN = config.options().get("delayEveryN"); + const auto delayTimeCRUs = config.options().get("delayTimeCRUs"); + const auto dropTFsRandom = config.options().get("dropTFsRandom"); + const auto tfLength = config.options().get("tfLength"); + const auto seed = config.options().get("seed"); + const auto amplitude = config.options().get("amplitude"); + const auto noise = config.options().get("noise"); + const auto inputFile = config.options().get("input-file"); + const auto inputEntry = static_cast(config.options().get("input-entry")); + + o2::conf::ConfigurableParam::updateFromString(config.options().get("configKeyValues")); + + WorkflowSpec workflow; + workflow.emplace_back(generateCMVsCRU(crus, delayCRUs, timeframes, delay, delayTime, delayEveryN, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, tfLength, amplitude, noise, seed, inputFile, inputEntry)); + + auto& hbfu = o2::raw::HBFUtils::Instance(); + long startTime = hbfu.startTime > 0 ? hbfu.startTime : std::chrono::time_point_cast(std::chrono::system_clock::now()).time_since_epoch().count(); + o2::conf::ConfigurableParam::updateFromString(fmt::format("HBFUtils.startTime={}", startTime).data()); + o2::conf::ConfigurableParam::updateFromString(fmt::format("HBFUtils.nHBFPerTF={}", hbfu.nHBFPerTF).data()); + o2::raw::HBFUtilsInitializer hbfIni(config, workflow); + + return workflow; +}