diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 18401554c4b4d..31e2aedf44418 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 = [&](ConcreteDataMatcher const& concrete, 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)); + 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(); @@ -334,7 +335,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const if (!skipInvalidReads) { throw; } - skipInvalidRead(concrete.origin, e); + skipInvalidRead(concrete, 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.*");