Logo ROOT   6.10/09
Reference Guide
List of all members | Public Member Functions | Private Types | Private Member Functions | Private Attributes | List of all members
ROOT::TProcessExecutor Class Reference

This class provides a simple interface to execute the same task multiple times in parallel, possibly with different arguments every time.

This mimics the behaviour of python's pool.Map method.

ROOT::TProcessExecutor::Map

This class inherits its interfaces from ROOT::TExecutor
. The two possible usages of the Map method are:

For either signature, func is executed as many times as needed by a pool of fNWorkers workers; the number of workers can be passed to the constructor or set via SetNWorkers. It defaults to the number of cores.
A collection containing the result of each execution is returned.
Note: the user is responsible for the deletion of any object that might be created upon execution of func, returned objects included: ROOT::TProcessExecutor never deletes what it returns, it simply forgets it.
Note: that the usage of ROOT::TProcessExecutor::Map is indicated only when the task to be executed takes more than a few seconds, otherwise the overhead introduced by Map will outrun the benefits of parallel execution on most machines.

Parameters
funca lambda expression, an std::function, a loaded macro, a functor class or a function that takes zero arguments (for the first signature) or one (for the second signature).
argsa standard vector, a ROOT::TSeq of integer type or an initializer list for the second signature. An integer only for the first.

Note: in cases where the function to be executed takes more than zero/one argument but all are fixed except zero/one, the function can be wrapped in a lambda or via std::bind to give it the right signature.
Note: the user should take care of initializing random seeds differently in each process (e.g. using the process id in the seed). Otherwise several parallel executions might generate the same sequence of pseudo-random numbers.

Return value:

An std::vector. The elements in the container will be the objects returned by func.

Examples:

root[] ROOT::TProcessExecutor pool; auto hists = pool.Map(CreateHisto, 10);
root[] ROOT::TProcessExecutor pool(2); auto squares = pool.Map([](int a) { return a*a; }, {1,2,3});

ROOT::TProcessExecutor::MapReduce

This set of methods behaves exactly like Map, but takes an additional function as a third argument. This function is applied to the set of objects returned by the corresponding Map execution to "squash" them to a single object.

Examples:

root[] ROOT::TProcessExecutor pool; auto ten = pool.MapReduce([]() { return 1; }, 10, [](std::vector<int> v) { return std::accumulate(v.begin(), v.end(), 0); })
root[] ROOT::TProcessExecutor pool; auto hist = pool.MapReduce(CreateAndFillHists, 10, PoolUtils::ReduceObjects);

Definition at line 37 of file TProcessExecutor.hxx.

Public Member Functions

 TProcessExecutor (unsigned nWorkers=0)
 Class constructor. More...
 
 TProcessExecutor (const TProcessExecutor &)=delete
 
 ~TProcessExecutor ()=default
 
unsigned GetNWorkers () const
 
template<class F , class Cond = noReferenceCond<F>>
auto Map (F func, unsigned nTimes) -> std::vector< typename std::result_of< F()>::type >
 Execute func (with no arguments) nTimes in parallel. More...
 
template<class F , class INTEGER , class Cond = noReferenceCond<F, INTEGER>>
auto Map (F func, ROOT::TSeq< INTEGER > args) -> std::vector< typename std::result_of< F(INTEGER)>::type >
 Execute func in parallel, taking an element of a sequence as argument. More...
 
template<class F , class T , class Cond = noReferenceCond<F, T>>
auto Map (F func, std::vector< T > &args) -> std::vector< typename std::result_of< F(T)>::type >
 Execute func in parallel, taking an element of an std::vector as argument. More...
 
TProcessExecutoroperator= (const TProcessExecutor &)=delete
 
template<class T , class R >
Reduce (const std::vector< T > &objs, R redfunc)
 "Reduce" an std::vector into a single object by passing a function as the second argument defining the reduction operation. More...
 
void SetNWorkers (unsigned n)
 
- Public Member Functions inherited from ROOT::TExecutor< TProcessExecutor >
 TExecutor ()=default
 
 TExecutor (size_t)
 
auto Map (F func, unsigned nTimes) -> std::vector< typename std::result_of< F()>::type >
 Execute func (with no arguments) nTimes in parallel. More...
 
auto Map (F func, ROOT::TSeq< INTEGER > args) -> std::vector< typename std::result_of< F(INTEGER)>::type >
 Execute func in parallel, taking an element of a sequence as argument. More...
 
auto Map (F func, std::vector< T > &args) -> std::vector< typename std::result_of< F(T)>::type >
 Execute func in parallel, taking an element of an std::vector as argument. More...
 
auto Map (F func, std::initializer_list< T > args) -> std::vector< typename std::result_of< F(T)>::type >
 Execute func in parallel, taking an element of the std::initializer_list as argument. More...
 
auto MapReduce (F func, unsigned nTimes, R redfunc) -> typename std::result_of< F()>::type
 This method behaves just like Map, but an additional redfunc function must be provided. More...
 
auto MapReduce (F func, ROOT::TSeq< INTEGER > args, R redfunc) -> typename std::result_of< F(INTEGER)>::type
 
auto MapReduce (F func, std::vector< T > &args, R redfunc) -> typename std::result_of< F(T)>::type
 
T * MapReduce (F func, std::vector< T * > &args)
 
auto MapReduce (F func, std::initializer_list< T > args, R redfunc) -> typename std::result_of< F(T)>::type
 
T * Reduce (const std::vector< T * > &mergeObjs)
 "Reduce" an std::vector into a single object by using the object's Merge More...
 

Private Types

enum  ETask : unsigned char { ETask::kNoTask, ETask::kMap, ETask::kMapWithArg }
 A collection of the types of tasks that TProcessExecutor can execute. More...
 

Private Member Functions

template<class T >
void Collect (std::vector< T > &reslist)
 Listen for messages sent by the workers and call the appropriate handler function. More...
 
template<class T >
void HandlePoolCode (MPCodeBufPair &msg, TSocket *sender, std::vector< T > &reslist)
 Handle message and reply to the worker. More...
 
void ReplyToFuncResult (TSocket *s)
 Reply to a worker who just sent a result. More...
 
void ReplyToIdle (TSocket *s)
 Reply to a worker who is idle. More...
 
void Reset ()
 Reset TProcessExecutor's state. More...
 
- Private Member Functions inherited from TMPClient
 TMPClient (unsigned nWorkers=0)
 Class constructor. More...
 
 TMPClient (const TMPClient &)=delete
 
 ~TMPClient ()
 Class destructor. More...
 
unsigned Broadcast (unsigned code, unsigned nMessages=0)
 Send a message with the specified code to at most nMessages workers. More...
 
template<class T >
unsigned Broadcast (unsigned code, const std::vector< T > &objs)
 Send a message with a different object to each server. More...
 
template<class T >
unsigned Broadcast (unsigned code, std::initializer_list< T > &objs)
 Send a message with a different object to each server. More...
 
template<class T >
unsigned Broadcast (unsigned code, T obj, unsigned nMessages=0)
 Send a message containing code and obj to each worker, up to a maximum number of nMessages workers. More...
 
void DeActivate (TSocket *s)
 DeActivate a certain socket. More...
 
bool Fork (TMPWorker &server)
 This method forks the ROOT session into fNWorkers children processes. More...
 
bool GetIsParent () const
 
TMonitorGetMonitor ()
 
unsigned GetNWorkers () const
 
void HandleMPCode (MPCodeBufPair &msg, TSocket *sender)
 Handle messages containing an EMPCode. More...
 
TMPClientoperator= (const TMPClient &)=delete
 
void ReapWorkers ()
 Wait on worker processes and remove their pids from fWorkerPids. More...
 
void Remove (TSocket *s)
 Remove a certain socket from the monitor. More...
 
void SetNWorkers (unsigned n)
 Set the number of workers that will be spawned by the next call to Fork() More...
 

Private Attributes

unsigned fNProcessed
 number of arguments already passed to the workers More...
 
unsigned fNToProcess
 total number of arguments to pass to the workers More...
 
ETask fTaskType = ETask::kNoTask
 the kind of task that is being executed, if any More...
 

Additional Inherited Members

- Public Types inherited from ROOT::TExecutor< TProcessExecutor >
using noReferenceCond = typename std::enable_if<"Function can't return a reference" &&!(std::is_reference< typename std::result_of< F(T...)>::type >::value)>::type
 

#include <ROOT/TProcessExecutor.hxx>

Inheritance diagram for ROOT::TProcessExecutor:
[legend]

Member Enumeration Documentation

◆ ETask

enum ROOT::TProcessExecutor::ETask : unsigned char
strongprivate

A collection of the types of tasks that TProcessExecutor can execute.

It is used to interpret in the right way and properly reply to the messages received (see, for example, TProcessExecutor::HandleInput)

Enumerator
kNoTask 

no task is being executed

kMap 

a Map method with no arguments is being executed

kMapWithArg 

a Map method with arguments is being executed

Definition at line 74 of file TProcessExecutor.hxx.

Constructor & Destructor Documentation

◆ TProcessExecutor() [1/2]

ROOT::TProcessExecutor::TProcessExecutor ( unsigned  nWorkers = 0)
explicit

Class constructor.

nWorkers is the number of times this ROOT session will be forked, i.e. the number of workers that will be spawned.

Definition at line 89 of file TProcessExecutor.cxx.

◆ ~TProcessExecutor()

ROOT::TProcessExecutor::~TProcessExecutor ( )
default

◆ TProcessExecutor() [2/2]

ROOT::TProcessExecutor::TProcessExecutor ( const TProcessExecutor )
delete

Member Function Documentation

◆ Collect()

template<class T >
void ROOT::TProcessExecutor::Collect ( std::vector< T > &  reslist)
private

Listen for messages sent by the workers and call the appropriate handler function.

TProcessExecutor::HandlePoolCode is called on messages with a code < 1000 and TMPClient::HandleMPCode is called on messages with a code >= 1000.

Definition at line 229 of file TProcessExecutor.hxx.

◆ GetNWorkers()

unsigned ROOT::TProcessExecutor::GetNWorkers ( ) const
inline

Definition at line 55 of file TProcessExecutor.hxx.

◆ HandlePoolCode()

template<class T >
void ROOT::TProcessExecutor::HandlePoolCode ( MPCodeBufPair msg,
TSocket sender,
std::vector< T > &  reslist 
)
private

Handle message and reply to the worker.

Definition at line 200 of file TProcessExecutor.hxx.

◆ Map() [1/3]

template<class F , class Cond >
auto ROOT::TProcessExecutor::Map ( F  func,
unsigned  nTimes 
) -> std::vector<typename std::result_of<F()>::type>

Execute func (with no arguments) nTimes in parallel.

A vector containg executions' results is returned. Functions that take more than zero arguments can be executed (with fixed arguments) by wrapping them in a lambda or with std::bind.

Definition at line 92 of file TProcessExecutor.hxx.

◆ Map() [2/3]

template<class F , class INTEGER , class Cond >
auto ROOT::TProcessExecutor::Map ( F  func,
ROOT::TSeq< INTEGER >  args 
) -> std::vector<typename std::result_of<F(INTEGER)>::type>

Execute func in parallel, taking an element of a sequence as argument.

A vector containg executions' results is returned.

Definition at line 178 of file TProcessExecutor.hxx.

◆ Map() [3/3]

template<class F , class T , class Cond >
auto ROOT::TProcessExecutor::Map ( F  func,
std::vector< T > &  args 
) -> std::vector<typename std::result_of<F(T)>::type>

Execute func in parallel, taking an element of an std::vector as argument.

A vector containg executions' results is returned.

Definition at line 134 of file TProcessExecutor.hxx.

◆ operator=()

TProcessExecutor& ROOT::TProcessExecutor::operator= ( const TProcessExecutor )
delete

◆ Reduce()

template<class T , class R >
T ROOT::TProcessExecutor::Reduce ( const std::vector< T > &  objs,
R  redfunc 
)

"Reduce" an std::vector into a single object by passing a function as the second argument defining the reduction operation.

Definition at line 190 of file TProcessExecutor.hxx.

◆ ReplyToFuncResult()

void ROOT::TProcessExecutor::ReplyToFuncResult ( TSocket s)
private

Reply to a worker who just sent a result.

If another argument to process exists, tell the worker. Otherwise send a shutdown order.

Definition at line 107 of file TProcessExecutor.cxx.

◆ ReplyToIdle()

void ROOT::TProcessExecutor::ReplyToIdle ( TSocket s)
private

Reply to a worker who is idle.

If another argument to process exists, tell the worker. Otherwise ask for a result

Definition at line 125 of file TProcessExecutor.cxx.

◆ Reset()

void ROOT::TProcessExecutor::Reset ( void  )
private

Reset TProcessExecutor's state.

Definition at line 96 of file TProcessExecutor.cxx.

◆ SetNWorkers()

void ROOT::TProcessExecutor::SetNWorkers ( unsigned  n)
inline

Definition at line 54 of file TProcessExecutor.hxx.

Member Data Documentation

◆ fNProcessed

unsigned ROOT::TProcessExecutor::fNProcessed
private

number of arguments already passed to the workers

Definition at line 68 of file TProcessExecutor.hxx.

◆ fNToProcess

unsigned ROOT::TProcessExecutor::fNToProcess
private

total number of arguments to pass to the workers

Definition at line 69 of file TProcessExecutor.hxx.

◆ fTaskType

ETask ROOT::TProcessExecutor::fTaskType = ETask::kNoTask
private

the kind of task that is being executed, if any

Definition at line 80 of file TProcessExecutor.hxx.


The documentation for this class was generated from the following files: