DPL Analysis: skip input files which cannot be opened

Author: autumn-mck

Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx

@@ -260,6 +260,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
       static size_t totalSizeCompressed = 0;
       static uint64_t totalDFSent = 0;
       static uint64_t totalInvalidReadSkipped = 0;
+      static uint64_t totalFileOpenSkipped = 0;
 
       // check if RuntimeLimit is reached
       if (!watchdog->update()) {
@@ -274,15 +275,18 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
 
       int64_t startTime = uv_hrtime();
       int64_t startSize = totalSizeCompressed;
-      auto skipInvalidRead = [&](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) {
-        auto skippedTimeframes = ++totalInvalidReadSkipped;
-        LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
-             concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
+      auto discardTimeframe = [&didir, &arrowContext, &messageContext, &stringContext]() {
         clean_all_runtime_errors();
         didir->finishTimeFrame(true);
         arrowContext.clear();
         messageContext.discard();
         stringContext.clear();
+      };
+      auto skipInvalidRead = [&fcnt, &ntf, &discardTimeframe, &dpstats, &fileCounter, &numTF, &device](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) {
+        auto skippedTimeframes = ++totalInvalidReadSkipped;
+        LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
+             concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
+        discardTimeframe();
         dpstats.updateStats({static_cast<short>(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), DataProcessingStats::Op::Add, 1});
         *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices;
         *numTF = ntf;
@@ -294,6 +298,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
         TRY_NEXT_FILE,
         TIMEFRAME_READ,
         INVALID_TIMEFRAME,
+        INVALID_FILE,
       };
       auto readState = TFReaderState::READ_FIRST_TABLE;
       [[maybe_unused]] auto stateName = [](TFReaderState state) -> char const* {
@@ -310,6 +315,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
             return "TIMEFRAME_READ";
           case TFReaderState::INVALID_TIMEFRAME:
             return "INVALID_TIMEFRAME";
+          case TFReaderState::INVALID_FILE:
+            return "INVALID_FILE";
         }
         return "UNKNOWN";
       };
@@ -344,7 +351,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
           outputs.make<std::string>(o2) = currentFilename;
         }
       };
-      auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, skipInvalidReads](TFReaderState currentState) -> TFReaderState {
+      auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, &discardTimeframe, &dpstats, skipInvalidReads](TFReaderState currentState) -> TFReaderState {
         while (routeIndex < requestedTables.size() &&
                (device.inputTimesliceId % requestedTables[routeIndex].maxTimeslices) != requestedTables[routeIndex].timeslice) {
           ++routeIndex;
@@ -365,6 +372,16 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
           if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) {
             return TFReaderState::TRY_NEXT_FILE;
           }
+        } catch (AODFileOpenError const& e) {
+          if (!skipInvalidReads) {
+            throw;
+          }
+          auto skippedFiles = ++totalFileOpenSkipped;
+          LOGP(error, "Unable to open AOD input for table {}: fileCounter {}, timeFrame {}. Skipping file (skipped files: {}). Reason: {}",
+               concrete.origin.as<std::string>(), fcnt, ntf, skippedFiles, describeException(e));
+          discardTimeframe();
+          dpstats.updateStats({static_cast<short>(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), DataProcessingStats::Op::Add, 1});
+          return TFReaderState::INVALID_FILE;
         } catch (InvalidAODReadError const& e) {
           if (!skipInvalidReads) {
             throw;
@@ -394,6 +411,10 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
               throw std::runtime_error("Processing is stopped!");
             }
             break;
+          case TFReaderState::INVALID_FILE:
+            // Undo the discard so outputs read from the next file are checked as usual
+            messageContext.clear();
+            [[fallthrough]];
           case TFReaderState::TRY_NEXT_FILE:
             didir->finishTimeFrame();
             fcnt += device.maxInputTimeslices;

Framework/AnalysisSupport/src/DataInputDirector.cxx

@@ -165,7 +165,7 @@ bool DataInputDescriptor::setFile(int counter, int wantedParentLevel, std::strin
     if (tfile && !externalFile) {
       delete tfile;
     }
-    throw std::runtime_error(fmt::format("Couldn't open file \"{}\"!", filename));
+    throw AODFileOpenError(fmt::format("Couldn't open file \"{}\"!", filename));
   }
   mCurrentFilesystem = std::make_shared<TFileFileSystem>(tfile, 50 * 1024 * 1024, mFactory, !externalFile);
   rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);

Framework/AnalysisSupport/src/DataInputDirector.h

@@ -39,6 +39,14 @@ class InvalidAODReadError : public std::runtime_error
   using std::runtime_error::runtime_error;
 };
 
+// Opening an input file can fail before its timeframes are known. The reader
+// must advance to the next file rather than retrying subsequent timeframe indices.
+class AODFileOpenError : public std::runtime_error
+{
+ public:
+  using std::runtime_error::runtime_error;
+};
+
 struct FileNameHolder {
   std::string fileName;
   int numberOfTimeFrames = 0;

Framework/Core/include/Framework/DataProcessingStats.h

@@ -75,6 +75,7 @@ enum struct ProcessingStatsId : short {
   CCDB_CACHE_FETCHED_BYTES,
   CCDB_CACHE_REQUESTED_BYTES,
   AOD_INVALID_READ_SKIPPED_TIMEFRAMES,
+  AOD_FILE_OPEN_SKIPPED_FILES,
   AVAILABLE_MANAGED_SHM_BASE = 512,
 };
 

Framework/Core/src/CommonServices.cxx

@@ -1127,6 +1127,13 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats()
                    .minPublishInterval = 0,
                    .maxRefreshLatency = 10000,
                    .sendInitialValue = true},
+        MetricSpec{.name = "aod-file-open-skipped-files",
+                   .metricId = static_cast<short>(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES),
+                   .kind = Kind::UInt64,
+                   .scope = Scope::DPL,
+                   .minPublishInterval = 0,
+                   .maxRefreshLatency = 10000,
+                   .sendInitialValue = true},
         MetricSpec{.name = "arrow-bytes-destroyed",
                    .enabled = arrowAndResourceLimitingMetrics,
                    .metricId = static_cast<short>(ProcessingStatsId::ARROW_BYTES_DESTROYED),

Framework/Core/src/runDataProcessing.cxx

@@ -1254,6 +1254,7 @@ std::vector<std::regex> getDumpableMetrics()
   dumpableMetrics.emplace_back("^aod-file-read-info$");
   dumpableMetrics.emplace_back("^aod-largest-object-written$");
   dumpableMetrics.emplace_back("^aod-invalid-read-skipped-timeframes$");
+  dumpableMetrics.emplace_back("^aod-file-open-skipped-files$");
   dumpableMetrics.emplace_back("^table-bytes-.*");
   dumpableMetrics.emplace_back("^total-timeframes.*");
   dumpableMetrics.emplace_back("^device_state.*");