Skip invalid timeframes during ROOT input (#15576)

Handle corrupt reads as recoverable and discard the affected timeframe when DPL_AOD_READER_SKIP_INVALID is enabled.

Author: autumn-mck

Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx

@@ -10,22 +10,31 @@
 // or submit itself to any jurisdiction.
 
 #include "AODJAlienReaderHelpers.h"
+#include <algorithm>
 #include <charconv>
+#include <cctype>
+#include <cstdlib>
+#include <exception>
 #include <memory>
 #include <ranges>
+#include <string>
+#include <string_view>
 #include <vector>
 #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"
 #include "Framework/EndOfStreamContext.h"
 #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 +110,31 @@ using o2::monitoring::tags::Value;
 
 namespace o2::framework::readers
 {
+static bool shouldSkipInvalidReads()
+{
+  auto const* envValue = getenv("DPL_AOD_READER_SKIP_INVALID");
+  if (envValue == nullptr) {
+    return false;
+  }
+
+  std::string value{envValue};
+  std::ranges::transform(value, value.begin(), [](unsigned char c) { return std::tolower(c); });
+  return !value.empty() && value != "0" && value != "false";
+}
+
+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,14 +227,16 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
     int level = originLevelMapping.empty() ? -1 : 0;
     auto fileCounter = std::make_shared<int>(0);
     auto numTF = std::make_shared<int>(-1);
+    bool const skipInvalidReads = shouldSkipInvalidReads();
     return adaptStateless([TFNumberHeader,
                            TFFileNameHeader,
                            requestedTables,
                            fileCounter,
                            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 +250,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 +268,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<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
+        arrowContext.clear();
+        messageContext.discard();
+        stringContext.clear();
+        dpstats.updateStats({static_cast<short>(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<uint64_t>(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<TFileFileSystem>(fileAndFolder.filesystem());
+          auto* f = dynamic_cast<TFile*>(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<std::string>(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;
         }
 
-        // create header
+        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<std::string>(), 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 +369,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<std::string>(), fcnt, ntf);
-              throw std::runtime_error("Processing is stopped!");
-            }
-          } else {
-            LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as<std::string>(), 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<uint64_t>(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<TFileFileSystem>(fileAndFolder.filesystem());
-            auto* f = dynamic_cast<TFile*>(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<std::string>(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;

Framework/AnalysisSupport/src/DataInputDirector.cxx

@@ -34,6 +34,7 @@
 #include <arrow/dataset/file_base.h>
 #include <arrow/dataset/dataset.h>
 #include <uv.h>
+#include <exception>
 #include <memory>
 
 #if __has_include(<TJAlienFile.h>)
@@ -536,18 +537,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<TFileFileSystem>(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<DataInputDescriptor> 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<TFileFileSystem>(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<TFileFileSystem>(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<TFileFileSystem>(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 +579,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);
+  char const* operation = "read";
+  try {
+    f2b->fill(datasetSchema, format);
+    operation = "finalize";
+    f2b.release();
+  } catch (...) {
+    f2b.discard();
+    std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to {} tree {}", operation, treename)));
+  }
 
   return true;
 }

Framework/AnalysisSupport/src/DataInputDirector.h

@@ -21,6 +21,7 @@
 #include <arrow/dataset/dataset.h>
 
 #include <regex>
+#include <stdexcept>
 #include <vector>
 #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;

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,
 };
 

Framework/Core/src/CommonServices.cxx

@@ -1120,6 +1120,13 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats()
         MetricSpec{.name = "dropped_computations", .metricId = static_cast<short>(ProcessingStatsId::DROPPED_COMPUTATIONS), .kind = Kind::UInt64, .minPublishInterval = quickUpdateInterval},
         MetricSpec{.name = "dropped_incoming_messages", .metricId = static_cast<short>(ProcessingStatsId::DROPPED_INCOMING_MESSAGES), .kind = Kind::UInt64, .minPublishInterval = quickUpdateInterval},
         MetricSpec{.name = "relayed_messages", .metricId = static_cast<short>(ProcessingStatsId::RELAYED_MESSAGES), .kind = Kind::UInt64, .minPublishInterval = quickUpdateInterval},
+        MetricSpec{.name = "aod-invalid-read-skipped-timeframes",
+                   .metricId = static_cast<short>(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<short>(ProcessingStatsId::ARROW_BYTES_DESTROYED),