Logo ROOT   6.16/01
Reference Guide
TTreeProcessorMP.hxx
Go to the documentation of this file.
1/* @(#)root/multiproc:$Id$ */
2// Author: Enrico Guiraud July 2015
3// Modified: G Ganis Jan 2017
4
5/*************************************************************************
6 * Copyright (C) 1995-2000, Rene Brun and Fons Rademakers. *
7 * All rights reserved. *
8 * *
9 * For the licensing terms see $ROOTSYS/LICENSE. *
10 * For the list of contributors see $ROOTSYS/README/CREDITS. *
11 *************************************************************************/
12
13#ifndef ROOT_TTreeProcessorMP
14#define ROOT_TTreeProcessorMP
15
16#include "MPCode.h"
17#include "MPSendRecv.h"
18#include "PoolUtils.h"
19#include "TChain.h"
20#include "TChainElement.h"
21#include "TError.h"
22#include "TFileCollection.h"
23#include "TFileInfo.h"
24#include "THashList.h"
25#include "TMPClient.h"
26#include "TMPWorkerTree.h"
27#include "TSelector.h"
28#include "TTreeReader.h"
29#include <algorithm> //std::generate
30#include <numeric> //std::iota
31#include <string>
32#include <type_traits> //std::result_of, std::enable_if
33#include <functional> //std::reference_wrapper
34#include <vector>
35
36namespace ROOT {
37
38class TTreeProcessorMP : private TMPClient {
39public:
40 explicit TTreeProcessorMP(unsigned nWorkers = 0); //default number of workers is the number of processors
41 ~TTreeProcessorMP() = default;
42 //it doesn't make sense for a TTreeProcessorMP to be copied
43 TTreeProcessorMP(const TTreeProcessorMP &) = delete;
44 TTreeProcessorMP &operator=(const TTreeProcessorMP &) = delete;
45
46 /// \brief Process a TTree dataset with a functor
47 /// \tparam F functor returning a pointer to TObject or inheriting classes and
48 /// taking a TTreeReader& (both enforced at compile-time)
49 ///
50 /// Dataset definition:
51 /// \param[in] fileNames vector of strings with the paths of the files with the TTree to process
52 /// \param[in] fileName string with the path of the files with the TTree to process
53 /// \param[in] collection TFileCollection with the files with the TTree to process
54 /// \param[in] chain TChain with the files with the TTree to process
55 /// \param[in] tree TTree to process
56 ///
57 /// \param[in] entries TEntryList to filter the dataset
58 /// \param[in] treeName Name of the TTree to process
59 /// \param[in] nToProcess Number of entries to process (0 means all)
60 /// \param[in] jFirst First entry to process (0 means the first of the first file)
61 ///
62 template<class F> auto Process(const std::vector<std::string>& fileNames, F procFunc, TEntryList &entries,
63 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
64 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
65 template<class F> auto Process(const std::string& fileName, F procFunc, TEntryList &entries,
66 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
67 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
68 template<class F> auto Process(TFileCollection& collection, F procFunc, TEntryList &entries,
69 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
70 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
71 template<class F> auto Process(TChain& chain, F procFunc, TEntryList &entries,
72 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
73 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
74 template<class F> auto Process(TTree& tree, F procFunc, TEntryList &entries,
75 ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
76 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
77
78 /// \brief Process a TTree dataset with a functor: version without entry list
79 /// \tparam F functor returning a pointer to TObject or inheriting classes and
80 /// taking a TTreeReader& (both enforced at compile-time)
81 ///
82 /// Dataset definition:
83 /// \param[in] fileNames vector of strings with the paths of the files with the TTree to process
84 /// \param[in] fileName string with the path of the files with the TTree to process
85 /// \param[in] collection TFileCollection with the files with the TTree to process
86 /// \param[in] chain TChain with the files with the TTree to process
87 /// \param[in] tree TTree to process
88 ///
89 /// \param[in] treeName Name of the TTree to process
90 /// \param[in] nToProcess Number of entries to process (0 means all)
91 /// \param[in] jFirst First entry to process (0 means the first of the first file)
92 ///
93 template<class F> auto Process(const std::vector<std::string>& fileNames, F procFunc,
94 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
95 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
96 template<class F> auto Process(const std::string& fileName, F procFunc,
97 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
98 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
99 template<class F> auto Process(TFileCollection& files, F procFunc,
100 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
101 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
102 template<class F> auto Process(TChain& files, F procFunc,
103 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
104 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
105 template<class F> auto Process(TTree& tree, F procFunc, ULong64_t nToProcess = 0, ULong64_t jFirst = 0)
106 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
107
108
109 /// \brief Process a TTree dataset with a selector
110 ///
111 /// Dataset definition:
112 /// \param[in] fileNames vector of strings with the paths of the files with the TTree to process
113 /// \param[in] fileName string with the path of the files with the TTree to process
114 /// \param[in] collection TFileCollection with the files with the TTree to process
115 /// \param[in] chain TChain with the files with the TTree to process
116 /// \param[in] tree TTree to process
117 ///
118 /// \param[in] selector Instance of TSelector to be applied to the dataset
119 /// \param[in] entries TEntryList to filter the dataset
120 /// \param[in] treeName Name of the TTree to process
121 /// \param[in] nToProcess Number of entries to process (0 means all)
122 /// \param[in] jFirst First entry to process (0 means the first of the first file)
123 ///
124 // these versions require a TSelector
125 TList* Process(const std::vector<std::string>& fileNames, TSelector& selector, TEntryList &entries,
126 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
127 TList* Process(const std::string &fileName, TSelector& selector, TEntryList &entries,
128 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
129 TList* Process(TFileCollection& files, TSelector& selector, TEntryList &entries,
130 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
131 TList* Process(TChain& files, TSelector& selector, TEntryList &entries,
132 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
133 TList* Process(TTree& tree, TSelector& selector, TEntryList &entries,
134 ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
135
136
137 /// \brief Process a TTree dataset with a selector: version without entry list
138 ///
139 /// Dataset definition:
140 /// \param[in] fileNames vector of strings with the paths of the files with the TTree to process
141 /// \param[in] fileName string with the path of the files with the TTree to process
142 /// \param[in] collection TFileCollection with the files with the TTree to process
143 /// \param[in] chain TChain with the files with the TTree to process
144 /// \param[in] tree TTree to process
145 ///
146 /// \param[in] selector Instance of TSelector to be applied to the dataset
147 /// \param[in] treeName Name of the TTree to process
148 /// \param[in] nToProcess Number of entries to process (0 means all)
149 /// \param[in] jFirst First entry to process (0 means the first of the first file)
150 ///
151 // these versions require a TSelector
152 TList* Process(const std::vector<std::string>& fileNames, TSelector& selector,
153 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
154 TList* Process(const std::string &fileName, TSelector& selector,
155 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
156 TList* Process(TFileCollection& files, TSelector& selector,
157 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
158 TList* Process(TChain& files, TSelector& selector,
159 const std::string& treeName = "", ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
160 TList* Process(TTree& tree, TSelector& selector, ULong64_t nToProcess = 0, ULong64_t jFirst = 0);
161
162 void SetNWorkers(unsigned n) { TMPClient::SetNWorkers(n); }
163 unsigned GetNWorkers() const { return TMPClient::GetNWorkers(); }
164
165private:
166 template<class T> void Collect(std::vector<T> &reslist);
167 template<class T> void HandlePoolCode(MPCodeBufPair &msg, TSocket *sender, std::vector<T> &reslist);
168
169 void FixLists(std::vector<TObject*> &lists);
170 void Reset();
171 void ReplyToIdle(TSocket *s);
172
173 unsigned fNProcessed; ///< number of arguments already passed to the workers
174 unsigned fNToProcess; ///< total number of arguments to pass to the workers
175
176 /// A collection of the types of tasks that TTreeProcessorMP can execute.
177 /// It is used to interpret in the right way and properly reply to the
178 /// messages received (see, for example, TTreeProcessorMP::HandleInput)
179 enum class ETask : unsigned char {
180 kNoTask, ///< no task is being executed
181 kProcByRange, ///< a Process method is being executed and each worker will process a certain range of each file
182 kProcByFile ///< a Process method is being executed and each worker will process a different file
183 };
184
185 ETask fTaskType = ETask::kNoTask; ///< the kind of task that is being executed, if any
186};
187
188template<class F>
189auto TTreeProcessorMP::Process(const std::vector<std::string>& fileNames, F procFunc, TEntryList &entries,
190 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
191 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
192{
193 using retType = typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
194 static_assert(std::is_constructible<TObject*, retType>::value,
195 "procFunc must return a pointer to a class inheriting from TObject,"
196 " and must take a reference to TTreeReader as the only argument");
197
198 // Warn for yet unimplemented functionality
199 if (jFirst > 0) {
200 Warning("Process", "support for generic 'first entry' (jFirst > 0) not implemented yet - ignoring");
201 jFirst = 0;
202 }
203
204 //prepare environment
205 Reset();
206 unsigned nWorkers = GetNWorkers();
207
208 // Check th entry list
209 TEntryList *elist = (entries.IsValid()) ? &entries : nullptr;
210 //fork
211 TMPWorkerTreeFunc<F> worker(procFunc, fileNames, elist, treeName, nWorkers, nToProcess, jFirst);
212 bool ok = Fork(worker);
213 if(!ok) {
214 Error("TTreeProcessorMP::Process", "[E][C] Could not fork. Aborting operation.");
215 return nullptr;
216 }
217
218
219 if(fileNames.size() < nWorkers) {
220 //TTree entry granularity. For each file, we divide entries equally between workers
221 fTaskType = ETask::kProcByRange;
222 //Tell workers to start processing entries
223 fNToProcess = nWorkers*fileNames.size(); //this is the total number of ranges that will be processed by all workers cumulatively
224 std::vector<unsigned> args(nWorkers);
225 std::iota(args.begin(), args.end(), 0);
226 fNProcessed = Broadcast(MPCode::kProcRange, args);
227 if(fNProcessed < nWorkers)
228 Error("TTreeProcessorMP::Process", "[E][C] There was an error while sending tasks to workers. Some entries might not be processed.");
229 } else {
230 //file granularity. each worker processes one whole file as a single task
231 fTaskType = ETask::kProcByFile;
232 fNToProcess = fileNames.size();
233 std::vector<unsigned> args(nWorkers);
234 std::iota(args.begin(), args.end(), 0);
235 fNProcessed = Broadcast(MPCode::kProcFile, args);
236 if(fNProcessed < nWorkers)
237 Error("TTreeProcessorMP::Process", "[E][C] There was an error while sending tasks to workers. Some entries might not be processed.");
238 }
239
240 //collect results, distribute new tasks
241 std::vector<TObject*> reslist;
242 Collect(reslist);
243
244 //merge
245 PoolUtils::ReduceObjects<TObject *> redfunc;
246 auto res = redfunc(reslist);
247
248 //clean-up and return
249 ReapWorkers();
250 fTaskType = ETask::kNoTask;
251 return static_cast<retType>(res);
252}
253
254
255template<class F>
256auto TTreeProcessorMP::Process(const std::string& fileName, F procFunc, TEntryList &entries,
257 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
258 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
259{
260 std::vector<std::string> singleFileName(1, fileName);
261 return Process(singleFileName, procFunc, entries, treeName, nToProcess, jFirst);
262}
263
264
265template<class F>
266auto TTreeProcessorMP::Process(TFileCollection& files, F procFunc, TEntryList &entries,
267 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
268 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
269{
270 std::vector<std::string> fileNames(files.GetNFiles());
271 unsigned count = 0;
272 for(auto f : *static_cast<THashList*>(files.GetList()))
273 fileNames[count++] = static_cast<TFileInfo*>(f)->GetCurrentUrl()->GetUrl();
274
275 return Process(fileNames, procFunc, entries, treeName, nToProcess, jFirst);
276}
277
278
279template<class F>
280auto TTreeProcessorMP::Process(TChain& files, F procFunc, TEntryList &entries,
281 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
282 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
283{
284 TObjArray* filelist = files.GetListOfFiles();
285 std::vector<std::string> fileNames(filelist->GetEntries());
286 unsigned count = 0;
287 for(auto f : *filelist)
288 fileNames[count++] = f->GetTitle();
289
290 return Process(fileNames, procFunc, entries, treeName, nToProcess, jFirst);
291}
292
293
294template<class F>
295auto TTreeProcessorMP::Process(TTree& tree, F procFunc, TEntryList &entries,
296 ULong64_t nToProcess, ULong64_t jFirst)
297 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
298{
299 using retType = typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type;
300 static_assert(std::is_constructible<TObject*, retType>::value, "procFunc must return a pointer to a class inheriting from TObject, and must take a reference to TTreeReader as the only argument");
301
302 // Warn for yet unimplemented functionality
303 if (jFirst > 0) {
304 Warning("Process", "support for generic 'first entry' (jFirst > 0) not implemented yet - ignoring");
305 jFirst = 0;
306 }
307
308 //prepare environment
309 Reset();
310 unsigned nWorkers = GetNWorkers();
311
312 // Check th entry list
313 TEntryList *elist = (entries.IsValid()) ? &entries : nullptr;
314 //fork
315 TMPWorkerTreeFunc<F> worker(procFunc, &tree, elist, nWorkers, nToProcess, jFirst);
316 bool ok = Fork(worker);
317 if(!ok) {
318 Error("TTreeProcessorMP::Process", "[E][C] Could not fork. Aborting operation.");
319 return nullptr;
320 }
321
322 //divide entries equally between workers
323 fTaskType = ETask::kProcByRange;
324
325 //tell workers to start processing entries
326 fNToProcess = nWorkers; //this is the total number of ranges that will be processed by all workers cumulatively
327 std::vector<unsigned> args(nWorkers);
328 std::iota(args.begin(), args.end(), 0);
329 fNProcessed = Broadcast(MPCode::kProcTree, args);
330 if(fNProcessed < nWorkers)
331 Error("TTreeProcessorMP::Process", "[E][C] There was an error while sending tasks to workers. Some entries might not be processed.");
332
333 //collect results, distribute new tasks
334 std::vector<TObject*> reslist;
335 Collect(reslist);
336
337 //merge
338 PoolUtils::ReduceObjects<TObject *> redfunc;
339 auto res = redfunc(reslist);
340
341 //clean-up and return
342 ReapWorkers();
343 fTaskType = ETask::kNoTask;
344 return static_cast<retType>(res);
345}
346
347
348///
349/// No TEntryList versions of generic processor
350///
351
352template<class F>
353auto TTreeProcessorMP::Process(const std::vector<std::string>& fileNames, F procFunc,
354 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
355 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
356{
357 TEntryList noelist;
358 return Process(fileNames, procFunc, noelist, treeName, nToProcess, jFirst);
359}
360
361
362template<class F>
363auto TTreeProcessorMP::Process(const std::string& fileName, F procFunc,
364 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
365 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
366{
367 TEntryList noelist;
368 return Process(fileName, procFunc, noelist, treeName, nToProcess, jFirst);
369}
370
371
372template<class F>
373auto TTreeProcessorMP::Process(TFileCollection& files, F procFunc,
374 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
375 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
376{
377 TEntryList noelist;
378 return Process(files, procFunc, noelist, treeName, nToProcess, jFirst);
379}
380
381
382template<class F>
383auto TTreeProcessorMP::Process(TChain& files, F procFunc,
384 const std::string& treeName, ULong64_t nToProcess, ULong64_t jFirst)
385 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
386{
387 TEntryList noelist;
388 return Process(files, procFunc, noelist, treeName, nToProcess, jFirst);
389}
390
391
392template<class F>
393auto TTreeProcessorMP::Process(TTree& tree, F procFunc,
394 ULong64_t nToProcess, ULong64_t jFirst)
395 -> typename std::result_of<F(std::reference_wrapper<TTreeReader>)>::type
396{
397 TEntryList noelist;
398 return Process(tree, procFunc, noelist, nToProcess, jFirst);
399}
400
401//////////////////////////////////////////////////////////////////////////
402/// Handle message and reply to the worker
403template<class T>
404void TTreeProcessorMP::HandlePoolCode(MPCodeBufPair &msg, TSocket *s, std::vector<T> &reslist)
405{
406 unsigned code = msg.first;
407 if (code == MPCode::kIdling) {
408 ReplyToIdle(s);
409 } else if(code == MPCode::kProcResult) {
410 if(msg.second != nullptr)
411 reslist.push_back(std::move(ReadBuffer<T>(msg.second.get())));
412 MPSend(s, MPCode::kShutdownOrder);
413 } else if(code == MPCode::kProcError) {
414 const char *str = ReadBuffer<const char*>(msg.second.get());
415 Error("TTreeProcessorMP::HandlePoolCode", "[E][C] a worker encountered an error: %s\n"
416 "Continuing execution ignoring these entries.", str);
417 ReplyToIdle(s);
418 delete [] str;
419 } else {
420 // UNKNOWN CODE
421 Error("TTreeProcessorMP::HandlePoolCode", "[W][C] unknown code received from server. code=%d", code);
422 }
423}
424
425//////////////////////////////////////////////////////////////////////////
426/// Listen for messages sent by the workers and call the appropriate handler function.
427/// TTreeProcessorMP::HandlePoolCode is called on messages with a code < 1000 and
428/// TMPClient::HandleMPCode is called on messages with a code >= 1000.
429template<class T>
430void TTreeProcessorMP::Collect(std::vector<T> &reslist)
431{
432 TMonitor &mon = GetMonitor();
433 mon.ActivateAll();
434 while (mon.GetActive() > 0) {
435 TSocket *s = mon.Select();
436 MPCodeBufPair msg = MPRecv(s);
437 if (msg.first == MPCode::kRecvError) {
438 Error("TTreeProcessorMP::Collect", "[E][C] Lost connection to a worker");
439 Remove(s);
440 } else if (msg.first < 1000)
441 HandlePoolCode(msg, s, reslist);
442 else
443 HandleMPCode(msg, s);
444 }
445}
446
447} // ROOT namespace
448
449#endif
MPCodeBufPair
std::pair< unsigned, std::unique_ptr< TBufferFile > > MPCodeBufPair
An std::pair that wraps the code and optional object contained in a message.
Definition: MPSendRecv.h:31
MPRecv
MPCodeBufPair MPRecv(TSocket *s)
Receive message from a socket.
Definition: MPSendRecv.cxx:54
MPSend
int MPSend(TSocket *s, unsigned code)
Send a message with the specified code on the specified socket.
Definition: MPSendRecv.cxx:32
f
#define f(i)
Definition: RSha256.hxx:104
ULong64_t
unsigned long long ULong64_t
Definition: RtypesCore.h:70
Error
void Error(const char *location, const char *msgfmt,...)
Warning
void Warning(const char *location, const char *msgfmt,...)
type
int type
Definition: TGX11.cxx:120
ROOT::ExecutorUtils::ReduceObjects
Merge collection of TObjects.
Definition: PoolUtils.h:35
ROOT::TTreeProcessorMP
This class provides an interface to process a TTree dataset in parallel with multi-process technology...
Definition: TTreeProcessorMP.hxx:38
ROOT::TTreeProcessorMP::Process
auto Process(const std::vector< std::string > &fileNames, F procFunc, TEntryList &entries, const std::string &treeName="", ULong64_t nToProcess=0, ULong64_t jFirst=0) -> typename std::result_of< F(std::reference_wrapper< TTreeReader >)>::type
Process a TTree dataset with a functor.
Definition: TTreeProcessorMP.hxx:189
ROOT::TTreeProcessorMP::fNProcessed
unsigned fNProcessed
number of arguments already passed to the workers
Definition: TTreeProcessorMP.hxx:173
ROOT::TTreeProcessorMP::HandlePoolCode
void HandlePoolCode(MPCodeBufPair &msg, TSocket *sender, std::vector< T > &reslist)
Handle message and reply to the worker.
Definition: TTreeProcessorMP.hxx:404
ROOT::TTreeProcessorMP::TTreeProcessorMP
TTreeProcessorMP(const TTreeProcessorMP &)=delete
ROOT::TTreeProcessorMP::TTreeProcessorMP
TTreeProcessorMP(unsigned nWorkers=0)
ROOT::TTreeProcessorMP::Reset
void Reset()
Reset TTreeProcessorMP's state.
Definition: TTreeProcessorMP.cxx:339
ROOT::TTreeProcessorMP::FixLists
void FixLists(std::vector< TObject * > &lists)
Fix list of lists before merging (to avoid errors about duplicated objects)
Definition: TTreeProcessorMP.cxx:323
ROOT::TTreeProcessorMP::Collect
void Collect(std::vector< T > &reslist)
Listen for messages sent by the workers and call the appropriate handler function.
Definition: TTreeProcessorMP.hxx:430
ROOT::TTreeProcessorMP::ETask
ETask
A collection of the types of tasks that TTreeProcessorMP can execute.
Definition: TTreeProcessorMP.hxx:179
ROOT::TTreeProcessorMP::ETask::kNoTask
@ kNoTask
no task is being executed
ROOT::TTreeProcessorMP::ETask::kProcByRange
@ kProcByRange
a Process method is being executed and each worker will process a certain range of each file
ROOT::TTreeProcessorMP::ETask::kProcByFile
@ kProcByFile
a Process method is being executed and each worker will process a different file
ROOT::TTreeProcessorMP::fTaskType
ETask fTaskType
the kind of task that is being executed, if any
Definition: TTreeProcessorMP.hxx:185
ROOT::TTreeProcessorMP::fNToProcess
unsigned fNToProcess
total number of arguments to pass to the workers
Definition: TTreeProcessorMP.hxx:174
ROOT::TTreeProcessorMP::ReplyToIdle
void ReplyToIdle(TSocket *s)
Reply to a worker who is idle.
Definition: TTreeProcessorMP.cxx:350
ROOT::TTreeProcessorMP::GetNWorkers
unsigned GetNWorkers() const
Definition: TTreeProcessorMP.hxx:163
ROOT::TTreeProcessorMP::SetNWorkers
void SetNWorkers(unsigned n)
Definition: TTreeProcessorMP.hxx:162
ROOT::TTreeProcessorMP::operator=
TTreeProcessorMP & operator=(const TTreeProcessorMP &)=delete
TChain
A chain is a collection of files containing TTree objects.
Definition: TChain.h:33
TEntryList
A List of entry numbers in a TTree or TChain.
Definition: TEntryList.h:26
TFileCollection
Class that contains a list of TFileInfo's and accumulated meta data information about its entries.
Definition: TFileCollection.h:39
TFileInfo
Class describing a generic file including meta information.
Definition: TFileInfo.h:38
THashList
THashList implements a hybrid collection class consisting of a hash table and a list to store TObject...
Definition: THashList.h:34
TList
A doubly linked list.
Definition: TList.h:44
TMPClient
Base class for multiprocess applications' clients.
Definition: TMPClient.h:23
TMPClient::GetNWorkers
unsigned GetNWorkers() const
Definition: TMPClient.h:40
TMPClient::HandleMPCode
void HandleMPCode(MPCodeBufPair &msg, TSocket *sender)
Handle messages containing an EMPCode.
Definition: TMPClient.cxx:329
TMPClient::SetNWorkers
void SetNWorkers(unsigned n)
Set the number of workers that will be spawned by the next call to Fork()
Definition: TMPClient.h:39
TMPClient::GetMonitor
TMonitor & GetMonitor()
Definition: TMPClient.h:36
TMPClient::Remove
void Remove(TSocket *s)
Remove a certain socket from the monitor.
Definition: TMPClient.cxx:295
TMPWorkerTreeFunc
Templated derivation of TMPWorkerTree handlign generic function tree processing.
Definition: TMPWorkerTree.h:79
TMonitor::ActivateAll
virtual void ActivateAll()
Activate all de-activated sockets.
Definition: TMonitor.cxx:268
TMonitor::Select
TSocket * Select()
Return pointer to socket for which an event is waiting.
Definition: TMonitor.cxx:322
TMonitor::GetActive
Int_t GetActive(Long_t timeout=-1) const
Return number of sockets in the active list.
Definition: TMonitor.cxx:438
TObjArray
An array of TObjects.
Definition: TObjArray.h:37
TObjArray::GetEntries
Int_t GetEntries() const
Return the number of objects in array (i.e.
Definition: TObjArray.cxx:522
TSelector
A TSelector object is used by the TTree::Draw, TTree::Scan, TTree::Process to navigate in a TTree and...
Definition: TSelector.h:33
TTree
A TTree object has a header with a name and a title.
Definition: TTree.h:71
collection
void collection()
Definition: collection.C:252
n
const Int_t n
Definition: legend1.C:16
F
#define F(x, y, z)
MPCode::kProcFile
@ kProcFile
Tell a TMPWorkerTree which tree to process. The object sent is a TreeInfo.
Definition: MPCode.h:38
MPCode::kRecvError
@ kRecvError
Error while reading from the socket.
Definition: MPCode.h:51
MPCode::kIdling
@ kIdling
We are ready for the next task.
Definition: MPCode.h:35
MPCode::kProcRange
@ kProcRange
Tell a TMPWorkerTree which tree and entries range to process. The object sent is a TreeRangeInfo.
Definition: MPCode.h:39
MPCode::kShutdownOrder
@ kShutdownOrder
Used by the client to tell servers to shutdown.
Definition: MPCode.h:49
MPCode::kProcTree
@ kProcTree
Tell a TMPWorkerTree to process the tree that was passed to it at construction time.
Definition: MPCode.h:40
MPCode::kProcError
@ kProcError
Tell the client there was an error while processing.
Definition: MPCode.h:44
MPCode::kProcResult
@ kProcResult
The message contains the result of the processing of a TTree.
Definition: MPCode.h:42
ROOT
Namespace for new ROOT classes and functions.
Definition: StringConv.hxx:21
TGeoUnit::s
static constexpr double s
Definition: TGeoSystemOfUnits.h:162
tree
Definition: tree.py:1