30#include <unordered_map>
44static std::string &GetCodeToJit()
46 static std::string code;
50static bool ContainsLeaf(
const std::set<TLeaf *> &leaves,
TLeaf *leaf)
52 return (leaves.find(leaf) != leaves.end());
58static void UpdateList(std::set<std::string> &bNamesReg, ColumnNames_t &bNames,
const std::string &branchName,
59 const std::string &friendName)
62 if (!friendName.empty()) {
64 const auto friendBName = friendName +
"." + branchName;
65 if (bNamesReg.insert(friendBName).second)
66 bNames.push_back(friendBName);
69 if (bNamesReg.insert(branchName).second)
70 bNames.push_back(branchName);
75static void UpdateList(std::set<std::string> &bNamesReg, ColumnNames_t &bNames,
const std::string &branchName,
76 const std::string &friendName, std::set<TLeaf *> &foundLeaves,
TLeaf *leaf,
bool allowDuplicates)
78 const bool canAdd = allowDuplicates ? true : !ContainsLeaf(foundLeaves, leaf);
83 UpdateList(bNamesReg, bNames, branchName, friendName);
85 foundLeaves.insert(leaf);
88static void ExploreBranch(
TTree &t, std::set<std::string> &bNamesReg, ColumnNames_t &bNames,
TBranch *
b,
89 std::string prefix, std::string &friendName)
91 for (
auto sb : *
b->GetListOfBranches()) {
93 auto subBranchName = std::string(subBranch->
GetName());
94 auto fullName = prefix + subBranchName;
96 std::string newPrefix;
98 newPrefix = fullName +
".";
100 ExploreBranch(t, bNamesReg, bNames, subBranch, newPrefix, friendName);
103 UpdateList(bNamesReg, bNames, fullName, friendName);
106 UpdateList(bNamesReg, bNames, subBranchName, friendName);
110static void GetBranchNamesImpl(
TTree &t, std::set<std::string> &bNamesReg, ColumnNames_t &bNames,
111 std::set<TTree *> &analysedTrees, std::string &friendName,
bool allowDuplicates)
113 std::set<TLeaf *> foundLeaves;
114 if (!analysedTrees.insert(&t).second) {
123 std::string err(
"GetBranchNames: error in opening the tree ");
125 throw std::runtime_error(err);
128 for (
auto b : *branches) {
130 const auto branchName = std::string(branch->
GetName());
134 if (listOfLeaves->GetEntries() == 1) {
135 auto leaf =
static_cast<TLeaf *
>(listOfLeaves->At(0));
136 const auto leafName = std::string(leaf->
GetName());
137 if (leafName == branchName) {
138 UpdateList(bNamesReg, bNames, branchName, friendName, foundLeaves, leaf, allowDuplicates);
142 for (
auto leaf : *listOfLeaves) {
143 auto castLeaf =
static_cast<TLeaf *
>(leaf);
144 const auto leafName = std::string(leaf->
GetName());
145 const auto fullName = branchName +
"." + leafName;
146 UpdateList(bNamesReg, bNames, fullName, friendName, foundLeaves, castLeaf, allowDuplicates);
150 ExploreBranch(t, bNamesReg, bNames, branch, branchName +
".", friendName);
151 UpdateList(bNamesReg, bNames, branchName, friendName);
156 bool dotIsImplied =
false;
159 throw std::runtime_error(
"GetBranchNames: unsupported branch type");
161 if (be->GetType() == 3 || be->GetType() == 4)
164 if (dotIsImplied || branchName.back() ==
'.')
165 ExploreBranch(t, bNamesReg, bNames, branch,
"", friendName);
167 ExploreBranch(t, bNamesReg, bNames, branch, branchName +
".", friendName);
169 UpdateList(bNamesReg, bNames, branchName, friendName);
179 for (
auto friendTreeObj : *friendTrees) {
184 if (alias !=
nullptr)
185 frName = std::string(alias);
187 frName = std::string(friendTree->GetName());
189 GetBranchNamesImpl(*friendTree, bNamesReg, bNames, analysedTrees, frName, allowDuplicates);
193static void ThrowIfPoolSizeChanged(
unsigned int nSlots)
196 const bool isSingleThreadRun = (poolSize == 0 && nSlots == 1);
197 if (!isSingleThreadRun && poolSize != nSlots) {
198 std::string msg =
"RLoopManager::Run: when the RDataFrame was constructed the size of the thread pool was " +
199 std::to_string(nSlots) +
", but when starting the event loop it was " +
200 std::to_string(poolSize) +
".";
201 if (poolSize > nSlots)
202 msg +=
" Maybe EnableImplicitMT() was called after the RDataFrame was constructed?";
204 msg +=
" Maybe DisableImplicitMT() was called after the RDataFrame was constructed?";
205 throw std::runtime_error(msg);
217struct MaxTreeSizeRAII {
220 MaxTreeSizeRAII() : fOldMaxTreeSize(
TTree::GetMaxTreeSize())
234 std::set<std::string> bNamesSet;
235 ColumnNames_t bNames;
236 std::set<TTree *> analysedTrees;
237 std::string emptyFrName =
"";
238 GetBranchNamesImpl(t, bNamesSet, bNames, analysedTrees, emptyFrName, allowDuplicates);
242RLoopManager::RLoopManager(
TTree *
tree,
const ColumnNames_t &defaultBranches)
243 : fTree(std::shared_ptr<
TTree>(
tree, [](
TTree *) {})), fDefaultColumns(defaultBranches),
258 fDataSource(std::move(ds))
266 auto friends =
fTree->GetListOfFriends();
269 for (
auto friendElObj : *friends) {
271 auto friendTree = friendEl->
GetTree();
272 if (friendTree && friendTree->GetTreeIndex()) {
273 std::string err =
fTree->GetName();
274 err +=
" has a friend, \"";
275 err += friendTree->GetName();
276 err +=
"\", which has an index. This is not supported.";
277 throw std::runtime_error(err);
291 std::vector<std::pair<ULong64_t, ULong64_t>> entryRanges;
299 entryRanges.emplace_back(start, end);
304 auto genFunction = [
this, &slotStack](
const std::pair<ULong64_t, ULong64_t> &range) {
305 auto slot = slotStack.
GetSlot();
308 for (
auto currEntry = range.first; currEntry < range.second; ++currEntry) {
314 std::cerr <<
"RDataFrame::Run: event loop was interrupted\n";
322 pool.
Foreach(genFunction, entryRanges);
337 std::cerr <<
"RDataFrame::Run: event loop was interrupted\n";
350 auto tp = std::make_unique<ROOT::TTreeProcessorMT>(*
fTree, entryList,
fNSlots);
352 std::atomic<ULong64_t> entryCount(0ull);
354 tp->Process([
this, &slotStack, &entryCount](
TTreeReader &
r) ->
void {
355 auto slot = slotStack.
GetSlot();
357 const auto entryRange =
r.GetEntriesRange();
358 const auto nEntries = entryRange.second - entryRange.first;
359 auto count = entryCount.fetch_add(nEntries);
367 std::cerr <<
"RDataFrame::Run: event loop was interrupted\n";
381 if (0 ==
fTree->GetEntriesFast())
393 std::cerr <<
"RDataFrame::Run: event loop was interrupted\n";
398 throw std::runtime_error(
"An error was encountered while processing the data. TTreeReader status code is: " +
399 std::to_string(
r.GetEntryStatus()));
414 for (
const auto &range : ranges) {
415 auto end = range.second;
424 std::cerr <<
"RDataFrame::Run: event loop was interrupted\n";
443 auto runOnRange = [
this, &slotStack](
const std::pair<ULong64_t, ULong64_t> &range) {
444 const auto slot = slotStack.
GetSlot();
447 const auto end = range.second;
449 for (
auto entry = range.first; entry < end; ++entry) {
456 std::cerr <<
"RDataFrame::Run: event loop was interrupted\n";
466 while (!ranges.empty()) {
467 pool.
Foreach(runOnRange, ranges);
479 actionPtr->Run(slot, entry);
481 namedFilterPtr->CheckFilters(slot, entry);
494 ptr->InitSlot(
r, slot);
496 ptr->InitSlot(
r, slot);
532 ptr->ResetChildrenCount();
534 ptr->ResetChildrenCount();
544 ptr->FinalizeSlot(slot);
546 ptr->ClearTask(slot);
553 const std::string code = std::move(GetCodeToJit());
569 actionPtr->TriggerChildrenCount();
571 namedFilterPtr->TriggerChildrenCount();
579 MaxTreeSizeRAII ctxtmts;
658 fPtr->FillReport(rep);
663 GetCodeToJit().append(code);
668 if (everyNEvents == 0ull)
676 std::vector<std::string>
filters;
678 auto name = (filter->HasName() ? filter->GetName() :
"Unnamed Filter");
686 std::vector<RDFInternal::RActionBase *> actions;
703 auto thisNode = std::make_shared<ROOT::Internal::RDF::GraphDrawing::GraphNode>(
name);
705 thisNode->SetCounter(0);
unsigned long long ULong64_t
RLoopManager(TTree *tree, const ColumnNames_t &defaultBranches)
unsigned int fNRuns
Number of event loops run.
bool CheckFilters(unsigned int, Long64_t) final
void EvalChildrenCounts()
Trigger counting of number of children nodes for each node of the functional graph.
void CleanUpNodes()
Perform clean-up operations. To be called at the end of each event loop.
void RunEmptySource()
Run event loop with no source files, in sequence.
void Report(ROOT::RDF::RCutFlowReport &rep) const final
Call FillReport on all booked filters.
std::vector< RFilterBase * > fBookedNamedFilters
Contains a subset of fBookedFilters, i.e. only the named filters.
void RunEmptySourceMT()
Run event loop with no source files, in parallel.
const ColumnNames_t & GetBranchNames()
Return all valid TTree::Branch names (caching results for subsequent calls).
void ToJitExec(const std::string &) const
bool fMustRunNamedFilters
std::shared_ptr< TTree > fTree
Shared pointer to the input TTree.
const ULong64_t fNEmptyEntries
void RunTreeReader()
Run event loop over one or multiple ROOT files, in sequence.
std::vector< RDFInternal::RActionBase * > GetAllActions()
For all the actions, either booked or run.
void CleanUpTask(unsigned int slot)
Perform clean-up operations. To be called at the end of each task execution.
std::vector< RDFInternal::RActionBase * > fRunActions
Non-owning pointers to actions already run.
void Run()
Start the event loop with a different mechanism depending on IMT/no IMT, data source/no data source.
std::vector< RRangeBase * > fBookedRanges
std::vector< TCallback > fCallbacks
Registered callbacks.
void RunAndCheckFilters(unsigned int slot, Long64_t entry)
Execute actions and make sure named filters are called for each event.
std::vector< RFilterBase * > fBookedFilters
void CheckIndexedFriends()
std::vector< RDFInternal::RActionBase * > fBookedActions
Non-owning pointers to actions to be run.
std::shared_ptr< ROOT::Internal::RDF::GraphDrawing::GraphNode > GetGraph()
const ELoopType fLoopType
The kind of event loop that is going to be run (e.g. on ROOT files, on no files)
ColumnNames_t fValidBranchNames
Cache of the tree/chain branch names. Never access directy, always use GetBranchNames().
const unsigned int fNSlots
const ColumnNames_t & GetDefaultColumnNames() const
Return the list of default columns – empty if none was provided when constructing the RDataFrame.
unsigned int GetNSlots() const
std::vector< TOneTimeCallback > fCallbacksOnce
Registered callbacks to invoke just once before running the loop.
void RunDataSourceMT()
Run event loop over data accessed through a DataSource, in parallel.
std::vector< std::string > GetFiltersNames()
For each booked filter, returns either the name or "Unnamed Filter".
const std::unique_ptr< RDataSource > fDataSource
Owning pointer to a data-source object. Null if no data-source.
const ColumnNames_t fDefaultColumns
void Book(RDFInternal::RActionBase *actionPtr)
void InitNodeSlots(TTreeReader *r, unsigned int slot)
Build TTreeReaderValues for all nodes This method loops over all filters, actions and other booked ob...
void RegisterCallback(ULong64_t everyNEvents, std::function< void(unsigned int)> &&f)
void RunDataSource()
Run event loop over data accessed through a DataSource, in sequence.
void Jit()
Add RDF nodes that require just-in-time compilation to the computation graph.
void RunTreeProcessorMT()
Run event loop over one or multiple ROOT files, in parallel.
void Deregister(RDFInternal::RActionBase *actionPtr)
void InitNodes()
Initialize all nodes of the functional graph before running the event loop.
unsigned int fNStopsReceived
Number of times that a children node signaled to stop processing entries.
unsigned int fNChildren
Number of nodes of the functional graph hanging from this object.
This is an helper class to allow to pick a slot resorting to a map indexed by thread ids.
void ReturnSlot(unsigned int slotNumber)
This class provides a simple interface to execute the same task multiple times in parallel,...
void Foreach(F func, unsigned nTimes, unsigned nChunks=0)
Execute func (with no arguments) nTimes in parallel.
A Branch for the case of an object.
A TTree is a list of TBranches.
TObjArray * GetListOfLeaves()
A List of entry numbers in a TTree or TChain.
A TFriendElement TF describes a TTree object TF in a file.
virtual TTree * GetTree()
Return pointer to friend TTree.
A TLeaf describes individual elements of a TBranch See TBranch structure in TTree.
virtual const char * GetName() const
Returns name of object.
A simple, robust and fast interface to read values from ROOT columnar datasets such as TTree,...
@ kEntryNotFound
the tree entry number does not exist
A TTree represents a columnar dataset.
virtual TBranch * FindBranch(const char *name)
Return the branch that correspond to the path 'branchname', which can include the name of the tree or...
virtual TBranch * GetBranch(const char *name)
Return pointer to the branch with the given name in this tree or its friends.
static void SetMaxTreeSize(Long64_t maxsize=100000000000LL)
Set the maximum size in bytes of a Tree file (static function).
virtual TObjArray * GetListOfBranches()
virtual TTree * GetTree() const
virtual TList * GetListOfFriends() const
virtual const char * GetFriendAlias(TTree *) const
If the 'tree' is a friend, this method returns its alias name.
ColumnNames_t GetBranchNames(TTree &t, bool allowDuplicates=true)
Get all the branches names, including the ones of the friend trees.
Long64_t InterpreterCalc(const std::string &code, const std::string &context)
void function(const Char_t *name_, T fun, const Char_t *docstring=0)
tbb::task_arena is an alias of tbb::interface7::task_arena, which doesn't allow to forward declare tb...
Bool_t IsImplicitMTEnabled()
Returns true if the implicit multi-threading in ROOT is enabled.
UInt_t GetThreadPoolSize()
Returns the size of ROOT's thread pool.