From 52f047d53108441711377801a3b6053e0a085f91 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 27 Aug 2026 15:50:29 +0200 Subject: [PATCH 1/2] Track skipped read timeframes in file statistics, dump --- .../src/AODJAlienReaderHelpers.cxx | 7 +++--- .../AnalysisSupport/src/DataInputDirector.cxx | 22 +++++++++++++++++-- .../AnalysisSupport/src/DataInputDirector.h | 3 +++ Framework/Core/src/runDataProcessing.cxx | 1 + 4 files changed, 28 insertions(+), 5 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 18401554c4b4d..8ae1febe6ef66 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -268,10 +268,11 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const int64_t startTime = uv_hrtime(); int64_t startSize = totalSizeCompressed; - auto skipInvalidRead = [&](o2::header::DataOrigin const& origin, InvalidAODReadError const& e) { + auto skipInvalidRead = [&](header::DataHeader const& dh, InvalidAODReadError const& e) { auto skippedTimeframes = ++totalInvalidReadSkipped; LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}", - origin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); + dh.dataOrigin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); + didir->markTimeFrameSkipped(dh, ntf); arrowContext.clear(); messageContext.discard(); stringContext.clear(); @@ -334,7 +335,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const if (!skipInvalidReads) { throw; } - skipInvalidRead(concrete.origin, e); + skipInvalidRead(dh, e); return TFReaderState::INVALID_TIMEFRAME; } diff --git a/Framework/AnalysisSupport/src/DataInputDirector.cxx b/Framework/AnalysisSupport/src/DataInputDirector.cxx index 9bc90e915d469..abcb2092c0ab5 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.cxx +++ b/Framework/AnalysisSupport/src/DataInputDirector.cxx @@ -288,6 +288,15 @@ arrow::dataset::FileSource DataInputDescriptor::getFileFolder(int counter, int n return {fmt::format("DF_{}", mfilenames[counter].listOfTimeFrameNumbers[numTF]), mCurrentFilesystem}; } +uint64_t DataInputDescriptor::markTimeFrameSkipped(int numTF) +{ + if (mCurrentFileID >= 0 && numTF >= 0 && numTF < mfilenames[mCurrentFileID].numberOfTimeFrames) { + mfilenames[mCurrentFileID].alreadyRead[numTF] = false; + return ++mfilenames[mCurrentFileID].invalidReadSkipped; + } + return 0; +} + std::shared_ptr DataInputDescriptor::getParentFile(int counter, int numTF, std::string treename, int wantedParentLevel, std::string_view wantedOrigin) { if (!mParentFileMap) { @@ -364,8 +373,8 @@ void DataInputDescriptor::printFileStatistics() } auto rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); auto f = dynamic_cast(rootFS->GetFile()); - std::string monitoringInfo(fmt::format("lfn={},size={},total_df={},read_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}", f->GetName(), - f->GetSize(), getTimeFramesInFile(mCurrentFileID), getReadTimeFramesInFile(mCurrentFileID), f->GetBytesRead(), f->GetReadCalls(), + std::string monitoringInfo(fmt::format("lfn={},size={},total_df={},read_df={},skipped_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}", f->GetName(), + f->GetSize(), getTimeFramesInFile(mCurrentFileID), getReadTimeFramesInFile(mCurrentFileID), mfilenames.at(mCurrentFileID).invalidReadSkipped, f->GetBytesRead(), f->GetReadCalls(), ((float)mIOTime / 1e9), ((float)wait_time / 1e9), mLevel)); #if __has_include() auto alienFile = dynamic_cast(f); @@ -879,6 +888,15 @@ arrow::dataset::FileSource DataInputDirector::getFileFolder(header::DataHeader d return didesc->getFileFolder(counter, numTF, wantedLevel, origin); } +void DataInputDirector::markTimeFrameSkipped(header::DataHeader dh, int numTF) +{ + auto didesc = getDataInputDescriptor(dh); + if (!didesc) { + didesc = mdefaultDataInputDescriptor.get(); + } + didesc->markTimeFrameSkipped(numTF); +} + int DataInputDirector::getTimeFramesInFile(header::DataHeader dh, int counter) { auto didesc = getDataInputDescriptor(dh); diff --git a/Framework/AnalysisSupport/src/DataInputDirector.h b/Framework/AnalysisSupport/src/DataInputDirector.h index a810619530e10..374f9f7e89e6f 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.h +++ b/Framework/AnalysisSupport/src/DataInputDirector.h @@ -44,6 +44,7 @@ struct FileNameHolder { int numberOfTimeFrames = 0; std::vector listOfTimeFrameNumbers; std::vector alreadyRead; + uint64_t invalidReadSkipped = 0; }; FileNameHolder makeFileNameHolder(std::string fileName); @@ -106,6 +107,7 @@ class DataInputDescriptor uint64_t getTimeFrameNumber(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin); arrow::dataset::FileSource getFileFolder(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin); + uint64_t markTimeFrameSkipped(int numTF); // Open the current file to populate the parent map, then return the parent descriptor and // the TF index within it that corresponds to numTF at this level. Returns {nullptr, -1} on failure. std::pair, int> navigateToLevel(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin); @@ -171,6 +173,7 @@ class DataInputDirector bool readTree(DataAllocator& outputs, header::DataHeader dh, int counter, int numTF, size_t& totalSizeCompressed, size_t& totalSizeUncompressed, bool wasAOD); uint64_t getTimeFrameNumber(header::DataHeader dh, int counter, int numTF); arrow::dataset::FileSource getFileFolder(header::DataHeader dh, int counter, int numTF); + void markTimeFrameSkipped(header::DataHeader dh, int numTF); int getTimeFramesInFile(header::DataHeader dh, int counter); uint64_t getTotalSizeCompressed(); diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx index 256ae10a9c17e..8cfbfddcc4067 100644 --- a/Framework/Core/src/runDataProcessing.cxx +++ b/Framework/Core/src/runDataProcessing.cxx @@ -1253,6 +1253,7 @@ std::vector getDumpableMetrics() dumpableMetrics.emplace_back("^aod-bytes-read-compressed$"); dumpableMetrics.emplace_back("^aod-file-read-info$"); dumpableMetrics.emplace_back("^aod-largest-object-written$"); + dumpableMetrics.emplace_back("^aod-invalid-read-skipped-timeframes$"); dumpableMetrics.emplace_back("^table-bytes-.*"); dumpableMetrics.emplace_back("^total-timeframes.*"); dumpableMetrics.emplace_back("^device_state.*"); From 4e47b191925dff888cc5ce8beb6b4ae5cfe2746f Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Fri, 28 Aug 2026 13:40:47 +0200 Subject: [PATCH 2/2] Use concrete when recording skipped timeframe --- Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 8ae1febe6ef66..31e2aedf44418 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -268,11 +268,11 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const int64_t startTime = uv_hrtime(); int64_t startSize = totalSizeCompressed; - auto skipInvalidRead = [&](header::DataHeader const& dh, InvalidAODReadError const& e) { + auto skipInvalidRead = [&](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) { auto skippedTimeframes = ++totalInvalidReadSkipped; LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}", - dh.dataOrigin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); - didir->markTimeFrameSkipped(dh, ntf); + concrete.origin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); + didir->markTimeFrameSkipped(header::DataHeader(concrete.description, concrete.origin, concrete.subSpec), ntf); arrowContext.clear(); messageContext.discard(); stringContext.clear(); @@ -335,7 +335,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const if (!skipInvalidReads) { throw; } - skipInvalidRead(dh, e); + skipInvalidRead(concrete, e); return TFReaderState::INVALID_TIMEFRAME; }