21#include "Monitoring/Tags.h"
22#include "Monitoring/Metric.h"
23#include "Monitoring/Monitoring.h"
25#include "rapidjson/document.h"
26#include "rapidjson/prettywriter.h"
27#include "rapidjson/filereadstream.h"
30#include "TObjString.h"
34#include <arrow/dataset/file_base.h>
35#include <arrow/dataset/dataset.h>
39#if __has_include(<TJAlienFile.h>)
40#include <TJAlienFile.h>
50using namespace rapidjson;
55 fileNameHolder->fileName = fileName;
57 return fileNameHolder;
61 : mAlienSupport(alienSupport),
62 mMonitoring(monitoring),
63 mAllowedParentLevel(allowedParentLevel),
64 mParentFileReplacement(
std::move(parentFileReplacement)),
67 std::vector<char const*> capabilitiesSpecs = {
68 "O2Framework:RNTupleObjectReadingCapability",
69 "O2Framework:TTreeObjectReadingCapability",
72 std::vector<LoadablePlugin> plugins;
73 for (
auto spec : capabilitiesSpecs) {
75 for (
auto& extra : morePlugins) {
76 plugins.push_back(extra);
80 PluginManager::loadFromPlugin<RootObjectReadingCapability, RootObjectReadingCapabilityPlugin>(plugins, mFactory.
capabilities);
85 LOGP(info,
"DataInputDescriptor");
86 LOGP(info,
" Table name : {}",
tablename);
87 LOGP(info,
" Tree name : {}",
treename);
90 LOGP(info,
" Input files : {}", mfilenames.size());
91 for (
auto fn : mfilenames) {
92 LOGP(info,
" {} {}", fn->fileName, fn->numberOfTimeFrames);
99 return (minputfilesFile.empty() && minputfilesFilePtr) ? (std::string)*minputfilesFilePtr : minputfilesFile;
104 return (mFilenameRegex.empty() && mFilenameRegexPtr) ? (std::string)*mFilenameRegexPtr : mFilenameRegex;
115 if (fn->
fileName.rfind(
"file://", 0) == 0) {
117 }
else if (!mAlienSupport && fn->
fileName.rfind(
"alien://", 0) == 0 && !gGrid) {
118 LOGP(
debug,
"AliEn file requested. Enabling support.");
119 TGrid::Connect(
"alien://");
120 mAlienSupport =
true;
124 mfilenames.emplace_back(fn);
136 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
138 if (rootFS->GetFile()->GetName() ==
filename) {
144 mCurrentFilesystem = std::make_shared<TFileFileSystem>(TFile::Open(
filename.c_str()), 50 * 1024 * 1024, mFactory);
145 if (!mCurrentFilesystem.get()) {
146 throw std::runtime_error(fmt::format(
"Couldn't open file \"{}\"!",
filename));
148 rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
152 mParentFileMap = (TMap*)rootFS->GetFile()->Get(
"parentFiles");
153 if (mParentFileMap && !mParentFileReplacement.empty()) {
154 auto pos = mParentFileReplacement.find(
';');
155 if (
pos == std::string::npos) {
156 throw std::runtime_error(fmt::format(
"Invalid syntax in aod-parent-base-path-replacement: \"{}\"", mParentFileReplacement.c_str()));
158 auto from = mParentFileReplacement.substr(0,
pos);
159 auto to = mParentFileReplacement.substr(
pos + 1);
161 auto it = mParentFileMap->MakeIterator();
162 while (
auto obj = it->Next()) {
163 auto objString = (TObjString*)mParentFileMap->GetValue(obj);
164 objString->String().ReplaceAll(from.c_str(), to.c_str());
170 if (mfilenames[
counter]->numberOfTimeFrames <= 0) {
171 const std::regex TFRegex = std::regex(
"/?DF_([0-9]+)(|-.*)$");
172 TList* keyList = rootFS->GetFile()->GetListOfKeys();
173 std::vector<std::string> finalList;
179 std::unordered_set<size_t> seen;
180 for (
auto key : *keyList) {
181 std::smatch matchResult;
182 std::string keyName = ((TObjString*)
key)->GetString().Data();
183 bool match = std::regex_match(keyName, matchResult, TFRegex);
185 auto folderNumber = std::stoul(matchResult[1].
str());
186 if (seen.find(folderNumber) == seen.end()) {
187 seen.insert(folderNumber);
188 mfilenames[
counter]->listOfTimeFrameNumbers.emplace_back(folderNumber);
193 if (mParentFileMap !=
nullptr) {
195 std::sort(mfilenames[
counter]->listOfTimeFrameNumbers.begin(), mfilenames[
counter]->listOfTimeFrameNumbers.end(),
196 [
this](
long const& l1,
long const& l2) ->
bool {
197 auto p1 = (TObjString*)this->mParentFileMap->GetValue((
"DF_" + std::to_string(l1)).c_str());
198 auto p2 = (TObjString*)this->mParentFileMap->GetValue((
"DF_" + std::to_string(l2)).c_str());
199 return p1->GetString().CompareTo(p2->GetString()) < 0;
202 std::sort(mfilenames[
counter]->listOfTimeFrameNumbers.begin(), mfilenames[
counter]->listOfTimeFrameNumbers.end());
205 mfilenames[
counter]->alreadyRead.resize(mfilenames[
counter]->alreadyRead.size() + mfilenames[
counter]->listOfTimeFrameNumbers.size(),
false);
206 mfilenames[
counter]->numberOfTimeFrames = mfilenames[
counter]->listOfTimeFrameNumbers.size();
210 mCurrentFileStartedAt = uv_hrtime();
225 if (mfilenames[
counter]->numberOfTimeFrames > 0 && numTF >= mfilenames[
counter]->numberOfTimeFrames) {
229 return (mfilenames[
counter]->listOfTimeFrameNumbers)[numTF];
240 if (mfilenames[
counter]->numberOfTimeFrames > 0 && numTF >= mfilenames[
counter]->numberOfTimeFrames) {
244 mfilenames[
counter]->alreadyRead[numTF] =
true;
246 return {fmt::format(
"DF_{}", mfilenames[
counter]->listOfTimeFrameNumbers[numTF]), mCurrentFilesystem};
251 if (!mParentFileMap) {
255 auto folderName = fmt::format(
"DF_{}", mfilenames[
counter]->listOfTimeFrameNumbers[numTF]);
256 auto parentFileName = (TObjString*)mParentFileMap->GetValue(folderName.c_str());
258 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
259 if (!parentFileName) {
260 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()));
266 auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(mParentFile->mCurrentFilesystem);
267 if (parentFileName->GetString().CompareTo(parentRootFS->GetFile()->GetName()) == 0) {
272 mParentFile =
nullptr;
276 if (mLevel == mAllowedParentLevel) {
277 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, mAllowedParentLevel, folderName.c_str(),
278 rootFS->GetFile()->GetName()));
281 LOGP(info, "Opening parent file {} for DF {}", parentFileName->GetString().Data(), folderName.c_str());
282 mParentFile =
new DataInputDescriptor(mAlienSupport, mLevel + 1, mMonitoring, mAllowedParentLevel, mParentFileReplacement);
283 mParentFile->mdefaultFilenamesPtr =
new std::vector<FileNameHolder*>;
284 mParentFile->mdefaultFilenamesPtr->emplace_back(
makeFileNameHolder(parentFileName->GetString().Data()));
292 return mfilenames.at(
counter)->numberOfTimeFrames;
298 return std::count(
list.begin(),
list.end(),
true);
303 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
304 auto f =
dynamic_cast<TFile*
>(rootFS->GetFile());
305 std::string monitoringInfo(fmt::format(
"lfn={},size={}",
f->GetName(),
f->GetSize()));
306#if __has_include(<TJAlienFile.h>)
307 auto alienFile =
dynamic_cast<TJAlienFile*
>(
f);
309 monitoringInfo += fmt::format(
",se={},open_time={:.1f}", alienFile->GetSE(), alienFile->GetElapsed());
312 mMonitoring->send(o2::monitoring::Metric{monitoringInfo,
"aod-file-open-info"}.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::DPL));
313 LOGP(info,
"Opening file: {}", monitoringInfo);
318 int64_t wait_time = (int64_t)uv_hrtime() - (int64_t)mCurrentFileStartedAt - (int64_t)mIOTime;
322 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
323 auto f =
dynamic_cast<TFile*
>(rootFS->GetFile());
324 std::string monitoringInfo(fmt::format(
"lfn={},size={},total_df={},read_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}",
f->GetName(),
326 ((
float)mIOTime / 1e9), ((
float)wait_time / 1e9), mLevel));
327#if __has_include(<TJAlienFile.h>)
328 auto alienFile =
dynamic_cast<TJAlienFile*
>(
f);
330 monitoringInfo += fmt::format(
",se={},open_time={:.1f}", alienFile->GetSE(), alienFile->GetElapsed());
333 mMonitoring->send(o2::monitoring::Metric{monitoringInfo,
"aod-file-read-info"}.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::DPL));
334 LOGP(info,
"Read info: {}", monitoringInfo);
339 if (mCurrentFilesystem.get()) {
343 mParentFile =
nullptr;
346 delete mParentFileMap;
347 mParentFileMap =
nullptr;
350 mCurrentFilesystem.reset();
362 if (!fileName.empty()) {
365 std::ifstream filelist(fileName);
366 if (!filelist.is_open()) {
367 throw std::runtime_error(fmt::format(R
"(Couldn't open file "{}")", fileName));
369 while (std::getline(filelist, fileName)) {
371 fileName.erase(std::remove_if(fileName.begin(), fileName.end(), ::isspace), fileName.end());
383 if (mdefaultFilenamesPtr) {
384 for (
auto fileNameHolder : *mdefaultFilenamesPtr) {
398 auto dfList = mfilenames[
file]->listOfTimeFrameNumbers;
399 auto it = std::find_if(dfList.begin(), dfList.end(), [dfName](
size_t i) { return fmt::format(
"DF_{}", i) == dfName; });
400 if (it == dfList.end()) {
403 return it - dfList.begin();
418 void (*dump_)(
const char*);
419 if (
void* sym = dlsym(
nullptr,
"igprof_dump_now")) {
420 dump_ = __extension__(
void (*)(
const char*)) sym;
422 std::string
filename = fmt::format(
"reader-memory-dump-{}.gz", uv_hrtime());
444 if (!folder.filesystem()) {
449 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(folder.filesystem());
453 throw std::runtime_error(fmt::format(R
"(Not a TFile filesystem!)"));
456 std::shared_ptr<arrow::dataset::FileFormat>
format;
459 auto fullpath = arrow::dataset::FileSource{folder.path() +
"/" +
treename, folder.filesystem()};
462 auto objectPath = capability.lfn2objectPath(fullpath.path());
463 void* handle = capability.getHandle(rootFS, objectPath);
465 format = capability.factory().format();
466 creator = capability.factory().deferredOutputStreamer;
475 LOGP(
debug,
"Could not find tree {}. Trying in parent file.", fullpath.path());
477 if (parentFile !=
nullptr) {
478 int parentNumTF = parentFile->findDFNumber(0, folder.path());
479 if (parentNumTF == -1) {
480 auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(parentFile->mCurrentFilesystem);
481 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()));
484 return parentFile->readTree(outputs, dh, 0, parentNumTF,
treename, totalSizeCompressed, totalSizeUncompressed);
486 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
487 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()));
490 auto schemaOpt =
format->Inspect(fullpath);
491 auto physicalSchema = schemaOpt;
492 std::vector<std::shared_ptr<arrow::Field>> fields;
493 for (
auto& original : (*schemaOpt)->fields()) {
494 if (original->name().ends_with(
"_size")) {
497 fields.push_back(original);
499 auto datasetSchema = std::make_shared<arrow::Schema>(fields);
501 auto fragment =
format->MakeFragment(fullpath, {}, *physicalSchema);
513 f2b->fill(datasetSchema,
format);
523DataInputDirector::DataInputDirector(std::string inputFile, o2::monitoring::Monitoring* monitoring,
int allowedParentLevel, std::string parentFileReplacement) : mMonitoring(monitoring), mAllowedParentLevel(allowedParentLevel), mParentFileReplacement(
std::move(parentFileReplacement))
525 if (inputFile.size() && inputFile[0] ==
'@') {
526 inputFile.erase(0, 1);
535DataInputDirector::DataInputDirector(std::vector<std::string> inputFiles, o2::monitoring::Monitoring* monitoring,
int allowedParentLevel, std::string parentFileReplacement) : mMonitoring(monitoring), mAllowedParentLevel(allowedParentLevel), mParentFileReplacement(
std::move(parentFileReplacement))
537 for (
auto inputFile : inputFiles) {
546 for (
auto fn : mdefaultInputFiles) {
549 mdefaultInputFiles.clear();
550 mdefaultDataInputDescriptor =
nullptr;
552 for (
auto fn : mdataInputDescriptors) {
555 mdataInputDescriptors.clear();
560 mdataInputDescriptors.clear();
561 mdefaultInputFiles.clear();
562 mFilenameRegex = std::string(
"");
567 if (mdefaultDataInputDescriptor) {
568 delete mdefaultDataInputDescriptor;
570 mdefaultDataInputDescriptor =
new DataInputDescriptor(mAlienSupport, 0, mMonitoring, mAllowedParentLevel, mParentFileReplacement);
575 mdefaultDataInputDescriptor->
tablename =
"any";
576 mdefaultDataInputDescriptor->
treename =
"any";
585 FILE*
f = fopen(fnjson.c_str(),
"r");
587 LOGP(error,
"Could not open JSON file \"{}\"!", fnjson);
592 char readBuffer[65536];
593 FileReadStream inputStream(
f, readBuffer,
sizeof(readBuffer));
597 jsonDoc.ParseStream(inputStream);
598 auto status = readJsonDocument(&jsonDoc);
606bool DataInputDirector::readJsonDocument(Document* jsonDoc)
609 std::string fileName(
"");
610 const char* itemName;
613 if (jsonDoc->HasParseError()) {
614 LOGP(error,
"Check the JSON document! There is a problem with the format!");
619 itemName =
"InputDirector";
620 const Value& didirItem = (*jsonDoc)[itemName];
621 if (!didirItem.IsObject()) {
622 LOGP(info,
"No \"{}\" object found in the JSON document!", itemName);
627 itemName =
"debugmode";
628 if (didirItem.HasMember(itemName)) {
629 if (didirItem[itemName].IsBool()) {
630 mDebugMode = (didirItem[itemName].GetBool());
632 LOGP(error,
"Check the JSON document! Item \"{}\" must be a boolean!", itemName);
642 PrettyWriter<StringBuffer> writer(
buffer);
643 didirItem.Accept(writer);
644 LOGP(info,
"InputDirector object: {}", std::string(
buffer.GetString()));
647 itemName =
"fileregex";
648 if (didirItem.HasMember(itemName)) {
649 if (didirItem[itemName].IsString()) {
652 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
657 itemName =
"resfiles";
658 if (didirItem.HasMember(itemName)) {
659 if (didirItem[itemName].IsString()) {
660 fileName = didirItem[itemName].GetString();
661 if (fileName.size() && fileName[0] ==
'@') {
662 fileName.erase(0, 1);
668 }
else if (didirItem[itemName].IsArray()) {
670 auto fns = didirItem[itemName].GetArray();
671 for (
auto& fn : fns) {
675 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string or an array!", itemName);
680 itemName =
"InputDescriptors";
681 if (didirItem.HasMember(itemName)) {
682 if (!didirItem[itemName].IsArray()) {
683 LOGP(error,
"Check the JSON document! Item \"{}\" must be an array!", itemName);
688 for (
auto& didescItem : didirItem[itemName].GetArray()) {
689 if (!didescItem.IsObject()) {
690 LOGP(error,
"Check the JSON document! \"{}\" must be objects!", itemName);
694 auto didesc =
new DataInputDescriptor(mAlienSupport, 0, mMonitoring, mAllowedParentLevel, mParentFileReplacement);
695 didesc->setDefaultInputfiles(&mdefaultInputFiles);
698 if (didescItem.HasMember(itemName)) {
699 if (didescItem[itemName].IsString()) {
700 didesc->tablename = didescItem[itemName].GetString();
703 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
707 LOGP(error,
"Check the JSON document! Item \"{}\" is missing!", itemName);
711 itemName =
"treename";
712 if (didescItem.HasMember(itemName)) {
713 if (didescItem[itemName].IsString()) {
714 didesc->treename = didescItem[itemName].GetString();
716 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
721 didesc->treename =
m[2];
724 itemName =
"fileregex";
725 if (didescItem.HasMember(itemName)) {
726 if (didescItem[itemName].IsString()) {
727 if (didesc->getNumberInputfiles() == 0) {
728 didesc->setFilenamesRegex(didescItem[itemName].GetString());
731 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
735 if (didesc->getNumberInputfiles() == 0) {
736 didesc->setFilenamesRegex(mFilenameRegexPtr);
740 itemName =
"resfiles";
741 if (didescItem.HasMember(itemName)) {
742 if (didescItem[itemName].IsString()) {
743 fileName = didescItem[itemName].GetString();
744 if (fileName.size() && fileName[0] ==
'@') {
745 didesc->setInputfilesFile(fileName.erase(0, 1));
747 if (didesc->getFilenamesRegexString().empty() ||
748 std::regex_match(fileName, didesc->getFilenamesRegex())) {
752 }
else if (didescItem[itemName].IsArray()) {
753 auto fns = didescItem[itemName].GetArray();
754 for (
auto& fn : fns) {
755 if (didesc->getFilenamesRegexString().empty() ||
756 std::regex_match(fn.GetString(), didesc->getFilenamesRegex())) {
761 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string or an array!", itemName);
765 didesc->setInputfilesFile(minputfilesFilePtr);
769 if (didesc->fillInputfiles() > 0) {
770 mdataInputDescriptors.emplace_back(didesc);
773 LOGP(info,
"This DataInputDescriptor is ignored because its file list is empty!");
775 mAlienSupport &= didesc->isAlienSupportOn();
803 for (
auto didesc : mdataInputDescriptors) {
804 if (didesc->matcher->match(dh, context)) {
818 didesc = mdefaultDataInputDescriptor;
829 didesc = mdefaultDataInputDescriptor;
840 didesc = mdefaultDataInputDescriptor;
848 std::string treename;
853 treename = didesc->treename;
858 didesc = mdefaultDataInputDescriptor;
862 auto result = didesc->readTree(outputs, dh,
counter, numTF, treename, totalSizeCompressed, totalSizeUncompressed);
869 for (
auto didesc : mdataInputDescriptors) {
870 didesc->closeInputFile();
874bool DataInputDirector::isValid()
878 for (
auto didesc : mdataInputDescriptors) {
879 status &= didesc->getNumberInputfiles() == numberFiles;
888 for (
auto didesc : mdataInputDescriptors) {
889 status &= (didesc->getNumberInputfiles() <=
counter);
897 LOGP(info,
"DataInputDirector");
898 LOGP(info,
" Default input files file : {}", minputfilesFile);
899 LOGP(info,
" Default file name regex : {}", mFilenameRegex);
900 LOGP(info,
" Default file names : {}", mdefaultInputFiles.size());
901 for (
auto const& fn : mdefaultInputFiles) {
902 LOGP(info,
" {} {}", fn->fileName, fn->numberOfTimeFrames);
904 LOGP(info,
" Default DataInputDescriptor:");
905 mdefaultDataInputDescriptor->
printOut();
907 for (
auto const& didesc : mdataInputDescriptors) {
o2::monitoring::tags::Value Value
#define O2_DECLARE_DYNAMIC_LOG(name)
#define O2_SIGNPOST_ACTION(log, callback)
decltype(auto) make(const Output &spec, Args... args)
std::function< std::shared_ptr< arrow::io::OutputStream >(std::shared_ptr< arrow::dataset::FileFragment >, const std::shared_ptr< arrow::ResizableBuffer > &buffer)> StreamerCreator
void setLabel(const char *label)
bool match(const std::vector< std::string > &queries, const char *pattern)
typedef void(APIENTRYP PFNGLCULLFACEPROC)(GLenum mode)
GLint GLint GLsizei GLint GLenum format
std::string getTreeName(header::DataHeader dh)
Defining PrimaryVertex explicitly as messageable.
FileNameHolder * makeFileNameHolder(std::string fileName)
Defining DataPointCompositeObject explicitly as copiable.
CalculateDelta(uint64_t &target)
static std::unique_ptr< data_matcher::DataDescriptorMatcher > buildNode(std::string const &nodeString)
static std::vector< std::string > getTokens(std::string const &nodeString)
static std::vector< LoadablePlugin > parsePluginSpecString(char const *str)
Parse a comma separated list of <library>:<plugin-name> plugin declarations.
std::vector< RootObjectReadingCapability > capabilities