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) {
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);
151 mParentFileMap = (TMap*)rootFS->GetFile()->Get(
"parentFiles");
152 if (mParentFileMap && !mParentFileReplacement.empty()) {
153 auto pos = mParentFileReplacement.find(
';');
154 if (
pos == std::string::npos) {
155 throw std::runtime_error(fmt::format(
"Invalid syntax in aod-parent-base-path-replacement: \"{}\"", mParentFileReplacement.c_str()));
157 auto from = mParentFileReplacement.substr(0,
pos);
158 auto to = mParentFileReplacement.substr(
pos + 1);
160 auto it = mParentFileMap->MakeIterator();
161 while (
auto obj = it->Next()) {
162 auto objString = (TObjString*)mParentFileMap->GetValue(obj);
163 objString->String().ReplaceAll(from.c_str(), to.c_str());
169 if (mfilenames[
counter]->numberOfTimeFrames <= 0) {
170 const std::regex TFRegex = std::regex(
"/?DF_([0-9]+)(|-.*)$");
171 TList* keyList = rootFS->GetFile()->GetListOfKeys();
172 std::vector<std::string> finalList;
178 std::unordered_set<size_t> seen;
179 for (
auto key : *keyList) {
180 std::smatch matchResult;
181 std::string keyName = ((TObjString*)
key)->GetString().Data();
182 bool match = std::regex_match(keyName, matchResult, TFRegex);
184 auto folderNumber = std::stoul(matchResult[1].
str());
185 if (seen.find(folderNumber) == seen.end()) {
186 seen.insert(folderNumber);
187 mfilenames[
counter]->listOfTimeFrameNumbers.emplace_back(folderNumber);
192 if (mParentFileMap !=
nullptr) {
194 std::sort(mfilenames[
counter]->listOfTimeFrameNumbers.begin(), mfilenames[
counter]->listOfTimeFrameNumbers.end(),
195 [
this](
long const& l1,
long const& l2) ->
bool {
196 auto p1 = (TObjString*)this->mParentFileMap->GetValue((
"DF_" + std::to_string(l1)).c_str());
197 auto p2 = (TObjString*)this->mParentFileMap->GetValue((
"DF_" + std::to_string(l2)).c_str());
198 return p1->GetString().CompareTo(p2->GetString()) < 0;
201 std::sort(mfilenames[
counter]->listOfTimeFrameNumbers.begin(), mfilenames[
counter]->listOfTimeFrameNumbers.end());
204 mfilenames[
counter]->alreadyRead.resize(mfilenames[
counter]->alreadyRead.size() + mfilenames[
counter]->listOfTimeFrameNumbers.size(),
false);
205 mfilenames[
counter]->numberOfTimeFrames = mfilenames[
counter]->listOfTimeFrameNumbers.size();
209 mCurrentFileStartedAt = uv_hrtime();
224 if (mfilenames[
counter]->numberOfTimeFrames > 0 && numTF >= mfilenames[
counter]->numberOfTimeFrames) {
228 return (mfilenames[
counter]->listOfTimeFrameNumbers)[numTF];
239 if (mfilenames[
counter]->numberOfTimeFrames > 0 && numTF >= mfilenames[
counter]->numberOfTimeFrames) {
243 mfilenames[
counter]->alreadyRead[numTF] =
true;
245 return {fmt::format(
"DF_{}", mfilenames[
counter]->listOfTimeFrameNumbers[numTF]), mCurrentFilesystem};
250 if (!mParentFileMap) {
254 auto folderName = fmt::format(
"DF_{}", mfilenames[
counter]->listOfTimeFrameNumbers[numTF]);
255 auto parentFileName = (TObjString*)mParentFileMap->GetValue(folderName.c_str());
257 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
258 if (!parentFileName) {
259 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()));
265 auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(mParentFile->mCurrentFilesystem);
266 if (parentFileName->GetString().CompareTo(parentRootFS->GetFile()->GetName()) == 0) {
271 mParentFile =
nullptr;
275 if (mLevel == mAllowedParentLevel) {
276 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(),
277 rootFS->GetFile()->GetName()));
280 LOGP(info, "Opening parent file {} for DF {}", parentFileName->GetString().Data(), folderName.c_str());
281 mParentFile =
new DataInputDescriptor(mAlienSupport, mLevel + 1, mMonitoring, mAllowedParentLevel, mParentFileReplacement);
282 mParentFile->mdefaultFilenamesPtr =
new std::vector<FileNameHolder*>;
283 mParentFile->mdefaultFilenamesPtr->emplace_back(
makeFileNameHolder(parentFileName->GetString().Data()));
291 return mfilenames.at(
counter)->numberOfTimeFrames;
297 return std::count(
list.begin(),
list.end(),
true);
302 int64_t wait_time = (int64_t)uv_hrtime() - (int64_t)mCurrentFileStartedAt - (int64_t)mIOTime;
306 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
307 auto f =
dynamic_cast<TFile*
>(rootFS->GetFile());
308 std::string monitoringInfo(fmt::format(
"lfn={},size={},total_df={},read_df={},read_bytes={},read_calls={},io_time={:.1f},wait_time={:.1f},level={}",
f->GetName(),
310 ((
float)mIOTime / 1e9), ((
float)wait_time / 1e9), mLevel));
311#if __has_include(<TJAlienFile.h>)
312 auto alienFile =
dynamic_cast<TJAlienFile*
>(
f);
314 monitoringInfo += fmt::format(
",se={},open_time={:.1f}", alienFile->GetSE(), alienFile->GetElapsed());
317 mMonitoring->send(o2::monitoring::Metric{monitoringInfo,
"aod-file-read-info"}.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::DPL));
318 LOGP(info,
"Read info: {}", monitoringInfo);
323 if (mCurrentFilesystem.get()) {
327 mParentFile =
nullptr;
330 delete mParentFileMap;
331 mParentFileMap =
nullptr;
334 mCurrentFilesystem.reset();
346 if (!fileName.empty()) {
349 std::ifstream filelist(fileName);
350 if (!filelist.is_open()) {
351 throw std::runtime_error(fmt::format(R
"(Couldn't open file "{}")", fileName));
353 while (std::getline(filelist, fileName)) {
355 fileName.erase(std::remove_if(fileName.begin(), fileName.end(), ::isspace), fileName.end());
367 if (mdefaultFilenamesPtr) {
368 for (
auto fileNameHolder : *mdefaultFilenamesPtr) {
382 auto dfList = mfilenames[
file]->listOfTimeFrameNumbers;
383 auto it = std::find_if(dfList.begin(), dfList.end(), [dfName](
size_t i) { return fmt::format(
"DF_{}", i) == dfName; });
384 if (it == dfList.end()) {
387 return it - dfList.begin();
402 void (*dump_)(
const char*);
403 if (
void* sym = dlsym(
nullptr,
"igprof_dump_now")) {
404 dump_ = __extension__(
void (*)(
const char*)) sym;
406 std::string
filename = fmt::format(
"reader-memory-dump-{}.gz", uv_hrtime());
428 if (!folder.filesystem()) {
433 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(folder.filesystem());
437 throw std::runtime_error(fmt::format(R
"(Not a TFile filesystem!)"));
440 std::shared_ptr<arrow::dataset::FileFormat>
format;
443 auto fullpath = arrow::dataset::FileSource{folder.path() +
"/" +
treename, folder.filesystem()};
446 auto objectPath = capability.lfn2objectPath(fullpath.path());
447 void* handle = capability.getHandle(rootFS, objectPath);
449 format = capability.factory().format();
450 creator = capability.factory().deferredOutputStreamer;
459 LOGP(
debug,
"Could not find tree {}. Trying in parent file.", fullpath.path());
461 if (parentFile !=
nullptr) {
462 int parentNumTF = parentFile->findDFNumber(0, folder.path());
463 if (parentNumTF == -1) {
464 auto parentRootFS = std::dynamic_pointer_cast<TFileFileSystem>(parentFile->mCurrentFilesystem);
465 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()));
468 return parentFile->readTree(outputs, dh, 0, parentNumTF,
treename, totalSizeCompressed, totalSizeUncompressed);
470 auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(mCurrentFilesystem);
471 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()));
474 auto schemaOpt =
format->Inspect(fullpath);
475 auto physicalSchema = schemaOpt;
476 std::vector<std::shared_ptr<arrow::Field>> fields;
477 for (
auto& original : (*schemaOpt)->fields()) {
478 if (original->name().ends_with(
"_size")) {
481 fields.push_back(original);
483 auto datasetSchema = std::make_shared<arrow::Schema>(fields);
485 auto fragment =
format->MakeFragment(fullpath, {}, *physicalSchema);
497 f2b->fill(datasetSchema,
format);
507DataInputDirector::DataInputDirector(std::string inputFile, o2::monitoring::Monitoring* monitoring,
int allowedParentLevel, std::string parentFileReplacement) : mMonitoring(monitoring), mAllowedParentLevel(allowedParentLevel), mParentFileReplacement(
std::move(parentFileReplacement))
509 if (inputFile.size() && inputFile[0] ==
'@') {
510 inputFile.erase(0, 1);
519DataInputDirector::DataInputDirector(std::vector<std::string> inputFiles, o2::monitoring::Monitoring* monitoring,
int allowedParentLevel, std::string parentFileReplacement) : mMonitoring(monitoring), mAllowedParentLevel(allowedParentLevel), mParentFileReplacement(
std::move(parentFileReplacement))
521 for (
auto inputFile : inputFiles) {
530 for (
auto fn : mdefaultInputFiles) {
533 mdefaultInputFiles.clear();
534 mdefaultDataInputDescriptor =
nullptr;
536 for (
auto fn : mdataInputDescriptors) {
539 mdataInputDescriptors.clear();
544 mdataInputDescriptors.clear();
545 mdefaultInputFiles.clear();
546 mFilenameRegex = std::string(
"");
551 if (mdefaultDataInputDescriptor) {
552 delete mdefaultDataInputDescriptor;
554 mdefaultDataInputDescriptor =
new DataInputDescriptor(mAlienSupport, 0, mMonitoring, mAllowedParentLevel, mParentFileReplacement);
559 mdefaultDataInputDescriptor->
tablename =
"any";
560 mdefaultDataInputDescriptor->
treename =
"any";
569 FILE*
f = fopen(fnjson.c_str(),
"r");
571 LOGP(error,
"Could not open JSON file \"{}\"!", fnjson);
576 char readBuffer[65536];
577 FileReadStream inputStream(
f, readBuffer,
sizeof(readBuffer));
581 jsonDoc.ParseStream(inputStream);
582 auto status = readJsonDocument(&jsonDoc);
590bool DataInputDirector::readJsonDocument(Document* jsonDoc)
593 std::string fileName(
"");
594 const char* itemName;
597 if (jsonDoc->HasParseError()) {
598 LOGP(error,
"Check the JSON document! There is a problem with the format!");
603 itemName =
"InputDirector";
604 const Value& didirItem = (*jsonDoc)[itemName];
605 if (!didirItem.IsObject()) {
606 LOGP(info,
"No \"{}\" object found in the JSON document!", itemName);
611 itemName =
"debugmode";
612 if (didirItem.HasMember(itemName)) {
613 if (didirItem[itemName].IsBool()) {
614 mDebugMode = (didirItem[itemName].GetBool());
616 LOGP(error,
"Check the JSON document! Item \"{}\" must be a boolean!", itemName);
626 PrettyWriter<StringBuffer> writer(
buffer);
627 didirItem.Accept(writer);
628 LOGP(info,
"InputDirector object: {}", std::string(
buffer.GetString()));
631 itemName =
"fileregex";
632 if (didirItem.HasMember(itemName)) {
633 if (didirItem[itemName].IsString()) {
636 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
641 itemName =
"resfiles";
642 if (didirItem.HasMember(itemName)) {
643 if (didirItem[itemName].IsString()) {
644 fileName = didirItem[itemName].GetString();
645 if (fileName.size() && fileName[0] ==
'@') {
646 fileName.erase(0, 1);
652 }
else if (didirItem[itemName].IsArray()) {
654 auto fns = didirItem[itemName].GetArray();
655 for (
auto& fn : fns) {
659 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string or an array!", itemName);
664 itemName =
"InputDescriptors";
665 if (didirItem.HasMember(itemName)) {
666 if (!didirItem[itemName].IsArray()) {
667 LOGP(error,
"Check the JSON document! Item \"{}\" must be an array!", itemName);
672 for (
auto& didescItem : didirItem[itemName].GetArray()) {
673 if (!didescItem.IsObject()) {
674 LOGP(error,
"Check the JSON document! \"{}\" must be objects!", itemName);
678 auto didesc =
new DataInputDescriptor(mAlienSupport, 0, mMonitoring, mAllowedParentLevel, mParentFileReplacement);
679 didesc->setDefaultInputfiles(&mdefaultInputFiles);
682 if (didescItem.HasMember(itemName)) {
683 if (didescItem[itemName].IsString()) {
684 didesc->tablename = didescItem[itemName].GetString();
687 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
691 LOGP(error,
"Check the JSON document! Item \"{}\" is missing!", itemName);
695 itemName =
"treename";
696 if (didescItem.HasMember(itemName)) {
697 if (didescItem[itemName].IsString()) {
698 didesc->treename = didescItem[itemName].GetString();
700 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
705 didesc->treename =
m[2];
708 itemName =
"fileregex";
709 if (didescItem.HasMember(itemName)) {
710 if (didescItem[itemName].IsString()) {
711 if (didesc->getNumberInputfiles() == 0) {
712 didesc->setFilenamesRegex(didescItem[itemName].GetString());
715 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string!", itemName);
719 if (didesc->getNumberInputfiles() == 0) {
720 didesc->setFilenamesRegex(mFilenameRegexPtr);
724 itemName =
"resfiles";
725 if (didescItem.HasMember(itemName)) {
726 if (didescItem[itemName].IsString()) {
727 fileName = didescItem[itemName].GetString();
728 if (fileName.size() && fileName[0] ==
'@') {
729 didesc->setInputfilesFile(fileName.erase(0, 1));
731 if (didesc->getFilenamesRegexString().empty() ||
732 std::regex_match(fileName, didesc->getFilenamesRegex())) {
736 }
else if (didescItem[itemName].IsArray()) {
737 auto fns = didescItem[itemName].GetArray();
738 for (
auto& fn : fns) {
739 if (didesc->getFilenamesRegexString().empty() ||
740 std::regex_match(fn.GetString(), didesc->getFilenamesRegex())) {
745 LOGP(error,
"Check the JSON document! Item \"{}\" must be a string or an array!", itemName);
749 didesc->setInputfilesFile(minputfilesFilePtr);
753 if (didesc->fillInputfiles() > 0) {
754 mdataInputDescriptors.emplace_back(didesc);
757 LOGP(info,
"This DataInputDescriptor is ignored because its file list is empty!");
759 mAlienSupport &= didesc->isAlienSupportOn();
787 for (
auto didesc : mdataInputDescriptors) {
788 if (didesc->matcher->match(dh, context)) {
802 didesc = mdefaultDataInputDescriptor;
813 didesc = mdefaultDataInputDescriptor;
824 didesc = mdefaultDataInputDescriptor;
832 std::string treename;
837 treename = didesc->treename;
842 didesc = mdefaultDataInputDescriptor;
846 auto result = didesc->readTree(outputs, dh,
counter, numTF, treename, totalSizeCompressed, totalSizeUncompressed);
853 for (
auto didesc : mdataInputDescriptors) {
854 didesc->closeInputFile();
858bool DataInputDirector::isValid()
862 for (
auto didesc : mdataInputDescriptors) {
863 status &= didesc->getNumberInputfiles() == numberFiles;
872 for (
auto didesc : mdataInputDescriptors) {
873 status &= (didesc->getNumberInputfiles() <=
counter);
881 LOGP(info,
"DataInputDirector");
882 LOGP(info,
" Default input files file : {}", minputfilesFile);
883 LOGP(info,
" Default file name regex : {}", mFilenameRegex);
884 LOGP(info,
" Default file names : {}", mdefaultInputFiles.size());
885 for (
auto const& fn : mdefaultInputFiles) {
886 LOGP(info,
" {} {}", fn->fileName, fn->numberOfTimeFrames);
888 LOGP(info,
" Default DataInputDescriptor:");
889 mdefaultDataInputDescriptor->
printOut();
891 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