diff --git a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx index 1abc4b9ffdd48..1c3d028dd9287 100644 --- a/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx +++ b/Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx @@ -10,15 +10,21 @@ // or submit itself to any jurisdiction. #include "AODJAlienReaderHelpers.h" +#include #include +#include +#include +#include #include #include +#include #include #include "Framework/TableTreeHelpers.h" #include "Framework/AnalysisHelpers.h" #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 +32,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" @@ -101,6 +109,27 @@ using o2::monitoring::tags::Value; namespace o2::framework::readers { +static bool shouldSkipInvalidReads() +{ + auto const* envValue = getenv("DPL_AOD_READER_SKIP_INVALID"); + return envValue != nullptr && + strcmp(envValue, "0") != 0 && + strcmp(envValue, "false") != 0; +} + +static std::string describeException(std::exception const& exception) +{ + std::string description{exception.what()}; + try { + std::rethrow_if_nested(exception); + } catch (std::exception const& nested) { + description += ": " + describeException(nested); + } catch (...) { + description += ": unknown exception"; + } + return description; +} + AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const& ctx) { // aod-parent-base-path-replacement is now a workflow option, so it needs to be @@ -193,6 +222,7 @@ 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 = shouldSkipInvalidReads(); return adaptStateless([TFNumberHeader, TFFileNameHeader, requestedTables, @@ -200,7 +230,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const numTF, watchdog, maxRate, - didir, reportTFN, reportTFFileName, level](Monitoring& monitoring, DataAllocator& outputs, ControlService& control, DeviceSpec const& device, DataProcessingStats& dpstats) { + 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 assert(device.inputTimesliceId < device.maxInputTimeslices); @@ -214,10 +245,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; // check if RuntimeLimit is reached if (!watchdog->update()) { @@ -232,19 +263,98 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const int64_t startTime = uv_hrtime(); int64_t startSize = totalSizeCompressed; - for (auto& route : requestedTables) { - if ((device.inputTimesliceId % route.maxTimeslices) != route.timeslice) { - continue; + 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, describeException(e)); + arrowContext.clear(); + messageContext.discard(); + stringContext.clear(); + dpstats.updateStats({static_cast(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), DataProcessingStats::Op::Add, 1}); + *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices; + *numTF = ntf; + }; + enum class TFReaderState { + READ_FIRST_TABLE, + READ_FIRST_TABLE_FROM_NEXT_FILE, + READ_NEXT_TABLE, + TRY_NEXT_FILE, + TIMEFRAME_READ, + INVALID_TIMEFRAME, + }; + auto readState = TFReaderState::READ_FIRST_TABLE; + size_t routeIndex = 0; + auto reportTimeframe = [&didir, &fcnt, &ntf, &outputs, &TFNumberHeader, &TFFileNameHeader, reportTFN, reportTFFileName](header::DataHeader const& dh) { + if (reportTFN) { + // TF number + auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf); + auto o = Output(TFNumberHeader); + outputs.make(o) = timeFrameNumber; } - // create header + 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; + } + }; + auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, skipInvalidReads](TFReaderState currentState) -> TFReaderState { + while (routeIndex < requestedTables.size() && + (device.inputTimesliceId % requestedTables[routeIndex].maxTimeslices) != requestedTables[routeIndex].timeslice) { + ++routeIndex; + } + if (routeIndex == requestedTables.size()) { + return TFReaderState::TIMEFRAME_READ; + } + + 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"); }); - if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { - if (first) { - // check if there is a next file to read + try { + if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) { + return TFReaderState::TRY_NEXT_FILE; + } + } catch (InvalidAODReadError const& e) { + if (!skipInvalidReads) { + throw; + } + skipInvalidRead(concrete.origin, e); + return TFReaderState::INVALID_TIMEFRAME; + } + + if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) { + reportTimeframe(dh); + } + ++routeIndex; + return TFReaderState::READ_NEXT_TABLE; + }; + while (readState != TFReaderState::TIMEFRAME_READ) { + switch (readState) { + case TFReaderState::READ_FIRST_TABLE: + readState = tryReadTable(readState); + break; + case TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE: + case TFReaderState::READ_NEXT_TABLE: + readState = tryReadTable(readState); + if (readState == TFReaderState::TRY_NEXT_FILE) { + // Once a file has been selected, every requested table must exist. + auto concrete = DataSpecUtils::asConcreteDataMatcher(requestedTables[routeIndex].matcher); + LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); + throw std::runtime_error("Processing is stopped!"); + } + break; + case TFReaderState::TRY_NEXT_FILE: fcnt += device.maxInputTimeslices; if (didir->atEnd(fcnt)) { LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId); @@ -254,42 +364,15 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const 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!"); - } - } else { - LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as(), fcnt, ntf); - throw std::runtime_error("Processing is stopped!"); - } - } - - if (first) { - 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; - } + routeIndex = 0; + readState = TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE; + break; + case TFReaderState::INVALID_TIMEFRAME: + return; + case TFReaderState::TIMEFRAME_READ: + break; } - first = false; } int64_t stopSize = totalSizeCompressed; int64_t bytesDelta = stopSize - startSize; diff --git a/Framework/AnalysisSupport/src/DataInputDirector.cxx b/Framework/AnalysisSupport/src/DataInputDirector.cxx index cfd578862fabd..ad4b73bfd916b 100644 --- a/Framework/AnalysisSupport/src/DataInputDirector.cxx +++ b/Framework/AnalysisSupport/src/DataInputDirector.cxx @@ -34,6 +34,8 @@ #include #include #include +#include +#include #include #if __has_include() @@ -536,18 +538,23 @@ 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::throw_with_nested(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 +580,15 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh //// add branches to read //// fill the table f2b->setLabel(treename.c_str()); - f2b->fill(datasetSchema, format); + std::experimental::scope_exit discardOnError{[&f2b] { f2b.discard(); }}; + char const* operation = "read"; + try { + f2b->fill(datasetSchema, format); + operation = "finalize"; + f2b.release(); + } catch (...) { + std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to {} tree {}", operation, 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/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),