Improving all-reduce collective operations for imbalanced process arrival patterns

All-reduce operation is one of the most common collective operations used in HPC software [11]. We can define it as a reduction in a vector of numbers using a defined operation, e.g., sum, which needs to be cumulative, and then distribution of the result into all participating processes, in short: all-\(\hbox {reduce} = \hbox {reduce} + \hbox {broadcast}\). There are plenty of all-reduce algorithms; Table 1 summarizes the ones used in two currently most popular, open source MPI implementations: OpenMPI [4] and MPICH [3].

The basic linear algorithm performs linear reduce (flat tree: the root process gathers and reduces all data from the other processes) followed by the broadcast without any message segmentation [4]. Segmented linear tree creates pipeline between the participating processes, where the data are split into segments and sent from process 0 to 1 to 2 ... to \(P-1\). The nonoverlapping algorithm uses default reduce followed by the broadcast; these operations are not overlapping: the broadcast is performed sequentially after the reduce, even if both use segmentation and some segments are ready after the reduce [4]. Recursive doubling (a.k.a. butterfly and binary split) is performed in rounds; in every round each process exchanges and reduces its data with another, corresponding process, whose rank changes round by round, after \(\log _2P\) rounds (P is the total number of participating processes); all data are reduced and distributed to the processes [20]. Ring is also performed in rounds, the data are divided into P segments, and each process sends one segment per round (the segment index depending on the round number and the process rank) to the next process (the processes are usually ordered according to their ranks). At the beginning, there are performed \(P-1\) rounds, where a segment delivery is followed by its data reduction; thus, after these rounds, each process has exactly one segment fully reduced and needs to forward it to other processes. This is performed during next \(P-1\) rounds, where the data are gathered, i.e., the processes transfer the reduced segments; thus, the final result is delivered to all processes [20]. Figure 2 presents an example of a ring execution. Segmented ring is similar to the ring, but after the segmentation related to process number, an additional one is performed for pipeline effect between corresponding processes [4]. In the Rabenseifner algorithm, the reduction is performed in two phases, first the scatter and reduction of data are executed (using recursive doubling on divided data), and then, the data gathering takes place (using recursive halving on gathered data), where the reduced data come back to all processes [19].

Not much work has been done for imbalanced process arrival patterns analysis. To the best of our knowledge, the following four papers cover the research performed in this area.

Faraj et al. [11] performed advanced analysis of process arrival patterns (PAPs) observed during execution of typical MPI programs. They executed a set of tests proving that the differences between process arrival times (PATs) in operations of the collective communication are significantly high and they influence the performance of the underlying computations. The authors defined a PAP to be imbalanced for a given collective operation with a specific message length when its imbalance factor (a ratio between the highest difference between the arrival times of the processes and time of the simple (point-to-point) message delivery between each other) is larger than 1. The authors provided examples of typical HPC benchmarks, e.g., NAS, LAMMPS or NBODY, where imbalance factor, during their execution in a typical cluster environment, equals 100 or even more. They observed that, such behavior usually cannot be controlled directly by the programmer, and the imbalances are going to occur in any typical HPC environment. The authors proposed a mini-benchmark for testing various collective operations and found out the conclusion that the algorithms which perform better with balanced PAPs tend to behave worse when dealing with imbalanced ones. Finally, they proposed solution: their self-tuning framework—STAR-MPI [10], which includes a large set of various implementations of collective operation and can be used for different PAPs, with automatic detection of the most suitable algorithm. The framework efficiency was proved by an example of tuned all-to-all operations, where the performance of the set of MPI benchmarks was significantly increased.

As a continuation of the above work, Patarasuk et al. proposed a new solution for broadcast operation used in MPI application concerning imbalanced PAPs of the cooperating processes [17]. The authors proposed a new metric for the algorithm performance: competitive ratio—\(\textit{PERF}(r)\), which describes the influence of the imbalanced PATs for the algorithm execution, regarding the behavior for its worst-case PAT. They evaluated well-known broadcast algorithms, using the above metric, and presented two new algorithms, which have constant (limited) value of the metric. The algorithms are meant for large messages and use the subsets of cooperating processes to accelerate the overall process: the data are sent to the earliest processes first. One of the algorithms is dedicated for nonblocking (arrival_nb) and other for blocking (arrival_b) message-passing systems. The authors proposed a benchmark for algorithms evaluation, which introduces random PAPs and measures their impact on the algorithm performance. The experiments were performed using two different 16-node compute clusters (one with InfiniBand and other with Ethernet interconnecting network), and 5 broadcast algorithms, i.e., arrival_b, flat, linear, binomial trees and the one native to the machine. The results of the experiments showed the advantage of the arrival_b algorithm for large messages and imbalanced PAPs.

Marendic et al. [16] focused on an analysis of reduce algorithms working with imbalanced PATs. They assumed atomicity of reduced data (the data cannot be split into segments and reduced piece by piece), as well as the Hockney [12] model of message-passing (time of message transmission depends on the link bandwidth: \(\beta \) and constant latency: \(\alpha \), with an additional computation speed parameter: \(\gamma \)) and presented related works for typical reduction algorithms. They proposed a new static load balancing optimized reduction algorithm requiring a priori information about current PATs of all cooperating processes. The authors performed a theoretical analysis proving the algorithm is nearly optimal for the assumed model. They showed that the algorithm gives the minimal completion time under the assumption that the corresponding point-to-point operations start exactly at the same time for any two communicating processes. However, if the model introduces a delay of the receive operation in comparison with the send one, which seems to be the case in real systems, the algorithm does not utilize this additional time in receiving process, although, in some cases, it could slightly improve the performance of the overall reduce operation. The other proposed algorithm, presented by the authors: a dynamic load balancing, can operate under the limited knowledge about PATs, being able to atomically reconfigure the message-passing tree structure while performing reduce operation using auxiliary short messages for signaling the PATs between the cooperating processes. The overhead is minimal in comparison with the gains of the PAP optimization. Finally, a mini-benchmark was presented and some typical PAPs were examined, the results showed the advantage of the proposed dynamic load balancing algorithm versus other algorithms: binary tree and all-to-all reduce.

Marendic et al. [15] continued the work with optimization of the MPI reduction operations dealing with the imbalanced PAPs. The main contribution is a new algorithm, called Clairvoyant, scheduling the exchange of data segments (fixed parts of reduced data) between reducing processes, without assumption of data atomicity, and taking into account PATs, thus causing as many as possible segments to be reduced by the early arriving processes. The idea of the algorithm bases on the assumption that the PAP is known during process scheduling. The paper provided a theoretical background for the PAPs, with its own definition of the time imbalances, including a PAT vector, absolute imbalance, absorption time as well as their normalized versions, followed by the analysis of the proposed algorithm, and its comparison to other typically used reduction algorithms. Its pseudo-code was described and the implementation details were roughly provided with two examples of its execution for balanced and imbalanced PAPs. Afterwards the performed experiments were described, including details about used mini-benchmark and the results of practical comparison with other solutions (typical algorithms with no support for imbalanced PAPs) were provided. Finally, the results of the experiments showing advantage of the proposed algorithm were presented and discussed.

The PAP collective communication algorithms require some knowledge about the PATs for their execution. Table 2 presents the summary of the approaches used in the works described above.

In [11] (dedicated for an all-to-all collective operation), there is assumption about the call site (a place in the code where the MPI collective operation is called) paired with the message size that they have a similar PAPs for the whole program execution, or at least their behavior changes infrequently. The proposed STAR-MPI [10] system periodically assesses the call site performance (exchanging measured times between processes) and adapts a proper algorithm, trying one after another. The authors claim that it requires 50–200 calls to provide desired optimization. This is a general approach, and it can be used even for other performance issues, e.g., network structure adaptation.

In [17] (dedicated for a broadcast operation), the algorithm uses additional, short messages sent to root process signaling process readiness for the operations. In case the some processes are ready, the root performs sub-group broadcast; thus, the a priori PATs are not necessary for this approach. Similar idea is used in [16] (dedicated for a reduce operation) where the additional messages are used not only to indicate readiness, but also to redirect delayed processes.

In [15] (dedicated for a reduce operation), the algorithm itself does not include any solution for the PAT estimation, that is why it is called Clairvoyant, but the authors assume recurring PAPs and give the suggestion that there can be used simple moving averages (SMA) approximation. This solution requires the additional communication to exchange the PAT values, what is performed every k iterations, thus introducing the additional communication time. The authors claim that the speedup introduced by the usage of the algorithm overcomes this cost and provide some experimental results showing the total time reduction in the overall computations.

The algorithm is an extension of the linear tree [4], which transfers the data segments sequentially through processes exploiting the pipeline parallel computation model. The proposed modification causes the processes to be sorted by their arrival times. While the faster processes start the communication earlier, the later ones have more time to finish the computations.

Figure 5 presents pseudo-code of the algorithm. At the beginning, new identifiers are assigned to the processes according to the arrival order; then, the data are split into equal segments, in our case we assume P segments, where P is a number of the cooperating processes. Afterwards the reduce loop is started (lines: 1–6), where the data segments are transferred and reduced, and then the override loop is executed (lines: 7–12), where the segments are distributed back to all processes.

Let’s assume \(\tau \) is a time period required for transfer and reduction/overriding of one data segment between any two processes. Figure 6b presents an example of SLT execution, where the process 0 arrival time is delayed for \(4\tau \) in comparison with all other processes. Using the regular linear tree reduction, the total time of the execution would be \(14\tau \) (see Fig. 6a), while the knowledge of the PAP and procedure of sorting the processes by their arrival times make the delayed process to be the last one in the pipeline and cause the whole operation time to be decreased to \(12\tau \).

This algorithm is an extended version of ring [20], where each data segment is reduced and then passed to other processes in synchronous manner. The idea of the algorithm is to perform a number of so-called, reducing pre-steps, between faster processes (with lower arrival times), and then the regular processing, like in the typical ring algorithm, is performed.

Figure 7 presents pseudo-code of the algorithm. First of all, the processes are sorted by their arrival times and the new ids are assigned, i.e., initially the processes perform communication in direction from the earliest to the latest ones. Then, data are split into P equal segments and for each such a segment the number of pre-steps is calculated (lines: 1–6), its value depending directly on the estimated process arrival times (PATs): \(a_i\) and the time of transfer and reduction/overriding of one segment: \(\tau \). Knowing the above, the algorithm resolves where each segment starts and finishes its processing; thus, two process ids are assigned for each segment: \(sp_i\) and \(rp_i\), respectively (lines: 7–13).

Afterwards, initial value of the segment index: \(s_i\) is calculated from which every process starts processing. In the case of the regular ring algorithm, its value depends on the process id only; however, for PRR it also regards the pre-steps performed by the involved processes (line: 14). Then, the reduce loop is started, the pre-steps and the regular steps are performed in one block, where the variable \(s_i\) controls which segments are sent, received and reduced (lines: 15–21). Similarly, the override loop is performed, being controlled by the same variable (lines: 22–27).

In the regular ring, every process performs \(2P-2\) receive and send operations; thus, total number is \(P\times (2P-2)\). In case of PRR, this total number is the same; however, the faster processes (the ones which finished computation earlier) tend to perform more communication while executing the pre-steps. In the case, when the arrival of the last process is largely delayed to the next one (i.e., more than \(P\times \tau \)) it performs only \(P-1\) send and receive operations, while every other process does \(2P-1\) ones.

Figure 8b presents an example of PRR execution, where the process 0 arrival time is delayed for \(2\tau \) (\(2\times \) time of transfer and reduction/overriding of one data segment) in comparison with all other processes. Using the regular ring reduction, the total time of the execution would be \(8\tau \) (see Fig. 8b), while the knowledge of the PAP, performing two additional reduction pre-steps and setting the delayed process to be the last one in the pipeline, causes the whole operation time to be reduced to \(7\tau \).

Figure 9 presents pseudo-code of the proposed benchmark evaluating the performance of the proposed algorithms in the real HPC environment using MPI [6] for communication purposes. For every execution, there is a sequence of repeated iterations consisting of the following actions:

For the comparison purposes, we used two typical all-reduce algorithms: ring [20] and Rabenseifner [19], they are implemented in the two most popular open source MPI implementations: OpenMPI [4] and MPICH [3], respectively, and used for large input data size. To the best knowledge of the author, there are no PAP optimized all-reduce algorithms described in the literature.

The benchmark was executed in a real HPC environment using cluster supercomputer Tryton, placed in Centre of Informatics – Tricity Academic Supercomputer and NetworK (CI TASK) at Gdansk University of Technology in Poland [14]. The cluster consists of homogeneous nodes, where each node contains 2 processors (Intel Xeon Processor E5 v3, 2.3 GHz, Haswell), with 12 physical cores (24 logical ones, due to hyperthreading technology) and 128 GB RAM memory. In total, the supercomputer consists of 40 racks with 1600 servers (nodes), 3200 processors, 38,400 compute cores, 48 accelerators and 202 TB RAM memory. It uses fast FDR 56 Gbps InfiniBand in fat tree topology and 1 Gbps Ethernet for networking. Total computing power is 1.48 PFLOPS. The cluster weighs over 20 metric tons.

The experiments were performed using a subset of the nodes, with HT switched off, grouped in one rack and connected to each other trough a 1 Gbps Ethernet switch (HP J9728A 2920-48G). The benchmark input parameters were set up to the following values:

Table 3 presents the results of the benchmark execution for 1 M of floats of reduced data, 1 Gbps Ethernet network, where only one process was delayed on 48 nodes in a cluster environment of Tryton [14] HPC computer. The results are presented as absolute values of average elapsed time: \(\bar{e}_\mathrm{alg}\) and speedup: \(s_\mathrm{alg}\), in comparison with ring algorithm \(s_\mathrm{alg}=\frac{\bar{e}_\mathrm{ring}}{\bar{e}_\mathrm{alg}}\), where alg is the evaluated algorithm.

In this setup, the ring algorithm seems to be more efficient than the Rabenseifner; thus, in further analysis we use the former for reference purposes. For more balanced PAPs, where the delay is below 5 ms the proposed algorithms perform worse than the ring, it is especially visible for SLT (about 25% slower); however, PRR shows only slight difference (below 5%). On the other hand, when the PAP is more imbalanced with the delay over 10 ms, both algorithms perform much better (up to 15% faster than the ring). For the really high delays, the results stabilize providing about 17 ms of average elapsed time savings, causing the total speedup to be lower. The comparison of influence of changes in the (maximum) delay on the algorithms’ performance is presented in Fig. 10.

Table 4 presents the benchmark results related to PRR in comparison with the ring algorithm, executed on 48 nodes with 1 Gbps Ethernet interconnecting network. While the absolute average elapsed time savings of PRR algorithm are higher for longer messages (up to 82 ms for 8 M), the high speedup values occur for all message sizes providing relative savings up to 15%. We can observe that the lower delays cause lager speedup losses (up to 22%, but for absolute time: 1.5 ms only), which is especially visible for the smallest message size: 128 K. In general, for the lower delays, the PRR is comparable with the ring (see Fig. 10 for 1 M of float numbers reduced data size).

The mode of the introduced PAT delay influences slightly the measured values. For larger delays and message sizes, when only one process is delayed the PRR algorithm provides slightly smaller relative savings than in the case when the delays were introduced for all processes, with the uniform probabilistic distribution, it is especially visible for 500–1000 ms delays and message sizes of 2–8 M of floats. However, in general, the PRR works fine for both modes of PAT delay.

Figure 11 presents the behavior of the tested algorithms in a context of the changing data size. While the PRR algorithm performs well for a wide range of size values, the SLT lags for the larger data, the threshold depends on the (maximum) delay, e.g., for 50 ms the SLT is worse than the regular ring for \({4}{+}\,\hbox {M}\) of float numbers data size.

Figure 12 shows the tested algorithms performance related to the increase in the number of the nodes/processes exchanging reduced data. For 32 nodes, we observed interesting behavior of the Rabenseifner algorithm, which was designed for power of 2 node/process numbers. Both the PRR and SLT algorithms present stable speedup over the regular ring algorithm, proving good usability and high scalability for imbalanced PAPs.