From 14146a3431348b58f47b754c700feb10b4c9bb59 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Mon, 3 Aug 2026 14:41:13 +0200 Subject: [PATCH 1/7] Skip invalid timeframes during ROOT input Handle corrupt reads as recoverable and discard the affected timeframe when DPL_AOD_READER_SKIP_INVALID is enabled. --- .../src/AODJAlienReaderHelpers.cxx | 74 ++++++++++++++----- .../AnalysisSupport/src/DataInputDirector.cxx | 48 +++++++++--- .../AnalysisSupport/src/DataInputDirector.h | 7 ++ 3 files changed, 98 insertions(+), 31 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 1abc4b9ffdd48..21c305d3e0b76 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -11,6 +11,7 @@ #include "AODJAlienReaderHelpers.h" #include +#include #include #include #include @@ -19,6 +20,7 @@ #include "Framework/DataProcessingStats.h" #include "Framework/RootArrowFilesystem.h" #include "Framework/AlgorithmSpec.h" +#include "Framework/ArrowContext.h" #include "Framework/ConfigParamRegistry.h" #include "Framework/ControlService.h" #include "Framework/CallbackService.h" @@ -26,6 +28,8 @@ #include "Framework/DeviceSpec.h" #include "Framework/RawDeviceService.h" #include "Framework/DataSpecUtils.h" +#include "Framework/MessageContext.h" +#include "Framework/StringContext.h" #include "Framework/ConfigContext.h" #include "DataInputDirector.h" #include "Framework/SourceInfoHeader.h" @@ -200,7 +204,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const numTF, watchdog, maxRate, - didir, reportTFN, reportTFFileName, level](Monitoring& monitoring, DataAllocator& outputs, ControlService& control, DeviceSpec const& device, DataProcessingStats& dpstats) { + didir, reportTFN, reportTFFileName, level](Monitoring& monitoring, DataAllocator& outputs, ControlService& control, DeviceSpec const& device, DataProcessingStats& dpstats, ArrowContext& arrowContext, MessageContext& messageContext, StringContext& stringContext) { // Each parallel reader device.inputTimesliceId reads the files fileCounter*device.maxInputTimeslices+device.inputTimesliceId // the TF to read is numTF assert(device.inputTimesliceId < device.maxInputTimeslices); @@ -218,6 +222,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const static size_t totalSizeUncompressed = 0; static size_t totalSizeCompressed = 0; static uint64_t totalDFSent = 0; + static uint64_t totalInvalidReadSkipped = 0; + static bool skipInvalidReads = getenv("DPL_AOD_READER_SKIP_INVALID") && atoi(getenv("DPL_AOD_READER_SKIP_INVALID")); // check if RuntimeLimit is reached if (!watchdog->update()) { @@ -232,6 +238,17 @@ 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 skippedTimeframes = ++totalInvalidReadSkipped; + LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}", + origin.as(), fcnt, ntf, skippedTimeframes, e.what()); + arrowContext.clear(); + messageContext.discard(); + stringContext.clear(); + monitoring.send(Metric{skippedTimeframes, "aod-invalid-read-skipped-timeframes"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL)); + *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices; + *numTF = ntf; + }; for (auto& route : requestedTables) { if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) { continue; @@ -242,25 +259,44 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); - if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - if (first) { - // check if there is a next file to read - fcnt += device.maxInputTimeslices; - if (didir->atEnd(fcnt)) { - LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); - didir->closeInputFiles(); - monitoring.flushBuffer(); - control.endOfStream(); - control.readyToQuit(QuitRequest::Me); - return; - } - // get first folder of next file - ntf = 0; - if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); - throw std::runtime_error("Processing is stopped!"); + bool treeRead = false; + try { + treeRead = didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD); + } catch (InvalidAODReadError const& e) { + if (!skipInvalidReads) { + throw; + } + skipInvalidRead(concrete.origin, e); + return; + } + + if (!treeRead) { + if (!first) { + LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); + throw std::runtime_error("Processing is stopped!"); + } + // check if there is a next file to read + fcnt += device.maxInputTimeslices; + if (didir->atEnd(fcnt)) { + LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); + didir->closeInputFiles(); + monitoring.flushBuffer(); + control.endOfStream(); + control.readyToQuit(QuitRequest::Me); + return; + } + // get first folder of next file + ntf = 0; + try { + treeRead = didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD); + } catch (InvalidAODReadError const& e) { + if (!skipInvalidReads) { + throw; } - } else { + skipInvalidRead(concrete.origin, e); + return; + } + if (!treeRead) { LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); throw std::runtime_error("Processing is stopped!"); } diff --git a/Framework/AnalysisSupport/src/DataInputDirector.cxx b/Framework/AnalysisSupport/src/DataInputDirector.cxx index cfd578862fabd..4431a95f8e0fb 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.cxx +++ b/Framework/AnalysisSupport/src/DataInputDirector.cxx @@ -34,6 +34,7 @@ #include #include #include +#include #include #if __has_include() @@ -536,18 +537,25 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh if (!format) { t.deactivate(); LOGP(debug, "Could not find tree {}. Trying in parent file.", fullpath.path()); - auto parentFile = getParentFile(counter, numTF, treename, wantedLevel, wantedOrigin); - if (parentFile != nullptr) { - int parentNumTF = parentFile->findDFNumber(0, folder.path()); - if (parentNumTF == -1) { - auto parentRootFS = std::dynamic_pointer_cast(parentFile->mCurrentFilesystem); - throw std::runtime_error(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); + std::shared_ptr parentFile; + try { + parentFile = getParentFile(counter, numTF, treename, wantedLevel, wantedOrigin); + } catch (std::exception const& e) { + throw InvalidAODReadError(fmt::format("Unable to resolve parent file for tree {}: {}", treename, e.what())); + } catch (...) { + throw InvalidAODReadError(fmt::format("Unable to resolve parent file for tree {}", treename)); + } + if (parentFile == nullptr) { + auto rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); + throw std::runtime_error(fmt::format(R"(Couldn't get TTree "{}" from "{}". Please check https://aliceo2group.github.io/analysis-framework/docs/troubleshooting/#tree-not-found for more information.)", fullpath.path(), rootFS->GetFile()->GetName())); } - auto rootFS = std::dynamic_pointer_cast(mCurrentFilesystem); - throw std::runtime_error(fmt::format(R"(Couldn't get TTree "{}" from "{}". Please check https://aliceo2group.github.io/analysis-framework/docs/troubleshooting/#tree-not-found for more information.)", fullpath.path(), rootFS->GetFile()->GetName())); + 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())); + } + // first argument is 0 as the parent file object contains only 1 file + return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed); } auto schemaOpt = format->Inspect(fullpath); @@ -573,7 +581,23 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh //// add branches to read //// fill the table f2b->setLabel(treename.c_str()); - f2b->fill(datasetSchema, format); + try { + f2b->fill(datasetSchema, format); + } catch (std::exception const& e) { + f2b.discard(); + throw InvalidAODReadError(fmt::format("Unable to read tree {}: {}", treename, e.what())); + } catch (...) { + f2b.discard(); + throw InvalidAODReadError(fmt::format("Unable to read tree {}", treename)); + } + + try { + f2b.release(); + } catch (std::exception const& e) { + throw InvalidAODReadError(fmt::format("Unable to finalize tree {}: {}", treename, e.what())); + } catch (...) { + throw InvalidAODReadError(fmt::format("Unable to finalize tree {}", treename)); + } return true; } diff --git a/Framework/AnalysisSupport/src/DataInputDirector.h b/Framework/AnalysisSupport/src/DataInputDirector.h index 17535f2935ba3..a810619530e10 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.h +++ b/Framework/AnalysisSupport/src/DataInputDirector.h @@ -21,6 +21,7 @@ #include #include +#include #include #include "rapidjson/fwd.h" @@ -32,6 +33,12 @@ class Monitoring; namespace o2::framework { +class InvalidAODReadError : public std::runtime_error +{ + public: + using std::runtime_error::runtime_error; +}; + struct FileNameHolder { std::string fileName; int numberOfTimeFrames = 0; From 8d1e4b20de60dbb05ec9a401869fb0c012ffc6c2 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 12:31:53 +0200 Subject: [PATCH 2/7] Use DataProcessingStats for skipped metric --- Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx | 2 +- Framework/Core/include/Framework/DataProcessingStats.h | 1 + Framework/Core/src/CommonServices.cxx | 7 +++++++ 3 files changed, 9 insertions(+), 1 deletion(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 21c305d3e0b76..938728f2ba9bf 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -245,7 +245,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const arrowContext.clear(); messageContext.discard(); stringContext.clear(); - monitoring.send(Metric{skippedTimeframes, "aod-invalid-read-skipped-timeframes"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL)); + dpstats.updateStats({static_cast(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), DataProcessingStats::Op::Add, 1}); *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices; *numTF = ntf; }; diff --git a/Framework/Core/include/Framework/DataProcessingStats.h b/Framework/Core/include/Framework/DataProcessingStats.h index edb04c4c5f752..e164e11cb2134 100644 --- a/Framework/Core/include/Framework/DataProcessingStats.h +++ b/Framework/Core/include/Framework/DataProcessingStats.h @@ -74,6 +74,7 @@ enum struct ProcessingStatsId : short { CCDB_CACHE_FAILURE, CCDB_CACHE_FETCHED_BYTES, CCDB_CACHE_REQUESTED_BYTES, + AOD_INVALID_READ_SKIPPED_TIMEFRAMES, AVAILABLE_MANAGED_SHM_BASE = 512, }; diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx index 2ac9dab40d20a..83cdf31833cce 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -1103,6 +1103,13 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats() MetricSpec{.name = "dropped_computations", .metricId = static_cast(ProcessingStatsId::DROPPED_COMPUTATIONS), .kind = Kind::UInt64, .minPublishInterval = quickUpdateInterval}, MetricSpec{.name = "dropped_incoming_messages", .metricId = static_cast(ProcessingStatsId::DROPPED_INCOMING_MESSAGES), .kind = Kind::UInt64, .minPublishInterval = quickUpdateInterval}, MetricSpec{.name = "relayed_messages", .metricId = static_cast(ProcessingStatsId::RELAYED_MESSAGES), .kind = Kind::UInt64, .minPublishInterval = quickUpdateInterval}, + MetricSpec{.name = "aod-invalid-read-skipped-timeframes", + .metricId = static_cast(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), + .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), From 9bad534414e12c2f732498dd4f9102cab5acfb46 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 12:32:39 +0200 Subject: [PATCH 3/7] Better parse skipInvalidReads env var --- Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 938728f2ba9bf..5f86b595701d4 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -223,7 +223,10 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const static size_t totalSizeCompressed = 0; static uint64_t totalDFSent = 0; static uint64_t totalInvalidReadSkipped = 0; - static bool skipInvalidReads = getenv("DPL_AOD_READER_SKIP_INVALID") && atoi(getenv("DPL_AOD_READER_SKIP_INVALID")); + static bool skipInvalidReads = [] { + auto const* envValue = getenv("DPL_AOD_READER_SKIP_INVALID"); + return envValue != nullptr && strcmp(envValue, "0") != 0 && strcmp(envValue, "false") != 0; + }(); // check if RuntimeLimit is reached if (!watchdog->update()) { From f03ffebee5251b6916a8113ceb410b8bbc89d1a1 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 12:39:08 +0200 Subject: [PATCH 4/7] Switch reading booleans to state --- .../src/AODJAlienReaderHelpers.cxx | 69 +++++++++++-------- 1 file changed, 40 insertions(+), 29 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 5f86b595701d4..25ccf669e2eed 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -262,9 +262,18 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); - bool treeRead = false; + enum class ReadState { + READ, + NOT_READ_AND_FIRST, + NOT_READ_AND_MIDDLE, + }; + ReadState readState; try { - treeRead = didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD); + if (didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { + readState = ReadState::READ; + } else { + readState = first ? ReadState::NOT_READ_AND_FIRST : ReadState::NOT_READ_AND_MIDDLE; + } } catch (InvalidAODReadError const& e) { if (!skipInvalidReads) { throw; @@ -273,36 +282,38 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const return; } - if (!treeRead) { - if (!first) { + switch (readState) { + case ReadState::READ: + break; + case ReadState::NOT_READ_AND_MIDDLE: LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); throw std::runtime_error("Processing is stopped!"); - } - // check if there is a next file to read - fcnt += device.maxInputTimeslices; - if (didir->atEnd(fcnt)) { - LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); - didir->closeInputFiles(); - monitoring.flushBuffer(); - control.endOfStream(); - control.readyToQuit(QuitRequest::Me); - return; - } - // get first folder of next file - ntf = 0; - try { - treeRead = didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD); - } catch (InvalidAODReadError const& e) { - if (!skipInvalidReads) { - throw; + case ReadState::NOT_READ_AND_FIRST: + // check if there is a next file to read + fcnt += device.maxInputTimeslices; + if (didir->atEnd(fcnt)) { + LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); + didir->closeInputFiles(); + monitoring.flushBuffer(); + control.endOfStream(); + control.readyToQuit(QuitRequest::Me); + return; } - skipInvalidRead(concrete.origin, e); - return; - } - if (!treeRead) { - LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); - throw std::runtime_error("Processing is stopped!"); - } + // get first folder of next file + ntf = 0; + try { + if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { + LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); + throw std::runtime_error("Processing is stopped!"); + } + } catch (InvalidAODReadError const& e) { + if (!skipInvalidReads) { + throw; + } + skipInvalidRead(concrete.origin, e); + return; + } + break; } if (first) { From a9a11c7ff11ea821b5b44949469a213b20293db7 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 15:21:20 +0200 Subject: [PATCH 5/7] Only checkskipInvalidReads env var once, track state across all loop iterations --- .../src/AODJAlienReaderHelpers.cxx | 36 +++++++++++-------- 1 file changed, 21 insertions(+), 15 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 25ccf669e2eed..e30e6bb2fcfc2 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -197,6 +197,10 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const int level = originLevelMapping.empty() ? -1 : 0; auto fileCounter = std::make_shared(0); auto numTF = std::make_shared(-1); + bool const skipInvalidReads = [] { + auto const* envValue = getenv("DPL_AOD_READER_SKIP_INVALID"); + return envValue != nullptr && strcmp(envValue, "0") != 0 && strcmp(envValue, "false") != 0; + }(); return adaptStateless([TFNumberHeader, TFFileNameHeader, requestedTables, @@ -204,6 +208,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const numTF, watchdog, maxRate, + skipInvalidReads, didir, reportTFN, reportTFFileName, level](Monitoring& monitoring, DataAllocator& outputs, ControlService& control, DeviceSpec const& device, DataProcessingStats& dpstats, ArrowContext& arrowContext, MessageContext& messageContext, StringContext& stringContext) { // Each parallel reader device.inputTimesliceId reads the files fileCounter*device.maxInputTimeslices+device.inputTimesliceId // the TF to read is numTF @@ -218,15 +223,10 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const } // loop over requested tables - bool first = true; static size_t totalSizeUncompressed = 0; static size_t totalSizeCompressed = 0; static uint64_t totalDFSent = 0; static uint64_t totalInvalidReadSkipped = 0; - static bool skipInvalidReads = [] { - auto const* envValue = getenv("DPL_AOD_READER_SKIP_INVALID"); - return envValue != nullptr && strcmp(envValue, "0") != 0 && strcmp(envValue, "false") != 0; - }(); // check if RuntimeLimit is reached if (!watchdog->update()) { @@ -252,6 +252,14 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices; *numTF = ntf; }; + enum class ReadState { + BEFORE_FIRST_READ, + FIRST_READ, + READ, + NOT_READ_AND_FIRST, + NOT_READ_AND_MIDDLE, + }; + auto readState = ReadState::BEFORE_FIRST_READ; for (auto& route : requestedTables) { if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) { continue; @@ -262,17 +270,11 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); - enum class ReadState { - READ, - NOT_READ_AND_FIRST, - NOT_READ_AND_MIDDLE, - }; - ReadState readState; try { if (didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - readState = ReadState::READ; + readState = readState == ReadState::BEFORE_FIRST_READ ? ReadState::FIRST_READ : ReadState::READ; } else { - readState = first ? ReadState::NOT_READ_AND_FIRST : ReadState::NOT_READ_AND_MIDDLE; + readState = readState == ReadState::BEFORE_FIRST_READ ? ReadState::NOT_READ_AND_FIRST : ReadState::NOT_READ_AND_MIDDLE; } } catch (InvalidAODReadError const& e) { if (!skipInvalidReads) { @@ -283,6 +285,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const } switch (readState) { + case ReadState::FIRST_READ: case ReadState::READ: break; case ReadState::NOT_READ_AND_MIDDLE: @@ -313,10 +316,13 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const skipInvalidRead(concrete.origin, e); return; } + readState = ReadState::FIRST_READ; break; + case ReadState::BEFORE_FIRST_READ: + throw std::logic_error("Invalid AOD read state"); } - if (first) { + if (readState == ReadState::FIRST_READ) { if (reportTFN) { // TF number auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf); @@ -339,7 +345,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const outputs.make(o2) = currentFilename; } } - first = false; + readState = ReadState::READ; } int64_t stopSize = totalSizeCompressed; int64_t bytesDelta = stopSize - startSize; From 46012556798ffa584aad79668cdb5e32da83beff Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 13 Aug 2026 09:47:11 +0200 Subject: [PATCH 6/7] Track AOD reader state per timeframe rather than table --- .../src/AODJAlienReaderHelpers.cxx | 160 +++++++++--------- 1 file changed, 79 insertions(+), 81 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index e30e6bb2fcfc2..5fd8f8c7858d7 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -252,100 +252,98 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices; *numTF = ntf; }; - enum class ReadState { - BEFORE_FIRST_READ, - FIRST_READ, - READ, - NOT_READ_AND_FIRST, - NOT_READ_AND_MIDDLE, + enum class TFReaderState { + READ_TIMEFRAME, + TRY_NEXT_FILE, + TIMEFRAME_READ, + INVALID_TIMEFRAME, + END_OF_INPUT, }; - auto readState = ReadState::BEFORE_FIRST_READ; - for (auto& route : requestedTables) { - if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) { - continue; + auto readState = TFReaderState::READ_TIMEFRAME; + bool triedNextFile = false; + while (readState == TFReaderState::READ_TIMEFRAME || readState == TFReaderState::TRY_NEXT_FILE) { + if (readState == TFReaderState::TRY_NEXT_FILE) { + fcnt += device.maxInputTimeslices; + if (didir->atEnd(fcnt)) { + readState = TFReaderState::END_OF_INPUT; + break; + } + ntf = 0; + triedNextFile = true; } - // create header - auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher); - auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); - bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); - - try { - if (didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - readState = readState == ReadState::BEFORE_FIRST_READ ? ReadState::FIRST_READ : ReadState::READ; - } else { - readState = readState == ReadState::BEFORE_FIRST_READ ? ReadState::NOT_READ_AND_FIRST : ReadState::NOT_READ_AND_MIDDLE; - } - } catch (InvalidAODReadError const& e) { - if (!skipInvalidReads) { - throw; + readState = TFReaderState::TIMEFRAME_READ; + bool firstTable = true; + for (auto& route : requestedTables) { + if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) { + continue; } - skipInvalidRead(concrete.origin, e); - return; - } - switch (readState) { - case ReadState::FIRST_READ: - case ReadState::READ: - break; - case ReadState::NOT_READ_AND_MIDDLE: - LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); - throw std::runtime_error("Processing is stopped!"); - case ReadState::NOT_READ_AND_FIRST: - // check if there is a next file to read - fcnt += device.maxInputTimeslices; - if (didir->atEnd(fcnt)) { - LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); - didir->closeInputFiles(); - monitoring.flushBuffer(); - control.endOfStream(); - control.readyToQuit(QuitRequest::Me); - return; - } - // get first folder of next file - ntf = 0; - try { - if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); - throw std::runtime_error("Processing is stopped!"); - } - } catch (InvalidAODReadError const& e) { - if (!skipInvalidReads) { - throw; + // create header + auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher); + auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); + bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); + + try { + if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { + if (firstTable && !triedNextFile) { + readState = TFReaderState::TRY_NEXT_FILE; + break; } - skipInvalidRead(concrete.origin, e); - return; + LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); + throw std::runtime_error("Processing is stopped!"); } - readState = ReadState::FIRST_READ; + } catch (InvalidAODReadError const& e) { + if (!skipInvalidReads) { + throw; + } + skipInvalidRead(concrete.origin, e); + readState = TFReaderState::INVALID_TIMEFRAME; break; - case ReadState::BEFORE_FIRST_READ: - throw std::logic_error("Invalid AOD read state"); - } - - if (readState == ReadState::FIRST_READ) { - if (reportTFN) { - // TF number - auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf); - auto o = Output(TFNumberHeader); - outputs.make(o) = timeFrameNumber; } - if (reportTFFileName) { - // Origin file name for derived output map - auto o2 = Output(TFFileNameHeader); - auto fileAndFolder = didir->getFileFolder(dh, fcnt, ntf); - auto rootFS = std::dynamic_pointer_cast(fileAndFolder.filesystem()); - auto* f = dynamic_cast(rootFS->GetFile()); - std::string currentFilename(f->GetFile()->GetName()); - if (strcmp(f->GetEndpointUrl()->GetProtocol(), "file") == 0 && f->GetEndpointUrl()->GetFile()[0] != '/') { - // This is not an absolute local path. Make it absolute. - static std::string pwd = gSystem->pwd() + std::string("/"); - currentFilename = pwd + std::string(f->GetName()); + if (firstTable) { + if (reportTFN) { + // TF number + auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf); + auto o = Output(TFNumberHeader); + outputs.make(o) = timeFrameNumber; + } + + if (reportTFFileName) { + // Origin file name for derived output map + auto o2 = Output(TFFileNameHeader); + auto fileAndFolder = didir->getFileFolder(dh, fcnt, ntf); + auto rootFS = std::dynamic_pointer_cast(fileAndFolder.filesystem()); + auto* f = dynamic_cast(rootFS->GetFile()); + std::string currentFilename(f->GetFile()->GetName()); + if (strcmp(f->GetEndpointUrl()->GetProtocol(), "file") == 0 && f->GetEndpointUrl()->GetFile()[0] != '/') { + // This is not an absolute local path. Make it absolute. + static std::string pwd = gSystem->pwd() + std::string("/"); + currentFilename = pwd + std::string(f->GetName()); + } + outputs.make(o2) = currentFilename; } - outputs.make(o2) = currentFilename; } + firstTable = false; } - readState = ReadState::READ; + } + + switch (readState) { + case TFReaderState::TIMEFRAME_READ: + break; + case TFReaderState::INVALID_TIMEFRAME: + return; + case TFReaderState::END_OF_INPUT: + LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); + didir->closeInputFiles(); + monitoring.flushBuffer(); + control.endOfStream(); + control.readyToQuit(QuitRequest::Me); + return; + case TFReaderState::READ_TIMEFRAME: + case TFReaderState::TRY_NEXT_FILE: + throw std::logic_error("Invalid timeframe read state"); } int64_t stopSize = totalSizeCompressed; int64_t bytesDelta = stopSize - startSize; From 4f77b72696890959144a460b7d4fc884c87bbbb0 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 13 Aug 2026 13:22:14 +0200 Subject: [PATCH 7/7] Track timeframe reading progress in reader state machine --- .../src/AODJAlienReaderHelpers.cxx | 116 ++++++++++-------- 1 file changed, 63 insertions(+), 53 deletions(-) diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 5fd8f8c7858d7..39878ce6b1127 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -253,15 +253,20 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const *numTF = ntf; }; enum class TFReaderState { - READ_TIMEFRAME, + READ_FIRST_TABLE, + READ_FIRST_TABLE_FROM_NEXT_FILE, + READ_NEXT_TABLE, TRY_NEXT_FILE, TIMEFRAME_READ, INVALID_TIMEFRAME, END_OF_INPUT, }; - auto readState = TFReaderState::READ_TIMEFRAME; - bool triedNextFile = false; - while (readState == TFReaderState::READ_TIMEFRAME || readState == TFReaderState::TRY_NEXT_FILE) { + auto readState = TFReaderState::READ_FIRST_TABLE; + size_t routeIndex = 0; + while (readState == TFReaderState::READ_FIRST_TABLE || + readState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE || + readState == TFReaderState::READ_NEXT_TABLE || + readState == TFReaderState::TRY_NEXT_FILE) { if (readState == TFReaderState::TRY_NEXT_FILE) { fcnt += device.maxInputTimeslices; if (didir->atEnd(fcnt)) { @@ -269,64 +274,67 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const break; } ntf = 0; - triedNextFile = true; + routeIndex = 0; + readState = TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE; } - readState = TFReaderState::TIMEFRAME_READ; - bool firstTable = true; - for (auto& route : requestedTables) { - if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) { - continue; - } + while (routeIndex < requestedTables.size() && + (device.inputTimesliceId % requestedTables[routeIndex].maxTimeslices) != requestedTables[routeIndex].timeslice) { + ++routeIndex; + } + if (routeIndex == requestedTables.size()) { + readState = TFReaderState::TIMEFRAME_READ; + break; + } - // create header - auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher); - auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); - bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); - - try { - if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - if (firstTable && !triedNextFile) { - readState = TFReaderState::TRY_NEXT_FILE; - break; - } - LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); - throw std::runtime_error("Processing is stopped!"); - } - } catch (InvalidAODReadError const& e) { - if (!skipInvalidReads) { - throw; + auto& route = requestedTables[routeIndex]; + auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher); + auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec); + bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); }); + + try { + if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { + if (readState == TFReaderState::READ_FIRST_TABLE) { + readState = TFReaderState::TRY_NEXT_FILE; + continue; } - skipInvalidRead(concrete.origin, e); - readState = TFReaderState::INVALID_TIMEFRAME; - break; + LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); + throw std::runtime_error("Processing is stopped!"); + } + } catch (InvalidAODReadError const& e) { + if (!skipInvalidReads) { + throw; } + skipInvalidRead(concrete.origin, e); + readState = TFReaderState::INVALID_TIMEFRAME; + break; + } - if (firstTable) { - if (reportTFN) { - // TF number - auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf); - auto o = Output(TFNumberHeader); - outputs.make(o) = timeFrameNumber; - } + if (readState == TFReaderState::READ_FIRST_TABLE || readState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) { + if (reportTFN) { + // TF number + auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf); + auto o = Output(TFNumberHeader); + outputs.make(o) = timeFrameNumber; + } - if (reportTFFileName) { - // Origin file name for derived output map - auto o2 = Output(TFFileNameHeader); - auto fileAndFolder = didir->getFileFolder(dh, fcnt, ntf); - auto rootFS = std::dynamic_pointer_cast(fileAndFolder.filesystem()); - auto* f = dynamic_cast(rootFS->GetFile()); - std::string currentFilename(f->GetFile()->GetName()); - if (strcmp(f->GetEndpointUrl()->GetProtocol(), "file") == 0 && f->GetEndpointUrl()->GetFile()[0] != '/') { - // This is not an absolute local path. Make it absolute. - static std::string pwd = gSystem->pwd() + std::string("/"); - currentFilename = pwd + std::string(f->GetName()); - } - outputs.make(o2) = currentFilename; + if (reportTFFileName) { + // Origin file name for derived output map + auto o2 = Output(TFFileNameHeader); + auto fileAndFolder = didir->getFileFolder(dh, fcnt, ntf); + auto rootFS = std::dynamic_pointer_cast(fileAndFolder.filesystem()); + auto* f = dynamic_cast(rootFS->GetFile()); + std::string currentFilename(f->GetFile()->GetName()); + if (strcmp(f->GetEndpointUrl()->GetProtocol(), "file") == 0 && f->GetEndpointUrl()->GetFile()[0] != '/') { + // This is not an absolute local path. Make it absolute. + static std::string pwd = gSystem->pwd() + std::string("/"); + currentFilename = pwd + std::string(f->GetName()); } + outputs.make(o2) = currentFilename; } - firstTable = false; } + ++routeIndex; + readState = TFReaderState::READ_NEXT_TABLE; } switch (readState) { @@ -341,7 +349,9 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const control.endOfStream(); control.readyToQuit(QuitRequest::Me); return; - case TFReaderState::READ_TIMEFRAME: + case TFReaderState::READ_FIRST_TABLE: + case TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE: + case TFReaderState::READ_NEXT_TABLE: case TFReaderState::TRY_NEXT_FILE: throw std::logic_error("Invalid timeframe read state"); }