From 39c7515180e4eca29f70dfb5c16711a687298696 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Mon, 3 Aug 2026 14:41:13 +0200 Subject: [PATCH 1/5] 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 ++ .../Core/include/Framework/MessageContext.h | 5 ++ Framework/Core/src/MessageContext.cxx | 7 ++ 5 files changed, 110 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; diff --git a/Framework/Core/include/Framework/MessageContext.h b/Framework/Core/include/Framework/MessageContext.h index 407bac0ceb00a..fce4b109a465e 100644 --- a/Framework/Core/include/Framework/MessageContext.h +++ b/Framework/Core/include/Framework/MessageContext.h @@ -466,6 +466,11 @@ class MessageContext /// discarded. void clear(); + /// Discard pending output messages without asserting that they were sent. This + /// is intended for exception teardown paths where normal post-processing will + /// not run. + void discard(); + FairMQDeviceProxy& proxy() { return mProxy; diff --git a/Framework/Core/src/MessageContext.cxx b/Framework/Core/src/MessageContext.cxx index 59dfc15837210..81bb3c7bb95c4 100644 --- a/Framework/Core/src/MessageContext.cxx +++ b/Framework/Core/src/MessageContext.cxx @@ -107,6 +107,13 @@ void MessageContext::clear() mMessages.clear(); } +void MessageContext::discard() +{ + mDidDispatch = false; + mScheduledMessages.clear(); + mMessages.clear(); +} + int64_t MessageContext::addToCache(std::unique_ptr& toCache) { auto&& cached = toCache->GetTransport()->CreateMessage(); From 31d8da101dc70f94150ab7cfdd4ebba8beeca785 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 12:31:53 +0200 Subject: [PATCH 2/5] 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 2cdd046dedc34..759659dac166e 100644 --- a/Framework/Core/src/CommonServices.cxx +++ b/Framework/Core/src/CommonServices.cxx @@ -1097,6 +1097,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 aaafa1060cadc5c41eb9862a5947be4de5505b45 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 12:32:39 +0200 Subject: [PATCH 3/5] 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 4f7dc69a20f3843c5d36f0e19848091b067d246a Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 12:39:08 +0200 Subject: [PATCH 4/5] 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 cac116ce6a0bd84935200827e40370974b886876 Mon Sep 17 00:00:00 2001 From: Autumn McKee Date: Thu, 6 Aug 2026 15:21:20 +0200 Subject: [PATCH 5/5] 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;