DPL Analysis: skip input files which cannot be opened

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, AODReadError 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,12 +372,24 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
           if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) {
             return TFReaderState::TRY_NEXT_FILE;
           }
-        } catch (InvalidAODReadError const& e) {
+        } catch (AODReadError const& e) {
           if (!skipInvalidReads) {
             throw;
           }
-          skipInvalidRead(concrete, e);
-          return TFReaderState::INVALID_TIMEFRAME;
+          switch (e.getKind()) {
+            case AODReadError::Kind::InvalidTimeframe:
+              skipInvalidRead(concrete, e);
+              return TFReaderState::INVALID_TIMEFRAME;
+            case AODReadError::Kind::FileOpen: {
+              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;
+            }
+          }
+          throw;
         }
 
         if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) {
@@ -394,6 +413,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 AODReadError(AODReadError::Kind::FileOpen, 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);
@@ -363,7 +363,7 @@ std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int coun
   // The current DF is not found in the parent map (this should not happen and is a fatal error)
   auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
   if (!parentFileName) {
-    throw InvalidAODReadError(fmt::format(R"(parent file map exists but does not contain the current DF "{}" in file "{}")", folderName.c_str(), rootFS->GetFile()->GetName()));
+    throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(parent file map exists but does not contain the current DF "{}" in file "{}")", folderName.c_str(), rootFS->GetFile()->GetName()));
   }
 
   if (mParentFile) {
@@ -392,7 +392,7 @@ std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int coun
     }
   } catch (...) {
     mParentFile.reset();
-    std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName)));
+    std::throw_with_nested(AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName)));
   }
   return mParentFile;
 }
@@ -575,11 +575,11 @@ try {
     }
     if (parentFile == nullptr) {
       auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
-      throw InvalidAODReadError(fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName()));
+      throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName()));
     }
     if (parentNumTF == -1) {
       auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(parentFile->mCurrentFilesystem);
-      throw InvalidAODReadError(fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName()));
+      throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName()));
     }
     t.deactivate();
     return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed);
@@ -634,15 +634,15 @@ try {
     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()));
+      throw AODReadError(AODReadError::Kind::InvalidTimeframe, 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);
   if (!schemaOpt.ok()) {
-    throw InvalidAODReadError(fmt::format("Unable to inspect tree {}: {}", treename, schemaOpt.status().ToString()));
+    throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to inspect tree {}: {}", treename, schemaOpt.status().ToString()));
   }
   auto physicalSchema = schemaOpt;
   std::vector<std::shared_ptr<arrow::Field>> fields;
@@ -673,16 +673,16 @@ try {
     f2b.release();
   } catch (...) {
     f2b.discard();
-    std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to {} tree {}", operation, treename)));
+    std::throw_with_nested(AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to {} tree {}", operation, treename)));
   }
 
   return true;
-} catch (InvalidAODReadError const&) {
+} catch (AODReadError const& e) {
   auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
   auto filename = rootFS ? std::string(rootFS->GetFile()->GetName()) : mfilenames.at(counter).fileName;
   auto const& dfs = mfilenames.at(counter).listOfTimeFrameNumbers;
   auto df = numTF >= 0 && static_cast<size_t>(numTF) < dfs.size() ? fmt::format("DF_{}", dfs[numTF]) : fmt::format("timeframe index {}", numTF);
-  std::throw_with_nested(InvalidAODReadError(fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel)));
+  std::throw_with_nested(AODReadError(e.getKind(), fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel)));
 }
 
 DataInputDirector::DataInputDirector(std::vector<std::string> inputFiles, DataInputDirectorContext&& context)

Framework/AnalysisSupport/src/DataInputDirector.h

@@ -33,10 +33,19 @@ class Monitoring;
 namespace o2::framework
 {
 
-class InvalidAODReadError : public std::runtime_error
+class AODReadError : public std::runtime_error
 {
  public:
-  using std::runtime_error::runtime_error;
+  enum class Kind {
+    InvalidTimeframe, // skip the timeframe
+    FileOpen,         // skip the file, its timeframes are unknown
+  };
+
+  AODReadError(Kind kind, std::string const& what) : std::runtime_error(what), mKind(kind) {}
+  Kind getKind() const { return mKind; }
+
+ private:
+  Kind mKind;
 };
 
 struct FileNameHolder {

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.*");