From d64a0fcf04e61dc4c20dd0293bc4eddd3797af5b Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Wed, 14 Oct 2026 12:35:46 +0200 Subject: [PATCH 1/2] DPL Analysis: skip input files which cannot be opened --- .../src/AODJAlienReaderHelpers.cxx | 31 ++++++++++++++++--- .../AnalysisSupport/src/DataInputDirector.cxx | 2 +- .../AnalysisSupport/src/DataInputDirector.h | 8 +++++ .../include/Framework/DataProcessingStats.h | 1 + Framework/Core/src/CommonServices.cxx | 7 +++++ Framework/Core/src/runDataProcessing.cxx | 1 + 6 files changed, 44 insertions(+), 6 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index e101288bf5b83..65974c6654230 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -260,6 +260,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const static size_t totalSizeCompressed = 0; static uint64_t totalDFSent = 0; static uint64_t totalInvalidReadSkipped = 0; + static uint64_t totalFileOpenSkipped = 0; // check if RuntimeLimit is reached if (!watchdog->update()) { @@ -274,15 +275,18 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const int64_t startTime = uv_hrtime(); int64_t startSize = totalSizeCompressed; - 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: {}", - concrete.origin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); + auto discardTimeframe = [&didir, &arrowContext, &messageContext, &stringContext]() { clean_all_runtime_errors(); didir->finishTimeFrame(true); arrowContext.clear(); messageContext.discard(); stringContext.clear(); + }; + auto skipInvalidRead = [&fcnt, &ntf, &discardTimeframe, &dpstats, &fileCounter, &numTF, &device](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) { + auto skippedTimeframes = ++totalInvalidReadSkipped; + LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}", + concrete.origin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); + discardTimeframe(); dpstats.updateStats({static_cast(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), DataProcessingStats::Op::Add, 1}); *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices; *numTF = ntf; @@ -294,6 +298,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const TRY_NEXT_FILE, TIMEFRAME_READ, INVALID_TIMEFRAME, + INVALID_FILE, }; auto readState = TFReaderState::READ_FIRST_TABLE; [[maybe_unused]] auto stateName = [](TFReaderState state) -> char const* { @@ -310,6 +315,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const return "TIMEFRAME_READ"; case TFReaderState::INVALID_TIMEFRAME: return "INVALID_TIMEFRAME"; + case TFReaderState::INVALID_FILE: + return "INVALID_FILE"; } return "UNKNOWN"; }; @@ -344,7 +351,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const outputs.make(o2) = currentFilename; } }; - auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, skipInvalidReads](TFReaderState currentState) -> TFReaderState { + auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, &discardTimeframe, &dpstats, skipInvalidReads](TFReaderState currentState) -> TFReaderState { while (routeIndex < requestedTables.size() && (device.inputTimesliceId % requestedTables[routeIndex].maxTimeslices) != requestedTables[routeIndex].timeslice) { ++routeIndex; @@ -365,6 +372,16 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { return TFReaderState::TRY_NEXT_FILE; } + } catch (AODFileOpenError const& e) { + if (!skipInvalidReads) { + throw; + } + auto skippedFiles = ++totalFileOpenSkipped; + LOGP(error, "Unable to open AOD input for table {}: fileCounter {}, timeFrame {}. Skipping file (skipped files: {}). Reason: {}", + concrete.origin.as(), fcnt, ntf, skippedFiles, describeException(e)); + discardTimeframe(); + dpstats.updateStats({static_cast(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), DataProcessingStats::Op::Add, 1}); + return TFReaderState::INVALID_FILE; } catch (InvalidAODReadError const& e) { if (!skipInvalidReads) { throw; @@ -394,6 +411,10 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const throw std::runtime_error("Processing is stopped!"); } break; + case TFReaderState::INVALID_FILE: + // Undo the discard so outputs read from the next file are checked as usual + messageContext.clear(); + [[fallthrough]]; case TFReaderState::TRY_NEXT_FILE: didir->finishTimeFrame(); fcnt += device.maxInputTimeslices; diff --git a/Framework/AnalysisSupport/src/DataInputDirector.cxx b/Framework/AnalysisSupport/src/DataInputDirector.cxx index 3c53f8fc0c632..bd36844327d94 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.cxx +++ b/Framework/AnalysisSupport/src/DataInputDirector.cxx @@ -165,7 +165,7 @@ bool DataInputDescriptor::setFile(int counter, int wantedParentLevel, std::strin if (tfile && !externalFile) { delete tfile; } - throw std::runtime_error(fmt::format("Couldn't open file \"{}\"!", filename)); + throw AODFileOpenError(fmt::format("Couldn't open file \"{}\"!", filename)); } mCurrentFilesystem = std::make_shared(tfile, 50 * 1024 * 1024, mFactory, !externalFile); rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); diff --git a/Framework/AnalysisSupport/src/DataInputDirector.h b/Framework/AnalysisSupport/src/DataInputDirector.h index 87977b4dd837a..c0c9827309471 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.h +++ b/Framework/AnalysisSupport/src/DataInputDirector.h @@ -39,6 +39,14 @@ class InvalidAODReadError : public std::runtime_error using std::runtime_error::runtime_error; }; +// Opening an input file can fail before its timeframes are known. The reader +// must advance to the next file rather than retrying subsequent timeframe indices. +class AODFileOpenError : public std::runtime_error +{ + public: + using std::runtime_error::runtime_error; +}; + struct FileNameHolder { std::string fileName; int numberOfTimeFrames = 0; diff --git a/Framework/Core/include/Framework/DataProcessingStats.h b/Framework/Core/include/Framework/DataProcessingStats.h index e164e11cb2134..d5e9ec14adc26 100644 --- a/Framework/Core/include/Framework/DataProcessingStats.h +++ b/Framework/Core/include/Framework/DataProcessingStats.h @@ -75,6 +75,7 @@ enum struct ProcessingStatsId : short { CCDB_CACHE_FETCHED_BYTES, CCDB_CACHE_REQUESTED_BYTES, AOD_INVALID_READ_SKIPPED_TIMEFRAMES, + AOD_FILE_OPEN_SKIPPED_FILES, AVAILABLE_MANAGED_SHM_BASE = 512, }; diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx index c36a102bde80d..751d7849df1c1 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -1127,6 +1127,13 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats() .minPublishInterval = 0, .maxRefreshLatency = 10000, .sendInitialValue = true}, + MetricSpec{.name = "aod-file-open-skipped-files", + .metricId = static_cast(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), + .kind = Kind::UInt64, + .scope = Scope::DPL, + .minPublishInterval = 0, + .maxRefreshLatency = 10000, + .sendInitialValue = true}, MetricSpec{.name = "arrow-bytes-destroyed", .enabled = arrowAndResourceLimitingMetrics, .metricId = static_cast(ProcessingStatsId::ARROW_BYTES_DESTROYED), diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx index 8cfbfddcc4067..ad5fa53fbd271 100644 --- a/Framework/Core/src/runDataProcessing.cxx +++ b/Framework/Core/src/runDataProcessing.cxx @@ -1254,6 +1254,7 @@ std::vector getDumpableMetrics() 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("^aod-file-open-skipped-files$"); dumpableMetrics.emplace_back("^table-bytes-.*"); dumpableMetrics.emplace_back("^total-timeframes.*"); dumpableMetrics.emplace_back("^device_state.*"); From 03f7f11bcf8f5e35b54fa7a5be1c97b66b80eb2f Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 15 Oct 2026 14:18:57 +0200 Subject: [PATCH 2/2] merge AOD read errors into one class with a kind --- .../src/AODJAlienReaderHelpers.cxx | 28 ++++++++++--------- .../AnalysisSupport/src/DataInputDirector.cxx | 20 ++++++------- .../AnalysisSupport/src/DataInputDirector.h | 19 +++++++------ 3 files changed, 35 insertions(+), 32 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 65974c6654230..3a6fe20f1c103 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -282,7 +282,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const messageContext.discard(); stringContext.clear(); }; - auto skipInvalidRead = [&fcnt, &ntf, &discardTimeframe, &dpstats, &fileCounter, &numTF, &device](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) { + auto skipInvalidRead = [&fcnt, &ntf, &discardTimeframe, &dpstats, &fileCounter, &numTF, &device](ConcreteDataMatcher const& concrete, AODReadError const& e) { auto skippedTimeframes = ++totalInvalidReadSkipped; LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}", concrete.origin.as(), fcnt, ntf, skippedTimeframes, describeException(e)); @@ -372,22 +372,24 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { return TFReaderState::TRY_NEXT_FILE; } - } catch (AODFileOpenError const& e) { + } catch (AODReadError const& e) { if (!skipInvalidReads) { throw; } - auto skippedFiles = ++totalFileOpenSkipped; - LOGP(error, "Unable to open AOD input for table {}: fileCounter {}, timeFrame {}. Skipping file (skipped files: {}). Reason: {}", - concrete.origin.as(), fcnt, ntf, skippedFiles, describeException(e)); - discardTimeframe(); - dpstats.updateStats({static_cast(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), DataProcessingStats::Op::Add, 1}); - return TFReaderState::INVALID_FILE; - } catch (InvalidAODReadError const& e) { - if (!skipInvalidReads) { - throw; + switch (e.getKind()) { + case AODReadError::Kind::InvalidTimeframe: + skipInvalidRead(concrete, e); + return TFReaderState::INVALID_TIMEFRAME; + case AODReadError::Kind::FileOpen: { + auto skippedFiles = ++totalFileOpenSkipped; + LOGP(error, "Unable to open AOD input for table {}: fileCounter {}, timeFrame {}. Skipping file (skipped files: {}). Reason: {}", + concrete.origin.as(), fcnt, ntf, skippedFiles, describeException(e)); + discardTimeframe(); + dpstats.updateStats({static_cast(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), DataProcessingStats::Op::Add, 1}); + return TFReaderState::INVALID_FILE; + } } - skipInvalidRead(concrete, e); - return TFReaderState::INVALID_TIMEFRAME; + throw; } if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) { diff --git a/Framework/AnalysisSupport/src/DataInputDirector.cxx b/Framework/AnalysisSupport/src/DataInputDirector.cxx index bd36844327d94..22de54f91c9db 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.cxx +++ b/Framework/AnalysisSupport/src/DataInputDirector.cxx @@ -165,7 +165,7 @@ bool DataInputDescriptor::setFile(int counter, int wantedParentLevel, std::strin if (tfile && !externalFile) { delete tfile; } - throw AODFileOpenError(fmt::format("Couldn't open file \"{}\"!", filename)); + throw AODReadError(AODReadError::Kind::FileOpen, fmt::format("Couldn't open file \"{}\"!", filename)); } mCurrentFilesystem = std::make_shared(tfile, 50 * 1024 * 1024, mFactory, !externalFile); rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); @@ -363,7 +363,7 @@ std::shared_ptr DataInputDescriptor::getParentFile(int coun // The current DF is not found in the parent map (this should not happen and is a fatal error) auto rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); if (!parentFileName) { - throw InvalidAODReadError(fmt::format(R"(parent file map exists but does not contain the current DF "{}" in file "{}")", folderName.c_str(), rootFS->GetFile()->GetName())); + throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(parent file map exists but does not contain the current DF "{}" in file "{}")", folderName.c_str(), rootFS->GetFile()->GetName())); } if (mParentFile) { @@ -392,7 +392,7 @@ std::shared_ptr DataInputDescriptor::getParentFile(int coun } } catch (...) { mParentFile.reset(); - std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName))); + std::throw_with_nested(AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName))); } return mParentFile; } @@ -575,11 +575,11 @@ try { } if (parentFile == nullptr) { auto rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); - throw InvalidAODReadError(fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName())); + throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName())); } if (parentNumTF == -1) { auto parentRootFS = std::dynamic_pointer_cast(parentFile->mCurrentFilesystem); - throw InvalidAODReadError(fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName())); + throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName())); } t.deactivate(); return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed); @@ -634,7 +634,7 @@ try { int parentNumTF = parentFile->findDFNumber(0, folder.path()); if (parentNumTF == -1) { auto parentRootFS = std::dynamic_pointer_cast(parentFile->mCurrentFilesystem); - throw InvalidAODReadError(fmt::format(R"(DF {} listed in parent file map but not found in the corresponding file "{}")", folder.path(), parentRootFS->GetFile()->GetName())); + throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(DF {} listed in parent file map but not found in the corresponding file "{}")", folder.path(), parentRootFS->GetFile()->GetName())); } // first argument is 0 as the parent file object contains only 1 file return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed); @@ -642,7 +642,7 @@ try { auto schemaOpt = format->Inspect(fullpath); if (!schemaOpt.ok()) { - throw InvalidAODReadError(fmt::format("Unable to inspect tree {}: {}", treename, schemaOpt.status().ToString())); + throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to inspect tree {}: {}", treename, schemaOpt.status().ToString())); } auto physicalSchema = schemaOpt; std::vector> fields; @@ -673,16 +673,16 @@ try { f2b.release(); } catch (...) { f2b.discard(); - std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to {} tree {}", operation, treename))); + std::throw_with_nested(AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to {} tree {}", operation, treename))); } return true; -} catch (InvalidAODReadError const&) { +} catch (AODReadError const& e) { auto rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); auto filename = rootFS ? std::string(rootFS->GetFile()->GetName()) : mfilenames.at(counter).fileName; auto const& dfs = mfilenames.at(counter).listOfTimeFrameNumbers; auto df = numTF >= 0 && static_cast(numTF) < dfs.size() ? fmt::format("DF_{}", dfs[numTF]) : fmt::format("timeframe index {}", numTF); - std::throw_with_nested(InvalidAODReadError(fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel))); + std::throw_with_nested(AODReadError(e.getKind(), fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel))); } DataInputDirector::DataInputDirector(std::vector inputFiles, DataInputDirectorContext&& context) diff --git a/Framework/AnalysisSupport/src/DataInputDirector.h b/Framework/AnalysisSupport/src/DataInputDirector.h index c0c9827309471..ef4b7b0269f85 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.h +++ b/Framework/AnalysisSupport/src/DataInputDirector.h @@ -33,18 +33,19 @@ class Monitoring; namespace o2::framework { -class InvalidAODReadError : public std::runtime_error +class AODReadError : public std::runtime_error { public: - using std::runtime_error::runtime_error; -}; + enum class Kind { + InvalidTimeframe, // skip the timeframe + FileOpen, // skip the file, its timeframes are unknown + }; -// Opening an input file can fail before its timeframes are known. The reader -// must advance to the next file rather than retrying subsequent timeframe indices. -class AODFileOpenError : public std::runtime_error -{ - public: - using std::runtime_error::runtime_error; + AODReadError(Kind kind, std::string const& what) : std::runtime_error(what), mKind(kind) {} + Kind getKind() const { return mKind; } + + private: + Kind mKind; }; struct FileNameHolder {