ROOT's executors: scaling task-level parallelism

Contemporary HEP analysis requires the processing of large quantities of data. With the breakdown of Dennard scaling in the early 2000s¹², and the subsequent decrease in CPU clock speed, alternative solutions were required to satisfy the demands of modern physics analyses.

Historically, this limitation has been addressed by incrementing the number of processing units, or cores, in a processor. This evolution of hardware towards increasing levels of parallelism drives the adoption of parallel workflows in the data analysis software, across many levels of parallelism²⁰, to exploit the possibilities offered by modern architectures.

At the same time, many HEP processes can be classified in generic patterns, or algorithms, reproduced over several analyses, adapted to the specific problems at hand. Our goal is to identify these patterns in the Physics analysis processes, and implement in the ROOT software framework a set of tools to parallelize them across distinct levels, from the most common patterns, such as MapReduce, to the ones that provide more potential speed up, such as the integration of tools for vectorization and the adoption of SIMD types. As part of this effort, new generic classes supporting a task-based parallelization and support for different SIMD intrinsics abstraction libraries have been introduced in ROOT.

This Chapter focuses on task-level parallelism. Section 4.1 familiarizes the reader with the event-processing paradigm, a core concept of HEP analyses, and lays out the ideas behind the implementation of task-level parallelism in both multiprocess and multithreaded execution policies.

Section 4.2 introduces TExecutor as the executors’ common interface, with a multiprocess implementation in TProcessExecutor (Section 4.3) and a multithreaded implementation in TThreadExecutor (Section 4.4). These executors’ interfaces allow the explicit specification of the task’s size to potentially improve performance, as analyzed in Section 4.6. Section 4.7 proposes a new executor that combines multiprocessing and multithreading to mitigate the negative performance impact when dealing with non-uniform memory access (NUMA) architectures.

Finally, Section 4.5 justifies the creation of a centralized manager for the pool of workers in the multithreaded case, to avoid undesirable interactions between the parallelization modes in ROOT and the different classes built on top of the TBB scheduler.

4.1 The event-processing paradigm and task-level parallelism#

In a typical HEP analysis workflow, the event—that is the data representation of single or multiple piled-up collisions detected and stored for offline analysis—is the minimum unit of data to process. The necessary operations required by the analysis at hand are applied independently and consecutively to each event in a discrete collection, an iterative process that is referred to as the event loop.

Task-level parallelism refers to the partition of the workload into several tasks, or work packages, that are evaluated concurrently. In terms of the event-processing paradigm described above, this results in a division of the event loop into partitions, or blocks of events. These work units typically contain the same number of events, which are then distributed between the different concurrent processing units in the system. The event loop is a natural fit for the task-level approach, as the events recorded by the detectors are many and independent from each other. This makes it easy to encapsulate, in independent tasks, the evaluation of the event loop over subsets of the total recorded events.

Task-level parallelism is implemented in ROOT through the executors, which are ROOT classes implementing the MapReduce pattern³—a flexible pattern that appears rather often when processing data composed of independent elements.

Traditionally, task-level parallelism is extracted by dividing the workload in as many blocks as processing units are available in the system. This scheme though is not ideal in terms of workload balancing between the workers when the tasks to process contain early termination instructions or aggressive branching, which applies to many real use cases.

Instead, we divide the work into smaller tasks and delegate the load balancing to the schedulers of the chosen executor implementation so that we benefit from the mechanisms (e.g. work stealing, greedy reduction) they integrate. By default, we produce as many tasks as data elements are passed as inputs. However, this can create a significant task creation overhead that can impair performance. Therefore we also provide the option for users to chunk themselves the data as appropriate, or to rely on the built-in chunking mechanisms in our interfaces. These three options—the default or the two user-driven approaches—improve over the traditional approach in unbalanced cases and, in addition, accommodate more flexibility in the program design phase. Section 4.6 analyzes the impact that task granularity has on the performance obtained when parallelizing differently balanced examples.

While the approach we adopt in order to exploit task-level parallelism potentially improves performance, it requires a change in how multithreading problems have traditionally been addressed. For example, thread local variables become useless if we allow task interleaving, or adding synchronization directives between tasks would defeat the purpose of creating additional tasks for better load balancing. If these cases cannot be avoided, the traditional approach of dividing the workload is the best option.

4.2 TExecutor: a MapReduce for ROOT#

The MapReduce programming model expresses problems in parallel by partitioning the input data and distributing it to several workers for processing. During the first step, each one of these workers will then apply transformations on its data partition and return an intermediate result (Map). During the second step, all the intermediate results will be combined into a single output (Reduce).

TExecutor is the generic interface for MapReduce operations in ROOT. In addition to applying Map and Reduce over input data, TExecutor contemplates the case where no data is received as an input but a method must be applied N times and then its outputs have to be reduced, as well as the case where the Map operation does not generate intermediate results (ForEach).

In more detail, TExecutor offers the following interfaces or methods:

• Map: Function that receives as an input the processing method to apply and either input data or the number of times to apply the method. It returns the results in a standard vector.

• Reduce: Combination or reduction of the results returned by map into a final result. This operation is usually sequential, but an overload has been provided for parallelizing the case where the reduction function is a binary operator.

• MapReduce: Concatenation of the previous two interfaces.

• Foreach: A Map that does not generate partial results.

TExecutor has specialized instances of this pattern for multiprocessing (in TProcessExecutor) and thread-based multitasking (in TThreadExecutor).

4.3 TProcessExecutor#

Traditionally, multiprocessing has been the common approach to parallelize ROOT due to the thread-unsafe implementation of most interfaces. While there existed several multiprocess solutions such as PROOF⁴ and PROOF-Lite, used to parallelize the work with the data model in multicore machines, there was a need for a lighter framework capable of parallelizing more general operations in a transparent way for the user.

TProcessExecutor⁵ provides a lighter multiprocessing framework in ROOT, oriented to single nodes, whose interface implements the MapReduce model defined by TExecutor.

Listing 4.1 shows a basic example of the usage of TProcessExecutor. First we define our map function, which implements the task to be computed in parallel. Then we propose our reduction function, accumulating the intermediate result from the map. We initialize TProcessExecutor and the pool of workers. Finally, by calling TProcessExecutor::MapReduce, we perform the operation in parallel.

auto mapFunc = [](const UInt_t &i){
    return i+1;
};

auto reduceFunc = [](const std::vector<UInt_t> &mapV){
    return std::accumulate(mapV.begin(), mapV.end());
};

ROOT::TProcessExecutor pool;
pool.MapReduce( mapFunction, ROOT::TSeq<int>(100), reductionFunction);

Listing 4.1: Simple working example of a multiprocessed call to the MapReduce framework.

As a framework, TProcessExecutor improves on the previous approaches to single-machine multiprocessing (PROOF, PROOF-Lite) by forking a ROOT session. Before TProcessExecutor was available, ROOT’s multiprocessing performance on a single machine was undermined by the overhead due to the creation of the child processes, with constraints such as the creation of ad-hoc binaries for each of the subprocesses, or the setting of the process environment by passing all the information on scripts and libraries loaded in the client session. Instead, the system call fork provides the workers with access to the client’s memory, which greatly reduces the amount of information passed; and copy-on-write, which avoids duplicating the memory of the original process until the subprocesses write it.

Thanks to these advantages, creating workers becomes cheaper and faster. Another benefit is flexibility. While PROOF and PROOF-Lite are designed for dealing with the data model, in particular the TTree class, TProcessExecutor can tackle more generic problems thanks to its MapReduce model.

The combination of these new features allows a new kind of workflow in which we are not tied to multiprocessing from start to end. Instead, we can run our program sequentially until it is switched to a multiprocessing mode to perform a MapReduce operation on the data, and then gather the results and continue executing the code on a single core.

As a performance enhancement over the requirements set by TExecutor, the TProcessExecutor’s MapReduce call implements the greedy worker paradigm. In this operation mode, instead of processing a task and returning the processed result, workers continue drawing tasks from the queue and processing them until all work has been consumed. Then, they perform a partial reduction and send the result to the parent for a final reduction of all the partial results, using ROOT I/O for communication.

4.4 TThreadExecutor#

TThreadExecutor⁶ is a task-oriented, multithreaded MapReduce for ROOT that provides a simple programming model for multithreaded MapReduce operations on decoupled-data based loops. It implements the programming model defined by TExecutor, which means that, as Listing 4.2 exemplifies, it shares TProcessExecutor’s public interfaces.

auto mapFunc = [](const UInt_t &i){
    return i+1;
};

auto reduceFunc = [](const std::vector<UInt_t> &mapV){
    return std::accumulate(mapV.begin(), mapV.end());
};

ROOT::TThreadExecutor pool;
pool.MapReduce( mapFunction, ROOT::TSeq<int>(100), reductionFunction);

Listing 4.2: TProcessExecutor and TThreadExecutor interfaces are mostly interchangeable.

TThreadExecutor is built on top of Intel’s Threading Building Blocks (TBB) library, and in particular, its parallel_for and parallel_reduce instructions. TBB provides several useful features, such as efficient load balancing, parallel algorithms, and a powerful and configurable interface for task-based multithreading. Unlike alternatives such as OpenMP or POSIX threads, TBB is a cross platform library that doesn’t require compiler support and offers support for complex parallel patterns and concurrent data structures.

TBB implements load balancing by relying on work stealing⁷ instead of distributing the tasks evenly among the threads. In this scheme, each thread has a limited task queue that is processed sequentially. The remaining tasks after the initial distribution are kept in a common pool of tasks. If a thread completes the processing of all tasks in its queue, but there still exist tasks in the task pool, it will withdraw them from there and add them to its queue. If the task pool is empty, it will try to steal from the queue of other threads. Finally, if there are no tasks available for stealing, it will remain idle until the other threads complete their work. This mechanism minimizes the time a thread spends idle.

On initialization, TBB creates a pool containing the maximum number of threads supported by the system, which is governed from tbb::task_scheduler⁸. This task scheduler creates an implicit task arena⁹ where threads may share and execute tasks, allowing the specification of the maximum number of threads to be executed simultaneously. Two task schedulers can only co-exist if they have been instantiated from different user threads. This is important, for example, for resource-allocated asynchronous execution and manual load balancing. By default, ROOT treats TBB’s task scheduler as a singleton, controlled by TPoolManager (Section 4.5). TThreadExecutor allows for another degree of customization, handling a subset of the threads initialized by the scheduler through a task arena. This allows for concurrent execution of multiple TThreadExecutors by assigning a subset of the total resources to each one.

TThreadExecutor extends the TExecutor interfaces with two useful features:

• Chunking: By default, Map evaluates the mapping function once per element in the input data. If these data is composed of a large number of elements, and the user does not partition it, this becomes inefficient both spatially (we need to generate a large return std::vector) and temporally (e.g. if the mapping function evaluation is short we have too much task overhead). To address this problem, TThreadExecutor introduces the concept of chunking. The input data is divided into N chunks, specified by the user, and the tasks are evaluated over each one of these chunks—i.e. several data points at a time—applying reduction after each data chunk is processed.

• Parallel Reduction: If the user provides a binary function, the reduction step is parallelized by spawning a binary tree with the elements to process, and parallelizing the evaluation starting from the leaves. This interface is only available if the reduction function evaluates on double or float types in order to reduce complexity adjusting to the ROOT user needs.

Currently, TThreadExecutor is used in ROOT fitting to compute the maximum likelihood and the least squares functions in parallel. In TMVA¹⁰, it is utilized to train BDTs, by parallelizing the loss function; and DNNs, by parallelizing the internal matrix operations. It is also applied for performing implicitly multithreaded operations in ROOT I/O (reading, parallel writing, de-serialization and decompression of tree branches in parallel) and in the parallel execution of functional chains in RDataFrame¹¹.

4.5 TPoolManager#

ROOT provides two modes for expressing thread-based parallelism: implicit and explicit. In implicit multithreading (IMT), users write sequential code using ROOT interfaces and ROOT parallelizes its execution. Many common operations in ROOT adhere to IMT and, while for the moment IMT is not enabled by default, activating this functionality only requires a call to ROOT::EnableImplicitMT() (Listing 4.3).

With explicit multithreading, the user explicitly states which operations should be parallelized and how. ROOT provides several classes for the expression of parallelism, for example, by using locks (e.g. TMutex, TReadWriteLock, TCondition), by employing thread locality mechanisms (TThreadedObject), or by applying multithreaded patterns (TThreadExecutor).

The user may need a mixed solution for those cases requiring additional flexibility, such as parallelizing interfaces still not available in IMT; or employing user-generated multithreaded code, e.g. via TThreadExecutor, that makes use of the same common TBB scheduler. For reliability, we need to solve the interoperability problems that may arise in these scenarios.

  // First enable implicit multi-threading globally, so that the implicit parallelisation is on.
  ROOT::EnableImplicitMT();

  // Open the file containing the tree
  TFile *file = TFile::Open("tree.root");

  // Get the tree
  TTree *tree = nullptr;
  file->GetObject<TTree>("tree_0", tree);

  // Read the branches in parallel.
  // Note that the interface does not change, the parallelisation is internal
  for (Long64_t i = 0; i < tree->GetEntries(); ++i) {
     tree->GetEntry(i);  // parallel read
  }

  // IMT can be also disabled globally.
  // As a result, no tree will run GetEntry in parallel from now on
  ROOT::DisableImplicitMT();

  // This is sequential: the global flag is disabled
  for (Long64_t i = 0; i < tree->GetEntries(); ++i) {
     tree->GetEntry(i);  // sequential read
  }

Listing 4.3: Simplified example from the multicore tutorials in ROOT showcasing implicit multithreading.

To tackle these problems, we developed the new internal class TPoolManager to handle the TBB library scheduler. It acts as a common interface to TBB’s task scheduler for all thread-based parallelization in ROOT, providing shared ownership, reference counting, and automatic destruction and termination when the scheduler is idle.

Due to TBB restrictions, once the TBB scheduler is active, the number of active threads cannot be changed. Because of that, all instances pointing to the TPoolManager scheduler will need to be destroyed before the pool of threads can be initialized again.

TPoolManager grants the following advantages:

• Provides a unique interface to the TBB scheduler: TThreadExecutor, IMT and any class relying on TBB will get a pointer to the scheduler through TPoolManager::GetScheduler(), which will return a reference to the only pointer to the TBB scheduler that will be active in any ROOT process.

• Solves multiple undefined behaviors. Guaranteeing that all classes use the same multithreading scheduler avoids interferences and undefined behavior by providing a single instance of the scheduler and automated bookkeeping, initialization and destruction.

4.6 Task granularity analysis#

Dividing the work at thread-level in the traditional way, that is, in as many partitions as processing units there exist, has a drawback: if, for instance, some of the partitions complete their part of the work earlier than others, the corresponding processing units will remain idle, waiting for the remaining ones to finish, and leading to a suboptimal exploitation of the hardware resources. However, we can mitigate the time workers spend in idle state due to an unbalanced distribution of the workload by reducing the granularity of the data partitions.

In this section, we analyze the impact that the number of partitions has on the performance when parallelizing a job with ROOT’s executors. Our analysis is executed over different sizes of data and increasing number of processing units, for both TProcessExecutor and TThreadExecutor. In all cases, we increase the number of tasks to execute exploiting the built-in chunking mechanism of these classes, moving progressively from coarse-grained partitions, when the workload is divided in as many partitions as processing units available, to the most fine-grained division, with one task per data element.

For the following experiment, we designed a benchmark consisting of the conditional evaluation of two different implementations of the Vavilov probability distribution function—fast and accurate—which provides with a simple way to introduce imbalance in the execution time for the chunks. The fast implementation of the Vavilov probability density function¹² is about 5 times faster for the calculation of the distribution function, while its accurate implementation¹³ sacrifices speed for correctness. We perform the evaluation of the Vavilov fast function on the positive values of a given data set, and the Vavilov accurate evaluation on the negative values of the same data set. Listing 4.4 displays the code to perform our benchmark with an explicitly chunked MapReduce operation with TThreadExecutor, given a container of data, data, and a number of chunks, nChunks.

This benchmark was executed on a Intel Core i7-4790 desktop server with 4 physical cores (8 logical threads) at 3.6 GHz and 8 GB of RAM¹⁴, each core supporting hyperthreading, over two different data sets, both of them featuring unbalanced executions. The first data set is generated randomly following a Gaussian distribution with σ = 1, μ = −0.25 for the first third of data elements; σ = 1, μ = 0.25, for the second third; and σ = 1, μ = −0.655 for the remaining third. The measurements were timed with the C++ standard high resolution clock, in several consistent executions with low variation, and then averaged and used to compute the speedup with respect to the sequential execution time measured.

// (... variable initialization: data, nChunks)

ROOT::Math::VavilovAccurate vavilovAc(0.3, 0.5);
ROOT::Math::VavilovFast vavilovFast(0.3, 0.5);

auto mapFunction = [&data, &vavilovAc, &vavilovFast](const unsigned i){
   return data[i]>0? vavilovAc.Pdf(data[i]) : vavilovFast.Pdf(data[i]);
};

auto redFunction = [](const std::vector<double> vec){
   return std::accumulate(vec.begin(), vec.end(), .0);
};

ROOT::TThreadExecutor threadPool(nThreads);
auto res = threadPool.Map(mapFunction, ROOT::TSeqU(data.size()), redFunction, nChunks);

Listing 4.4: Example code of our benchmark computing the Vavilov probability density function in its fast implementation for negative values of the data, and its accurate implementation for positive values of the same data.

Figure 4.1 displays the speed ups obtained for different sizes of the data set when using a different number of threads (including hyperthreading) in TThreadExecutor, increasing the number of data partitions in each case. For 100 points, the execution time is dominated by the task creation overhead, causing a slowdown in the total execution time, independently of the number of data partitions. The performance improves as we increase the dataset size until it stabilizes. There we observe significant speed ups, which improve slightly until, again, the task overhead negatively affects the execution time when dealing with smaller tasks.

This example exhibits higher performances when exploiting hyperthreading. With this option enabled we observe a gradual, more pronounced improvement in performance, reaching up to a remarkable 45% higher speed up than its non-hyperthreaded counterpart. However it exhibits a more sharp decline in speed up when the task creation overhead starts to be noticeable.

Figure 4.2 shows the results obtained for a multiprocessing execution with TProcessExecutor of the same example. While a multiprocessing execution is more useful in several contexts (e.g. when performing non thread-safe operations), and the results display a fair speed up for big data sets, the task overhead introduced by TProcessExecutor is several orders of magnitude larger than that of TThreadExecutor, requiring a greater amount of work for the acceleration to be noticeable. For this same reason, the plots in Figure 4.2 exhibit an earlier decline of performance with a higher number of tasks.

The near-optimal speed up we obtain in the theoretically most unbalanced case (dividing the workload into as many tasks as processing units), seems to indicate that the data set may not be unbalanced enough to observe dramatic improvements when tuning task granularity. We observe this situation in a performance trace produced using Intel’s Vtune Amplifier 2018, where we visualize the performance trace of an execution of our benchmark with 800 million points, 4 processing units, and 4 tasks for both TProcessExecutor and TThreadExecutor. The green color in the timeline indicates process running time, the brown graphs represent the CPU time, and the red spikes correspond to spin and overhead time, indicating synchronization areas. A second trace, with the same parameters but an increase in the number of tasks to 8000, demonstrates an improvement in workload balancing.

The second data set, which is more extremely unbalanced, is composed of two distinct parts: the first two thirds of the data set evaluate the fast version of the Vavilov distribution, and the remaining third performs the accurate evaluation. This case exhibits a more accentuated imbalance among the processing units: the main thread, after completing the execution of its assigned tasks, spends a noticeable amount of time in a locked state, spinning while it waits for the remaining threads to complete their work. Again, increasing the number of tasks to 8000 provides a better balanced example of the same execution.

Figures 4.7 and 4.8 present the results of executing our benchmark with this second data set. For the multithreaded and the multiprocessed cases, they respectively display the increase in speed up when varying task granularity in different configuration pairs of number of threads and size of the workload. In this case, we observe an earlier steeper increase in speed up for the higher values of the chunk size, indicating the convenience of splitting tasks into smaller partitions. For the largest workloads, we perceive a gradual decrease in speed up caused by excessive fine-graining of the tasks, yielding an execution time that is dominated by the task overhead. The multiprocessed case exhibits, again, an earlier decay of the speed up due to the higher task overhead inherent to a multiprocess solution.

These results support the importance of an adequate task size and showcase the impact it has on parallel performance for unbalanced workloads. Moreover, they indicate that granularity has to be treated differently for the multithreaded and the multiprocessed cases, keeping task creation overhead in mind.

4.7 A NUMA-aware executor#

For convenience, modern high-end, many-core processors often adopt multi-socket designs, in which each socket is a processor package composed of multiple cores with dedicated physical RAM and integrated memory controllers. Despite the clear advantages of this approach in terms of modularity and scalability, these architectures may exhibit undesirable effects resulting from non-uniform memory accesses (NUMA).

We refer to these undesirable consequences as NUMA effects: negative impacts on performance (e.g. increased latency, bandwidth overload) resulting from remote memory accesses in a heterogeneous memory hierarchy. These negative effects are easily observable in multithreaded operations where there coexist several tasks accessing the same block of memory concurrently from different sockets or NUMA domains.

In ROOT, TThreadExecutor shares a common pool of tasks, where the different threads of the system draw the next item to process once their task queue is depleted. These tasks are most frequently tied to a partition of the data, where the mapping function will be applied. The data may be stored in one of the NUMA domains or shared between them, depending on the size and the locality of the writing or reading policy. Frequently, threads are responsible for the creation of tasks which require accesses to memory of a remote domain. This memory is brought to the shared cache of the socket and fetched by the higher-level caches for faster accesses to data. If several misses on the last-level cache, shared between threads, occur, this will cause frequent accesses to remote memory and, therefore, may overload the interconnection.

In addition, we cannot currently pin tasks to specific threads. Even if we could, it is not possible to assert that this would be the optimal solution. Threads can execute in any of the processors (in the case of TBB, they may even switch execution between processors dynamically), and it might happen that data is moved from one NUMA node to another in order to increase the locality access and, as a consequence, decrease aggregate bandwidth usage, latency, and NUMA effects overall.

To address the common problem of NUMA effects, we have developed a new NUMA-aware hybrid multiprocess-multithread executor for handling multithreaded operations in NUMA architectures. First, we split the MapReduce processing, by means of TProcessExecutor, into as many partitions as NUMA domains exist in the system, and we distribute the work among them. The execution of each partition is restricted to a set of cores belonging to the same NUMA node; that is, each processing unit execution is confined to a NUMA domain. Each task will then be split following the typical multithreaded strategy with TThreadExecutor. Each TThreadExecutor internal to the TProcessExecutors’ tasks generates its own independent pool of threads that are only able to execute tasks in the NUMA node to which they are confined.

Libnuma¹⁵ is the library that manages the restricted execution of processes in a NUMA node. In our case, we employ its numa_max_node() directive to obtain the number of NUMA nodes in the system, and the numa_run_on_node(int i) and numa_run_on_node_mask(numa_all_nodes_ptr) directives to modify the scope of the execution.

Listing 4.5 displays a code snippet extracted from the TNUMAExecutor class, implementing the process described above. The instantiation and execution of TProcessExecutor uses fNDomains = numa_max_node()+1 worker units and fNDomains tasks (described by the sequence TSeq(fNDomains)) that execute the C++ lambda runOnNode. This lambda receives the task index (the same as the domain index in our case) and commences by restricting its execution to the corresponding domain. This restriction is lifted in the last instruction of the lambda before returning. In between, it instantiates and executes a NUMA domain specific TThreadExecutor, allowing the specification of the chunking granularity.

template<class F, class R>
auto TNUMAExecutor::MapReduce(F func, unsigned nTimes, R redfunc,
                                unsigned nChunks) -> typename std::result_of<F()>::type
{
   auto runOnNode = [&](unsigned int i) {
      numa_run_on_node(i);
      ROOT::TThreadExecutorImpl pool{fDomainNThreads};
      auto res = nChunks?
                    pool.MapReduce(func, nTimes, redfunc, nChunks/fNDomains) :
                    pool.MapReduce(func, nTimes, redfunc);
      numa_run_on_node_mask(numa_all_nodes_ptr);
      return res;
   };

   ROOT::TProcessExecutor proc(fNDomains);
   return proc.MapReduce(runOnNode, ROOT::TSeqU(fNDomains), redfunc);
}

Listing 4.5: Example of the interaction of libnuma, TProcessExecutor and TThreadExecutor, corresponding to TNUMAExecutor’s N-execution MapReduce overload.

To benchmark TNUMAExecutor, we performed an unbinned fit (Listing 4.6) of the diphoton invariant mass distribution resulting from a Higgs boson, with an analytically computed integral. This parallelization analysis is performed in the Case example section of the article on the parallelization of the fit, but here we focus on the monitoring and potential decrease of the NUMA effects. To augment the probability of observing uncore events¹⁶ (i.e. events that occur outside of the processor such as remote memory accesses), we fit an amount of data that doubles the size of the last-level cache of a NUMA domain. This way we are more likely to observe cache misses at the last-level cache of a NUMA node, as data must be fetched from memory which is potentially stored in another domain.

This fit is benchmarked on a Intel Xeon E5-2683 v3¹⁷ server composed of two NUMA domains with 14 physical cores and 28 logical threads each at 2.00 GHz with 32 GB of RAM per NUMA node. Each NUMA domain of the system has a last-level cache of 35 MB, which implies that fitting over 9.5 million (double precision) data points should be sufficient to provoke a considerable number of NUMA effects.

double func(const double *data, const double *params)
{
    //... Model function
}

// Enable implicit multithreading
ROOT::EnableImplicitMT();

TF1 f("model", func, 100, 200, 4);
f.SetParameters(1, 1000, 7.5, 1.5);
ROOT::Math::WrappedMultiTF1 wF(f, f.GetNdim());
ROOT::Fit::UnBinData dataSB(nPoints);
for (unsigned i = 0; i < nPoints; ++i) dataSB.Add(f.GetRandom());

ROOT::Fit::Fitter fitter;
fitter.SetFunction(wF, false);
fitter.Fit(data, 0);

Listing 4.6: Implicitly parallel unbinned fit of a model function.

In an attempt to observe and quantify NUMA effects, from all the uncore events available, we choose to monitor and record, with 1 second samples, the DIMM¹⁸ page misses in the DDR memory controller; that is, the number of requests to remote RAM memory. Running the benchmark with 14 threads to fit 9.5 million data points, we notice that TNUMAExecutor exhibits a considerable improvement with respect to fitting via the plain TThreadExecutor. Figure 4.9 compares the number of remote memory accesses monitored with and without the NUMA executor, by displaying the distribution density of these accesses. We can infer that our choices of benchmark and uncore events were appropriate as, by using TNUMAExecutor, we observe a 28.09% reduction in the average number of page misses in the memory controller, and we obtain a 1.89 times improvement in the execution time (Table 4.1).

Table 4.1: Comparison of the execution times before and after substituting TThreadExecutor by TNUMAExecutor and speed up in the later case.

Unfortunately, due to the implicit use of multiprocessing in TNUMAExecutor, we cannot simply replace TThreadExecutor by TNUMAExecutor. If so, these promising results would be hindered by the overhead introduced by TProcessExecutor. In Figure 4.10 we analyze the convenience of using TNUMAExecutor for an increasing number of threads of the system, and an increasing amount of data points. By reducing the problem and performing single evaluations of the objective function, in this problem the maximum likelihood, instead of a complete fitting minimization process, we conclude that it is only convenient to rely on TNUMAExecutor when the size of our data is of the same magnitude of the last level cache, 35 MB, in this case O(10⁶). Finally, for large datasets, we notice a large gap in speed up with respect to TThreadExecutor between the hyperthreaded and the non hyperthreaded case. This indicates that employing hyperthreading helps to mitigate the observed NUMA effects in this case when employing only TThreadExecutor.

4.8 Conclusions#

This chapter introduces the executors, a performant set of tools to exploit task-level parallelism in ROOT-based analyses. The executors are ROOT classes implementing the parallel MapReduce, identified as one of the most frequent patterns in ROOT analyses.

These easily-deployable classes provide a set of common interfaces that hide the complexity of concurrent programming from the user. We factored out the MapReduce functionality of TPool, the precursor of the executors, into TProcessExecutor, ROOT’s multiprocess executor. We then defined and implemented TThreadExecutor, a multithreaded executor leveraging TBB as a backend, and abstracted the executors’ common interfaces into TExecutor. The Executors provide good performance and an improved programming model. On the other hand, they require adapting the parallel analysis to avoid synchronization and, in the multithreaded case, thread-local storage and thread race conditions.

We benchmarked and analyzed the executors’ performance and demonstrated the benefits derived from tuning task granularity with the chunking interfaces of TProcessExecutor and TThreadExecutor when dealing with unbalanced workloads. TThreadExecutor exhibits close to ideal performance, higher than TProcessExecutor, which experiences a larger task overhead inherent to a multiprocessed implementation, of O(10^-3 s) compared to O(10^-7 s) for TBB tasks in TThreadExecutor. Thanks to its lower task overhead and the work stealing mechanisms provided by TBB, TThreadExecutor exhibits an earlier balancing of the workload when increasing the number of tasks and a later decay in performance. TProcessExecutor showcases good speed up as well, but it requires a larger amount of data to do so.

In addition, with TPoolManager, a single point of management for the TBB pool of threads, we successfully introduced a solution for undefined behaviours arising from interactions between the implicit and explicit multithreaded modes of ROOT.

Finally, we provided an executor for NUMA architectures. By combining TProcessExecutor with TThreadExecutor and concealing the execution of the latter in each NUMA domain via libnuma, we reduce the impact of NUMA effects in performance. This new executor, TNUMAExecutor, diminishes the number of uncore events monitored when dealing with sizeable data, up to a remarkable 30% less when performing a fit in a NUMA machine with two NUMA domains of 14 cores (+14 hyperthreaded). As a result, this example displays a speed up to 1.8x against the baseline TThreadExecutor execution.