Skip to content

Commit 36c9bb1

Browse files
autumn-mckautumn-mck
committed
Support skipping invalid timeframes across parent files
1 parent 940909c commit 36c9bb1

3 files changed

Lines changed: 139 additions & 37 deletions

File tree

‎Framework/AnalysisSupport/src/AODJAlienReaderHelpers.cxx‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -275,7 +275,8 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
275275
auto skippedTimeframes = ++totalInvalidReadSkipped;
276276
LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
277277
concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
278-
didir->markTimeFrameSkipped(header::DataHeader(concrete.description, concrete.origin, concrete.subSpec), ntf);
278+
clean_all_runtime_errors();
279+
didir->finishTimeFrame(true);
279280
arrowContext.clear();
280281
messageContext.discard();
281282
stringContext.clear();
@@ -354,6 +355,9 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
354355
auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec);
355356
bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); });
356357

358+
if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) {
359+
didir->beginTimeFrame();
360+
}
357361
try {
358362
if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) {
359363
return TFReaderState::TRY_NEXT_FILE;
@@ -388,6 +392,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
388392
}
389393
break;
390394
case TFReaderState::TRY_NEXT_FILE:
395+
didir->finishTimeFrame();
391396
fcnt += device.maxInputTimeslices;
392397
if (didir->atEnd(fcnt)) {
393398
LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId);
@@ -407,6 +412,7 @@ AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const
407412
break;
408413
}
409414
}
415+
didir->finishTimeFrame();
410416
int64_t stopSize = totalSizeCompressed;
411417
int64_t bytesDelta = stopSize - startSize;
412418
int64_t stopTime = uv_hrtime();

‎Framework/AnalysisSupport/src/DataInputDirector.cxx‎

Lines changed: 115 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -161,10 +161,13 @@ bool DataInputDescriptor::setFile(int counter, int wantedParentLevel, std::strin
161161
if (tfile == nullptr) {
162162
tfile = TFile::Open(filename.c_str());
163163
}
164-
mCurrentFilesystem = std::make_shared<TFileFileSystem>(tfile, 50 * 1024 * 1024, mFactory, !externalFile);
165-
if (!mCurrentFilesystem.get()) {
164+
if (!tfile || tfile->IsZombie()) {
165+
if (tfile && !externalFile) {
166+
delete tfile;
167+
}
166168
throw std::runtime_error(fmt::format("Couldn't open file \"{}\"!", filename));
167169
}
170+
mCurrentFilesystem = std::make_shared<TFileFileSystem>(tfile, 50 * 1024 * 1024, mFactory, !externalFile);
168171
rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
169172
printFileOpening();
170173

@@ -254,6 +257,10 @@ std::pair<std::shared_ptr<DataInputDescriptor>, int> DataInputDescriptor::naviga
254257
if (!setFile(counter, wantedParentLevel, wantedOrigin)) {
255258
return {nullptr, -1};
256259
}
260+
if (numTF < 0 || numTF >= mfilenames[counter].numberOfTimeFrames) {
261+
return {nullptr, -1};
262+
}
263+
recordTimeFrameRead(counter, numTF);
257264
auto folderName = fmt::format("DF_{}", mfilenames[counter].listOfTimeFrameNumbers[numTF]);
258265
auto parentFile = getParentFile(counter, numTF, "", wantedParentLevel, wantedOrigin);
259266
if (parentFile == nullptr) {
@@ -283,18 +290,65 @@ arrow::dataset::FileSource DataInputDescriptor::getFileFolder(int counter, int n
283290
return {};
284291
}
285292

293+
recordTimeFrameRead(counter, numTF);
286294
mfilenames[counter].alreadyRead[numTF] = true;
287295

288296
return {fmt::format("DF_{}", mfilenames[counter].listOfTimeFrameNumbers[numTF]), mCurrentFilesystem};
289297
}
290298

291-
uint64_t DataInputDescriptor::markTimeFrameSkipped(int numTF)
299+
void DataInputDescriptor::recordTimeFrameRead(int counter, int numTF)
292300
{
293-
if (mCurrentFileID >= 0 && numTF >= 0 && numTF < mfilenames[mCurrentFileID].numberOfTimeFrames) {
294-
mfilenames[mCurrentFileID].alreadyRead[numTF] = false;
295-
return ++mfilenames[mCurrentFileID].invalidReadSkipped;
301+
auto read = std::pair{counter, numTF};
302+
if (std::find(mTimeFrameReads.begin(), mTimeFrameReads.end(), read) == mTimeFrameReads.end()) {
303+
mTimeFrameReads.push_back(read);
304+
}
305+
}
306+
307+
void DataInputDescriptor::beginTimeFrame()
308+
{
309+
mTimeFrameReads.clear();
310+
mTimeFrameActive = true;
311+
if (mParentFile) {
312+
mParentFile->beginTimeFrame();
313+
}
314+
}
315+
316+
void DataInputDescriptor::finishTimeFrame(bool skipped)
317+
{
318+
mTimeFrameActive = false;
319+
if (skipped) {
320+
for (auto [file, df] : mTimeFrameReads) {
321+
mfilenames[file].alreadyRead[df] = false;
322+
++mfilenames[file].invalidReadSkipped;
323+
}
324+
}
325+
mTimeFrameReads.clear();
326+
if (mParentFile) {
327+
mParentFile->finishTimeFrame(skipped);
328+
}
329+
for (auto& parent : mRetainedParents) {
330+
parent->finishTimeFrame(skipped);
331+
parent->closeInputFile();
332+
}
333+
mRetainedParents.clear();
334+
for (auto& [counter, info] : mPendingFileStatistics) {
335+
reportFileStatistics(counter, std::move(info));
336+
}
337+
mPendingFileStatistics.clear();
338+
}
339+
340+
void DataInputDescriptor::releaseParentFile()
341+
{
342+
if (!mParentFile) {
343+
return;
344+
}
345+
if (mTimeFrameActive) {
346+
// Keep both the read bookkeeping and the open file for final statistics.
347+
mRetainedParents.push_back(std::move(mParentFile));
348+
} else {
349+
mParentFile->closeInputFile();
350+
mParentFile.reset();
296351
}
297-
return 0;
298352
}
299353

300354
std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int counter, int numTF, std::string treename, int wantedParentLevel, std::string_view wantedOrigin)
@@ -309,8 +363,7 @@ std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int coun
309363
// The current DF is not found in the parent map (this should not happen and is a fatal error)
310364
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
311365
if (!parentFileName) {
312-
throw std::runtime_error(fmt::format(R"(parent file map exists but does not contain the current DF "{}" in file "{}")", folderName.c_str(), rootFS->GetFile()->GetName()));
313-
return nullptr;
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()));
314367
}
315368

316369
if (mParentFile) {
@@ -319,21 +372,28 @@ std::shared_ptr<DataInputDescriptor> DataInputDescriptor::getParentFile(int coun
319372
if (parentFileName->GetString().CompareTo(parentRootFS->GetFile()->GetName()) == 0) {
320373
return mParentFile;
321374
} else {
322-
mParentFile->closeInputFile();
323-
mParentFile.reset();
375+
releaseParentFile();
324376
}
325377
}
326378

327379
if (mLevel == mContext.allowedParentLevel) {
328-
throw std::runtime_error(fmt::format(R"(while looking for tree "{}", the parent file was requested but we are already at level {} of maximal allowed level {} for DF "{}" in file "{}")", treename.c_str(), mLevel, mContext.allowedParentLevel, folderName.c_str(),
329-
rootFS->GetFile()->GetName()));
380+
throw InvalidAODReadError(fmt::format(R"(while looking for tree "{}", the parent file was requested but we are already at level {} of maximal allowed level {} for DF "{}" in file "{}")", treename.c_str(), mLevel, mContext.allowedParentLevel, folderName.c_str(),
381+
rootFS->GetFile()->GetName()));
330382
}
331383

332384
LOGP(info, "Opening parent file {} for DF {}", parentFileName->GetString().Data(), folderName.c_str());
333385
mParentFile = std::make_shared<DataInputDescriptor>(mAlienSupport, mLevel + 1, mContext);
334386
mParentFile->mdefaultFilenamesPtr.emplace_back(makeFileNameHolder(parentFileName->GetString().Data()));
335387
mParentFile->fillInputfiles();
336-
mParentFile->setFile(0, wantedParentLevel, wantedOrigin);
388+
try {
389+
mParentFile->setFile(0, wantedParentLevel, wantedOrigin);
390+
if (mTimeFrameActive) {
391+
mParentFile->beginTimeFrame();
392+
}
393+
} catch (...) {
394+
mParentFile.reset();
395+
std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to open parent file \"{}\" for {}", parentFileName->GetString().Data(), folderName)));
396+
}
337397
return mParentFile;
338398
}
339399

@@ -373,15 +433,26 @@ void DataInputDescriptor::printFileStatistics()
373433
}
374434
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
375435
auto f = dynamic_cast<TFile*>(rootFS->GetFile());
376-
std::string monitoringInfo(fmt::format("lfn={},size={},total_df={},read_df={},skipped_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}", f->GetName(),
377-
f->GetSize(), getTimeFramesInFile(mCurrentFileID), getReadTimeFramesInFile(mCurrentFileID), mfilenames.at(mCurrentFileID).invalidReadSkipped, f->GetBytesRead(), f->GetReadCalls(),
436+
std::string monitoringInfo(fmt::format("lfn={},size={},total_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}", f->GetName(),
437+
f->GetSize(), getTimeFramesInFile(mCurrentFileID), f->GetBytesRead(), f->GetReadCalls(),
378438
((float)mIOTime / 1e9), ((float)wait_time / 1e9), mLevel));
379439
#if __has_include(<TJAlienFile.h>)
380440
auto alienFile = dynamic_cast<TJAlienFile*>(f);
381441
if (alienFile) {
382442
monitoringInfo += fmt::format(",se={},open_time={:.1f}", alienFile->GetSE(), alienFile->GetElapsed());
383443
}
384444
#endif
445+
if (mTimeFrameActive) {
446+
// Snapshot I/O statistics now, but publish DF counts after commit or rollback.
447+
mPendingFileStatistics.emplace_back(mCurrentFileID, std::move(monitoringInfo));
448+
} else {
449+
reportFileStatistics(mCurrentFileID, std::move(monitoringInfo));
450+
}
451+
}
452+
453+
void DataInputDescriptor::reportFileStatistics(int counter, std::string monitoringInfo)
454+
{
455+
monitoringInfo += fmt::format(",read_df={},skipped_df={}", getReadTimeFramesInFile(counter), mfilenames.at(counter).invalidReadSkipped);
385456
if (mContext.monitoring) {
386457
mContext.monitoring->send(o2::monitoring::Metric{monitoringInfo, "aod-file-read-info"}.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::DPL));
387458
}
@@ -391,10 +462,7 @@ void DataInputDescriptor::printFileStatistics()
391462
void DataInputDescriptor::closeInputFile()
392463
{
393464
if (mCurrentFilesystem.get()) {
394-
if (mParentFile) {
395-
mParentFile->closeInputFile();
396-
mParentFile.reset();
397-
}
465+
releaseParentFile();
398466

399467
delete mParentFileMap;
400468
mParentFileMap = nullptr;
@@ -492,7 +560,7 @@ struct CalculateDelta {
492560
};
493561

494562
bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh, int counter, int numTF, std::string treename, size_t& totalSizeCompressed, size_t& totalSizeUncompressed)
495-
{
563+
try {
496564
CalculateDelta t(mIOTime);
497565
std::string wantedOrigin = dh.dataOrigin.as<std::string>();
498566
int wantedLevel = mContext.levelForOrigin(wantedOrigin);
@@ -501,13 +569,17 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh
501569
// attempting to read from this level.
502570
if (wantedLevel != -1 && mLevel < wantedLevel) {
503571
auto [parentFile, parentNumTF] = navigateToLevel(counter, numTF, wantedLevel, wantedOrigin);
572+
if (counter >= getNumberInputfiles() || numTF < 0 || numTF >= mfilenames[counter].numberOfTimeFrames) {
573+
t.deactivate();
574+
return false;
575+
}
504576
if (parentFile == nullptr) {
505577
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
506-
throw std::runtime_error(fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName()));
578+
throw InvalidAODReadError(fmt::format(R"(No parent file found for "{}" while looking for level {} in "{}")", treename, wantedLevel, rootFS->GetFile()->GetName()));
507579
}
508580
if (parentNumTF == -1) {
509581
auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(parentFile->mCurrentFilesystem);
510-
throw std::runtime_error(fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName()));
582+
throw InvalidAODReadError(fmt::format(R"(DF not found in parent file "{}")", parentRootFS->GetFile()->GetName()));
511583
}
512584
t.deactivate();
513585
return parentFile->readTree(outputs, dh, 0, parentNumTF, treename, totalSizeCompressed, totalSizeUncompressed);
@@ -546,12 +618,7 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh
546618
if (!format) {
547619
t.deactivate();
548620
LOGP(debug, "Could not find tree {}. Trying in parent file.", fullpath.path());
549-
std::shared_ptr<DataInputDescriptor> parentFile;
550-
try {
551-
parentFile = getParentFile(counter, numTF, treename, wantedLevel, wantedOrigin);
552-
} catch (...) {
553-
std::throw_with_nested(InvalidAODReadError(fmt::format("Unable to resolve parent file for tree {}", treename)));
554-
}
621+
auto parentFile = getParentFile(counter, numTF, treename, wantedLevel, wantedOrigin);
555622
if (parentFile == nullptr) {
556623
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
557624
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()));
@@ -599,6 +666,12 @@ bool DataInputDescriptor::readTree(DataAllocator& outputs, header::DataHeader dh
599666
}
600667

601668
return true;
669+
} catch (InvalidAODReadError const&) {
670+
auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
671+
auto filename = rootFS ? std::string(rootFS->GetFile()->GetName()) : mfilenames.at(counter).fileName;
672+
auto const& dfs = mfilenames.at(counter).listOfTimeFrameNumbers;
673+
auto df = numTF >= 0 && static_cast<size_t>(numTF) < dfs.size() ? fmt::format("DF_{}", dfs[numTF]) : fmt::format("timeframe index {}", numTF);
674+
std::throw_with_nested(InvalidAODReadError(fmt::format("Reading tree {} in {} from file \"{}\" at parent level {}", treename, df, filename, mLevel)));
602675
}
603676

604677
DataInputDirector::DataInputDirector(std::vector<std::string> inputFiles, DataInputDirectorContext&& context)
@@ -888,13 +961,20 @@ arrow::dataset::FileSource DataInputDirector::getFileFolder(header::DataHeader d
888961
return didesc->getFileFolder(counter, numTF, wantedLevel, origin);
889962
}
890963

891-
void DataInputDirector::markTimeFrameSkipped(header::DataHeader dh, int numTF)
964+
void DataInputDirector::beginTimeFrame()
892965
{
893-
auto didesc = getDataInputDescriptor(dh);
894-
if (!didesc) {
895-
didesc = mdefaultDataInputDescriptor.get();
966+
mdefaultDataInputDescriptor->beginTimeFrame();
967+
for (auto& descriptor : mdataInputDescriptors) {
968+
descriptor.beginTimeFrame();
969+
}
970+
}
971+
972+
void DataInputDirector::finishTimeFrame(bool skipped)
973+
{
974+
mdefaultDataInputDescriptor->finishTimeFrame(skipped);
975+
for (auto& descriptor : mdataInputDescriptors) {
976+
descriptor.finishTimeFrame(skipped);
896977
}
897-
didesc->markTimeFrameSkipped(numTF);
898978
}
899979

900980
int DataInputDirector::getTimeFramesInFile(header::DataHeader dh, int counter)
@@ -944,6 +1024,7 @@ bool DataInputDirector::readTree(DataAllocator& outputs, header::DataHeader dh,
9441024

9451025
void DataInputDirector::closeInputFiles()
9461026
{
1027+
finishTimeFrame();
9471028
mdefaultDataInputDescriptor->closeInputFile();
9481029
for (auto& didesc : mdataInputDescriptors) {
9491030
didesc.closeInputFile();

‎Framework/AnalysisSupport/src/DataInputDirector.h‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,10 @@ class DataInputDescriptor
107107

108108
uint64_t getTimeFrameNumber(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin);
109109
arrow::dataset::FileSource getFileFolder(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin);
110-
uint64_t markTimeFrameSkipped(int numTF);
110+
// Start tracking the file/DF pairs participating in one output timeframe.
111+
void beginTimeFrame();
112+
// Commit or discard a timeframe, then close retained parents and report statistics.
113+
void finishTimeFrame(bool skipped = false);
111114
// Open the current file to populate the parent map, then return the parent descriptor and
112115
// the TF index within it that corresponds to numTF at this level. Returns {nullptr, -1} on failure.
113116
std::pair<std::shared_ptr<DataInputDescriptor>, int> navigateToLevel(int counter, int numTF, int wantedParentLevel, std::string_view wantedOrigin);
@@ -123,6 +126,10 @@ class DataInputDescriptor
123126
bool isAlienSupportOn() { return mAlienSupport; }
124127

125128
private:
129+
void recordTimeFrameRead(int counter, int numTF);
130+
void releaseParentFile();
131+
void reportFileStatistics(int counter, std::string monitoringInfo);
132+
126133
o2::framework::RootObjectReadingFactory mFactory;
127134
std::string minputfilesFile;
128135
std::string* minputfilesFilePtr = nullptr;
@@ -132,11 +139,16 @@ class DataInputDescriptor
132139
std::vector<FileNameHolder> mdefaultFilenamesPtr;
133140
std::shared_ptr<arrow::fs::FileSystem> mCurrentFilesystem;
134141
int mCurrentFileID = -1;
142+
// File-local DF indices visited during the current output timeframe.
143+
std::vector<std::pair<int, int>> mTimeFrameReads;
135144
bool mAlienSupport = false;
136145

137146
DataInputDirectorContext& mContext;
138147
TMap* mParentFileMap = nullptr;
139148
std::shared_ptr<DataInputDescriptor> mParentFile = nullptr;
149+
std::vector<std::shared_ptr<DataInputDescriptor>> mRetainedParents;
150+
bool mTimeFrameActive = false;
151+
std::vector<std::pair<int, std::string>> mPendingFileStatistics;
140152
int mLevel = 0; // level of parent files
141153

142154
int mtotalNumberTimeFrames = 0;
@@ -173,7 +185,10 @@ class DataInputDirector
173185
bool readTree(DataAllocator& outputs, header::DataHeader dh, int counter, int numTF, size_t& totalSizeCompressed, size_t& totalSizeUncompressed, bool wasAOD);
174186
uint64_t getTimeFrameNumber(header::DataHeader dh, int counter, int numTF);
175187
arrow::dataset::FileSource getFileFolder(header::DataHeader dh, int counter, int numTF);
176-
void markTimeFrameSkipped(header::DataHeader dh, int numTF);
188+
// Start tracking the file/DF pairs participating in one output timeframe.
189+
void beginTimeFrame();
190+
// Commit or discard a timeframe, then close retained parents and report statistics.
191+
void finishTimeFrame(bool skipped = false);
177192
int getTimeFramesInFile(header::DataHeader dh, int counter);
178193

179194
uint64_t getTotalSizeCompressed();

0 commit comments

Comments
 (0)