Skip to content

Commit 64b2855

Browse files
autumn-mckautumn-mck
authored
DPL Analysis: skip input files which cannot be opened (#15965)
1 parent 3ec04ce commit 64b2855

6 files changed

Lines changed: 61 additions & 20 deletions

File tree

‎Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx‎

Lines changed: 31 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
260260
static size_t totalSizeCompressed = 0;
261261
static uint64_t totalDFSent = 0;
262262
static uint64_t totalInvalidReadSkipped = 0;
263+
static uint64_t totalFileOpenSkipped = 0;
263264

264265
// check if RuntimeLimit is reached
265266
if (!watchdog->update()) {
@@ -274,15 +275,18 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
274275

275276
int64_t startTime = uv_hrtime();
276277
int64_t startSize = totalSizeCompressed;
277-
auto skipInvalidRead = [&](ConcreteDataMatcher const& concrete, InvalidAODReadError const& e) {
278-
auto skippedTimeframes = ++totalInvalidReadSkipped;
279-
LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
280-
concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
278+
auto discardTimeframe = [&didir, &arrowContext, &messageContext, &stringContext]() {
281279
clean_all_runtime_errors();
282280
didir->finishTimeFrame(true);
283281
arrowContext.clear();
284282
messageContext.discard();
285283
stringContext.clear();
284+
};
285+
auto skipInvalidRead = [&fcnt, &ntf, &discardTimeframe, &dpstats, &fileCounter, &numTF, &device](ConcreteDataMatcher const& concrete, AODReadError const& e) {
286+
auto skippedTimeframes = ++totalInvalidReadSkipped;
287+
LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
288+
concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
289+
discardTimeframe();
286290
dpstats.updateStats({static_cast<short>(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), DataProcessingStats::Op::Add, 1});
287291
*fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices;
288292
*numTF = ntf;
@@ -294,6 +298,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
294298
TRY_NEXT_FILE,
295299
TIMEFRAME_READ,
296300
INVALID_TIMEFRAME,
301+
INVALID_FILE,
297302
};
298303
auto readState = TFReaderState::READ_FIRST_TABLE;
299304
[[maybe_unused]] auto stateName = [](TFReaderState state) -> char const* {
@@ -310,6 +315,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
310315
return "TIMEFRAME_READ";
311316
case TFReaderState::INVALID_TIMEFRAME:
312317
return "INVALID_TIMEFRAME";
318+
case TFReaderState::INVALID_FILE:
319+
return "INVALID_FILE";
313320
}
314321
return "UNKNOWN";
315322
};
@@ -344,7 +351,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
344351
outputs.make<std::string>(o2) = currentFilename;
345352
}
346353
};
347-
auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, skipInvalidReads](TFReaderState currentState) -> TFReaderState {
354+
auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, &discardTimeframe, &dpstats, skipInvalidReads](TFReaderState currentState) -> TFReaderState {
348355
while (routeIndex < requestedTables.size() &&
349356
(device.inputTimesliceId % requestedTables[routeIndex].maxTimeslices) != requestedTables[routeIndex].timeslice) {
350357
++routeIndex;
@@ -365,12 +372,24 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
365372
if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) {
366373
return TFReaderState::TRY_NEXT_FILE;
367374
}
368-
} catch (InvalidAODReadError const& e) {
375+
} catch (AODReadError const& e) {
369376
if (!skipInvalidReads) {
370377
throw;
371378
}
372-
skipInvalidRead(concrete, e);
373-
return TFReaderState::INVALID_TIMEFRAME;
379+
switch (e.getKind()) {
380+
case AODReadError::Kind::InvalidTimeframe:
381+
skipInvalidRead(concrete, e);
382+
return TFReaderState::INVALID_TIMEFRAME;
383+
case AODReadError::Kind::FileOpen: {
384+
auto skippedFiles = ++totalFileOpenSkipped;
385+
LOGP(error, "Unable to open AOD input for table {}: fileCounter {}, timeFrame {}. Skipping file (skipped files: {}). Reason: {}",
386+
concrete.origin.as<std::string>(), fcnt, ntf, skippedFiles, describeException(e));
387+
discardTimeframe();
388+
dpstats.updateStats({static_cast<short>(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), DataProcessingStats::Op::Add, 1});
389+
return TFReaderState::INVALID_FILE;
390+
}
391+
}
392+
throw;
374393
}
375394

376395
if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) {
@@ -394,6 +413,10 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
394413
throw std::runtime_error("Processing is stopped!");
395414
}
396415
break;
416+
case TFReaderState::INVALID_FILE:
417+
// Undo the discard so outputs read from the next file are checked as usual
418+
messageContext.clear();
419+
[[fallthrough]];
397420
case TFReaderState::TRY_NEXT_FILE:
398421
didir->finishTimeFrame();
399422
fcnt += device.maxInputTimeslices;

‎Framework/AnalysisSupport/src/DataInputDirector.cxx‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ bool DataInputDescriptor::setFile(int counter, int wantedParentLevel, std::strin
165165
if (tfile && !externalFile) {
166166
delete tfile;
167167
}
168-
throw std::runtime_error(fmt::format("Couldn't open file \"{}\"!", filename));
168+
throw AODReadError(AODReadError::Kind::FileOpen, fmt::format("Couldn't open file \"{}\"!", filename));
169169
}
170170
mCurrentFilesystem = std::make_shared<TFileFileSystem>(tfile, 50 * 1024 * 1024, mFactory, !externalFile);
171171
rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
@@ -363,7 +363,7 @@ std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int coun
363363
// The current DF is not found in the parent map (this should not happen and is a fatal error)
364364
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
365365
if (!parentFileName) {
366-
throw InvalidAODReadError(fmt::format(R"(parent file map exists but does not contain the current DF "{}" in file "{}")", folderName.c_str(), rootFS->GetFile()->GetName()));
366+
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()));
367367
}
368368

369369
if (mParentFile) {
@@ -392,7 +392,7 @@ std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int coun
392392
}
393393
} catch (...) {
394394
mParentFile.reset();
395-
std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName)));
395+
std::throw_with_nested(AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName)));
396396
}
397397
return mParentFile;
398398
}
@@ -575,11 +575,11 @@ try {
575575
}
576576
if (parentFile == nullptr) {
577577
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
578-
throw InvalidAODReadError(fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName()));
578+
throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName()));
579579
}
580580
if (parentNumTF == -1) {
581581
auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(parentFile->mCurrentFilesystem);
582-
throw InvalidAODReadError(fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName()));
582+
throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName()));
583583
}
584584
t.deactivate();
585585
return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed);
@@ -634,15 +634,15 @@ try {
634634
int parentNumTF = parentFile->findDFNumber(0, folder.path());
635635
if (parentNumTF == -1) {
636636
auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(parentFile->mCurrentFilesystem);
637-
throw InvalidAODReadError(fmt::format(R"(DF {} listed in parent file map but not found in the corresponding file "{}")", folder.path(), parentRootFS->GetFile()->GetName()));
637+
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()));
638638
}
639639
// first argument is 0 as the parent file object contains only 1 file
640640
return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed);
641641
}
642642

643643
auto schemaOpt = format->Inspect(fullpath);
644644
if (!schemaOpt.ok()) {
645-
throw InvalidAODReadError(fmt::format("Unable to inspect tree {}: {}", treename, schemaOpt.status().ToString()));
645+
throw AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to inspect tree {}: {}", treename, schemaOpt.status().ToString()));
646646
}
647647
auto physicalSchema = schemaOpt;
648648
std::vector<std::shared_ptr<arrow::Field>> fields;
@@ -673,16 +673,16 @@ try {
673673
f2b.release();
674674
} catch (...) {
675675
f2b.discard();
676-
std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to {} tree {}", operation, treename)));
676+
std::throw_with_nested(AODReadError(AODReadError::Kind::InvalidTimeframe, fmt::format("Unable to {} tree {}", operation, treename)));
677677
}
678678

679679
return true;
680-
} catch (InvalidAODReadError const&) {
680+
} catch (AODReadError const& e) {
681681
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
682682
auto filename = rootFS ? std::string(rootFS->GetFile()->GetName()) : mfilenames.at(counter).fileName;
683683
auto const& dfs = mfilenames.at(counter).listOfTimeFrameNumbers;
684684
auto df = numTF >= 0 && static_cast<size_t>(numTF) < dfs.size() ? fmt::format("DF_{}", dfs[numTF]) : fmt::format("timeframe index {}", numTF);
685-
std::throw_with_nested(InvalidAODReadError(fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel)));
685+
std::throw_with_nested(AODReadError(e.getKind(), fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel)));
686686
}
687687

688688
DataInputDirector::DataInputDirector(std::vector<std::string> inputFiles, DataInputDirectorContext&& context)

‎Framework/AnalysisSupport/src/DataInputDirector.h‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,19 @@ class Monitoring;
3333
namespace o2::framework
3434
{
3535

36-
class InvalidAODReadError : public std::runtime_error
36+
class AODReadError : public std::runtime_error
3737
{
3838
public:
39-
using std::runtime_error::runtime_error;
39+
enum class Kind {
40+
InvalidTimeframe, // skip the timeframe
41+
FileOpen, // skip the file, its timeframes are unknown
42+
};
43+
44+
AODReadError(Kind kind, std::string const& what) : std::runtime_error(what), mKind(kind) {}
45+
Kind getKind() const { return mKind; }
46+
47+
private:
48+
Kind mKind;
4049
};
4150

4251
struct FileNameHolder {

‎Framework/Core/include/Framework/DataProcessingStats.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ enum struct ProcessingStatsId : short {
7575
CCDB_CACHE_FETCHED_BYTES,
7676
CCDB_CACHE_REQUESTED_BYTES,
7777
AOD_INVALID_READ_SKIPPED_TIMEFRAMES,
78+
AOD_FILE_OPEN_SKIPPED_FILES,
7879
AVAILABLE_MANAGED_SHM_BASE = 512,
7980
};
8081

‎Framework/Core/src/CommonServices.cxx‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1127,6 +1127,13 @@ o2::framework::ServiceSpec CommonServices::dataProcessingStats()
11271127
.minPublishInterval = 0,
11281128
.maxRefreshLatency = 10000,
11291129
.sendInitialValue = true},
1130+
MetricSpec{.name = "aod-file-open-skipped-files",
1131+
.metricId = static_cast<short>(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES),
1132+
.kind = Kind::UInt64,
1133+
.scope = Scope::DPL,
1134+
.minPublishInterval = 0,
1135+
.maxRefreshLatency = 10000,
1136+
.sendInitialValue = true},
11301137
MetricSpec{.name = "arrow-bytes-destroyed",
11311138
.enabled = arrowAndResourceLimitingMetrics,
11321139
.metricId = static_cast<short>(ProcessingStatsId::ARROW_BYTES_DESTROYED),

‎Framework/Core/src/runDataProcessing.cxx‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1254,6 +1254,7 @@ std::vector<std::regex> getDumpableMetrics()
12541254
dumpableMetrics.emplace_back("^aod-file-read-info$");
12551255
dumpableMetrics.emplace_back("^aod-largest-object-written$");
12561256
dumpableMetrics.emplace_back("^aod-invalid-read-skipped-timeframes$");
1257+
dumpableMetrics.emplace_back("^aod-file-open-skipped-files$");
12571258
dumpableMetrics.emplace_back("^table-bytes-.*");
12581259
dumpableMetrics.emplace_back("^total-timeframes.*");
12591260
dumpableMetrics.emplace_back("^device_state.*");

0 commit comments

Comments
 (0)