Abstract
We consider an efficient realization of the all-reduce operation with large data sizes in cluster environments, under the assumption that the reduce operator is associative and commutative. We derive a tight lower bound of the amount of data that must be communicated in order to complete this operation and propose a ring-based algorithm that only requires tree connectivity to achieve bandwidth optimality. Unlike the widely used butterfly-like all-reduce algorithm that incurs network contention in SMP/multi-core clusters, the proposed algorithm can achieve contention-free communication in almost all contemporary clusters including SMP/multi-core clusters and Ethernet switched clusters with multiple switches. We demonstrate that the proposed algorithm is more efficient than other algorithms on clusters with different nodal architectures and networking technologies when the data size is sufficiently large.
Introduction
The all-reduce operation combines values from all processes and distributes the results to all processes. It is commonly used in parallel computing. In the Message Passing Interface (MPI) standard [18] , the routine for this operation is MPI Allreduce.
We consider an efficient realization of the all-reduce operation with large data sizes in cluster environments, under the assumption that the reduce operator is associative and commutative. We derive a tight lower bound of the amount of data that must be communicated of the proposed algorithm.
The rest of the paper is organized as follows. Section 2 introduces the all-reduce operation and the communication model that we use. In Section 3, we derive the theoretical lower bound on the communication time required for this operation. Section 4 presents the proposed bandwidth optimal all-reduce algorithm. Section 5 reports the results of our experiments. The related work is discussed in Section 6. Section 7 concludes the paper.
Background

All-reduce operation
We will use a generic operator ⊕ to denote the reduce operator in the all-reduce operation.
MPI Allreduce requires the reduce operator to be associative, that is, (a ⊕ b) ⊕ c = a ⊕ (b ⊕ c).
Moreover, all built-in operations for MPI Allreduce are also commutative, that is, a⊕b = b⊕a.
We assume that the reduce operator is both associative and commutative in this paper.
In terms of operating results, an all-reduce operation is equivalent to a reduction operation that reduces the results to one process, followed by a broadcast operation that distributes the results to all processes. Specifically, let the N processes be denoted as p 0 , p 1 , ..., p N −1 .
Before the all-reduce operation, each process p i , 0 ≤ i ≤ N − 1, has X data items a In the presence of rounding errors, MPI poses some quality requirements for this operation [23] . It is required that all processes must receive the same resulting data. In addition, the same reduction order and bracketing for all elements is not strictly required, but should be strived for.
Communication performance model
Since we consider operations with large data sizes, we will use a linear model to model the In analyzing all-reduce algorithms, we assume that all processes start the operation at the same time. The time for an all-reduce operation is defined as the time when all processes start the operation until the last process receives the last message (and computes the final result). Our analysis focuses on the communication time and ignore the computation time. In practice, both the communication start-up overheads and the computation in the operation, which are omitted in our analysis, can be significant.
The lower bounds
In a general-purpose all-reduce operation, data items are independent of each other. The reduction on different items are independent of one another. Hence, the amount of data that must be communicated in order to complete an all-reduce operation on X items is equal to X times the amount for the single-item all-reduce operation. To obtain the lower bound for a general all-reduce operation on X items, we only need to find the lower bound for the operation on a single item.
In a one-item all-reduce operation on N processes, each process p i , 0 ≤ i ≤ N − 1, has one initial item a i . At the end of the all-reduce operation, all processes contain the final result r = a 0 ⊕ a 1 ⊕ ..... ⊕ a N −1 . Let B = {b 0 , ..., b i } be a set of initial items. We will use the notation
operation on the initial items in B. Let ALL = {a 0 , a 1 , ..., a N −1 } be the set that contains all initial items, r = ⊕(ALL). Let B = {b 0 , b 1 , ..., b i } ⊆ ALL be a subset of initial items,
, as well as the final result r are partial results. Let φ be the empty set. Since φ ⊆ ALL, for completeness, we introduce a null (empty) partial result, denoted as ⊥ = ⊕(φ). Intuitively, any computation (reduction) result that is not a partial result will not contribute to the calculation of the final result in any way and can be represented as ⊥. Each partial result except ⊥ has the same size as an initial item. Let ⊕(B) be a partial result and a ∈ B, we say that the effect of item a is in the partial result ⊕(B). Let pr 1 = ⊕(A) and pr 2 = ⊕(B) be two partial results.
We extend the reduce operator to take partial results as parameters. Specifically,
Basically, if two partial results cover disjoint sets of initial items, applying the reduce operation on these two partial results yields another partial result that covers the union of the initial items. When the items covered by these two partial results intersect, applying the reduce operation on the two partial results yields something that is not a partial result (some initial items appear multiple times), which is denoted by ⊥ as discussed earlier.
In our proofs, we also use a operator on partial results, which is defined as follows. Let pr 1 = ⊕(A) and pr 2 = ⊕(B) be two partial results,
Basically, the operator removes the effects of items in B from the partial result pr 1 . For
During the all-reduce operation, each process may send a partial result that it can compute to another process or receive a partial result from another process. The partial results that can be computed by a process in a given time depend on the initial item at the process as well as the set of partial results received by the process. We define the state of a process at a given time to be the set of partial results that include the initial item in the process and all partial results that the process received from other processes. We call the set of all partial results that can be computed at a given time in a process the coverage set of the process.
Clearly, the coverage set is a function of the state. Let the state of a process be S, we denote the coverage set as COV ER(S):
The partial results in a state are atomic items that cannot be separated even though they can be combined with other partial results. For example, if a process has a state S = {a, b ⊕ c}, the coverage set is COV ER(S) = {⊥, a, b ⊕ c, a ⊕ b ⊕ c}, which does not include a ⊕ b or a ⊕ c.
In the derivation of the lower bound, we focus on all-reduce schemes whose communications do no have race conditions such that all messages in an operation can be sequentialized.
All-reduce algorithms with race conditions are usually incorrect and not considered in this paper. Without loss of generality, we will also assume that each message transfers one partial result between two processes: a message with multiple partial results are treated as multiple messages. A message is denoted as (src → dst, pr), where src is the source process, dst is the destination process, and pr is a partial result. Let a generic all-reduce scheme use n messages. Under the assumption that the messages do not have race conditions and can be sequentialized, the n messages can be ordered based on the timing when each message occurs.
For concurrent messages that happen at the same time, the tie can be broken arbitrarily. Let messages M 1 , M 2 , ..., M n be the n messages in an all-reduce scheme ordered based on their
be the state of process p i before M j , 1 ≤ j ≤ n.
We will use the notation S
to denote the state of p i after the last message (M n ). Clearly, we have pr j ∈ COV ER(S M j j ). Since we are only interested in the number of partial results communicated, we will use the sequence of messages in an algorithm to represent the algorithm. Let a message sequence be
We say that the sequence of messages is valid
Clearly, any all-reduce algorithm can be represented by a valid sequence of messages. Similarly, any valid sequence of messages can be realized by an algorithm.
Lemma 1: The minimum number of partial results to be communicated to complete a one-item all-reduce operation on N processes over all algorithms is at most 2 × (N − 1).
Proof:
The proof is straight-forward by constructing an algorithm that communicates 2 × (N − 1) partial results to complete the operation. The all-reduce operation can be completed with a flat tree reduce algorithm that communicates N − 1 partial results followed by a flat tree broadcast algorithm that communicates N − 1 partial results. 2 Lemma 2: The number of partial results to be communicated to complete a one-item all-reduce operation on N processes using any algorithm is at least 2 × (N − 1).
The lemma applies both when the messages are within the N processes and when additional processes are used to relay messages. We prove the lemma by induction on N .
Base case: N = 2. With one message, at most one process can send its data to the other process, and at most one process can compute a 0 ⊕ a 1 , which cannot complete the all-reduce operation. Hence, at least 1 + 1 = 2 = 2 × (N − 1) communications are needed.
Induction case: The induction hypothesis is that the minimum number of partial results to be communicated in order to complete a one-item all-reduce operation on N processes is at least 2×(N −1). Using this hypothesis, we will prove that the minimum number of messages required to complete the operation on N + 1 processes is at least 2 × ((N + 1) − 1) = 2N .
Let the N + 1 processes be p 0 , p 1 , ..., p N with p i having initial item a i before the operation.
Let the minimum number of messages required to complete the operation on the N + 1 processes be X+1, the ordered sequence of messages be E 1 , E 2 , ..., E X , E X+1 , where E j = (s j → d j , pr j ), 1 ≤ j ≤ X + 1, and the state of process
denotes the state after E X+1 (the last message). We have r = a 0 ⊕a 1 ⊕.....⊕a N ∈ COV ER(S
We will prove X + 1 ≥ 2N by constructing a sequence of at most X − 1 messages that allows the all-reduce operation on N processes to be completed, which requires at least 2N − 2 messages (induction hypothesis). Let us assume that p N is the process that receives the last message (this assumption is valid since ⊕ is associative and commutative). The new sequence is constructed from the X-message sequence E 1 , E 2 , ..., E X , in two steps.
In the first step, we construct a sequence F of X messages as follows. For each
Message sequence F is exactly the same as the message sequence E except that the effect of a N is removed. Note that some messages in sequence F may contain the empty partial result (⊥). Such empty messages will be removed in the next step. Let the state of process
i (sequence F is applied from the initial condition of the operation). By induction on j, it is straight-forward to show that
If pr ∈ COV ER(S
Using the message sequence E, before the last message (E X+1 ), the final result r = a 0 ⊕ a 1 ⊕ ..... ⊕ a N must be in the coverage set of each of p 0 , ..., p N −1 since these processes do not receive the last message and their states do not change after the message. Hence,
are in all of the N processes after the X messages in sequence F . Next, we will construct from sequence F a sequence of at most X − 1 messages that allows the all-reduce operation on N processes p 0 , p 1 , ..., p N −1 to complete. There are two cases.
Case (a): In message sequence F , if there is any message whose content is ⊥ (empty message), this message can be removed and we obtain a sequence of at most X − 1 messages that allows the all-reduce operation on N processes to complete.
Case (b): If there are no such messages in sequence F , each message in F contains a real partial result. Since r = a 0 ⊕ ... ⊕ a N is in the coverage set of all processes p 0 , p 1 , ..., p N −1 after messages E 1 , E 2 , ..., E X , process p N must be the sender of at least one of these messages so that the effect of a N can be distributed to other processes. Since we do not change senders and receivers from sequence E to sequence F , process p N also must be the sender at least once in sequence F .
Let the first message in sequence F with p N as the sender be
We create a new message sequence from sequence F as follows: (1) Combining Lemma 1 and Lemma 2, we obtain the lower bound result in the following lemma. Note that while we prove the lower bound for the case when the reduce operator is commutative, the bound also applies when the reduce operator is non-commutative since the set of all all-reduce algorithms with a non-commutative reduce operator is a subset of the set of all all-reduce algorithms with a commutative reduce operator.
Lemma 3:
The minimum number of partial results to be communicated to complete a one-item all-reduce operation on N processes over all algorithms is 2 × (N − 1). 2 Lemma 4: Assume that (1) data are not compressed during the all-reduce operation; and (2) initial items are independent of one another. To perform an all-reduce operation on X items with itsize bytes item size on N processes, there exists at least one process that must communicate a total of at least 2 × N −1 N × X × itsize bytes data assuming the minimum unit for communication is an item.
Proof: Since the minimum number of items to be communicated in order to complete a one-item all-reduce operation is 2 × (N − 1) (Lemma 3), the minimum number of items to be communicated to complete an X item all-reduce operation is X × 2 × (N − 1). Since there are N processes that carry out the operation collectively, the communications can be distributed among the N processes. Hence, at least one process needs to communicate
Lemma 5: Under the assumptions stated in Lemma 4, the minimum time for the all-reduce operation is at least T ( 2 ×
Let msize = X × itsize be the total data size in the operation. Using the linear commu-
. This indicates that the best of any all-reduce algorithm can do is to achieve roughly 2 times the time to send the msize-byte data.
A ring-based bandwidth optimal algorithm
We will describe a ring-based bandwidth optimal all-reduce algorithm for tree topologies, and show how such an algorithm can be applied to high-end clusters. The tree topology by itself is not particularly interesting. We develop bandwidth optimal algorithms for the tree topology only because tree provides the minimum connectivity and most networks have tree embeddings: our bandwidth optimal all-reduce algorithm for trees can be applied to any topology with a tree embedding. Note that since our algorithm achieves bandwidth optimality, no algorithms can have a better theoretical performance on any topology, assuming that each node has one network port.
Bandwidth optimal algorithm for tree topologies
Our proposed bandwidth optimal all-reduce algorithm combines three existing ideas: (1) realizing the all-reduce operation by a reduce-scatter operation followed by an all-gather operation [22] , (2) realizing both the reduce-scatter operation and the all-gather operation using logical ring based algorithms [7, 25] , and (3) constructing a contention-free logical ring on the tree topology [7] . Our contribution lies in combining these three components in an all-reduce algorithm and in showing that the resulting all-reduce algorithm is bandwidth optimal.
To achieve bandwidth optimality using the reduce-scatter followed by all-gather algorithm [22] , both the reduce-scatter and the all-gather operations must be bandwidth optimal, which means that: (1) each process should communicate the minimum amount of data required to realize the operations; and (2) the network contention must be avoided. The optimality is achieved by using logical ring based reduce-scatter and all-gather algorithms, and by finding a contention-free logical ring on the tree topology. The logical ring all-gather algorithm is described in [25] and the technique to compute a contention-free logical ring on a tree topology can be found in [7] . Next, we will describe the logical ring reduce-scatter scheme.
Let the N processes be p 0 , p 1 , ..., p N −1 . Let F : {0, ..., N − 1} → {0, ..., N − 1} be a one-to-one mapping. Thus,
The logical ring pattern contains the following communications:
Using the logical ring pattern, the reduce-scatter operation is performed as follows. First, the msize source data in each process is partitioned into N segments, all segments having Figure 1 shows the logical ring implementation of reduce-scatter on four processes. As shown in the figure, in
to n F (3) ; and n F (3) sends SEG 2 to n F (0) . After the communication, the reduction operation is performed: n F (0) on SEG 2 , n F (1) on SEG 3 , n F (2) on SEG 0 , and n F (0) on SEG 1 . After three iterations, n F (0) has SEG 0 results, n F (1) has SEG 1 results, n F (2) has SEG 2 results, and n F (3) has SEG 3 results. Put it all together, the proposed algorithm first finds a contention-free logical ring on the tree topology. It then performs the all-reduce operation over the contention-free logical ring by using logical ring algorithms to carry out a reduce-scatter operation followed by an allgather operation. Both the ring based reduce-scatter and all-gather algorithms perform the communications in N − 1 rounds with each node sending msize N bytes data in each round.
Theorem 1:
Consider an all-reduce operation on X items with the size of each item being itsize (msize = X ×itsize). When X is divisible by N , the proposed algorithm is bandwidth optimal.
Proof: Both the reduce-scatter operation and the all-gather operation are performed in N − 1 steps with each process sending and receiving in all processors will be the same after the ring-based algorithm. Hence, the algorithm meets the minimum MPI quality requirement [23] . However, the bracketing for computing the results is not the same for different data items.
Algorithm for high-end clusters
A high-end SMP/Multi-core cluster is typically formed by SMP and/or multi-core nodes connected by a high speed interconnect. The interconnect is usually a single cross-bar switch for small clusters and a fat-tree for large clusters, whose performance is close to a cross-bar switch. The communication between processors in different SMP nodes (internode communication) goes through the interconnect while the intra-node communication is performed within an SMP node, typically through memory operations. For the common case when one network interface card is equipped in each node, the system can be approximated by a two-level tree topology as shown in Figure 2 . The algorithm described in the previous subsection can be applied when the tree topology is determined. In a high-end SMP cluster, even though the system can be approximated with a two-level tree, the exact tree topology is unknown until the SMP nodes are allocated.
Hence, this algorithm cannot be directly used in a high-end SMP cluster since we need to develop the all-reduce routine before it can be used in programs running on a particular platform. However, by default, many SMP clusters assign MPI processes with consecutive ranks to processors (or cores) in each SMP node. For example, in a system with 4-core SMP nodes, n 0 , n 1 , n 2 , and n 3 are assigned one SMP node (one process per core); n 4 , n 5 , n 6 , and n 7 are assigned to another SMP node; and so on. For such clusters, using the 2-level tree representation, a contention-free logical ring for the default process assignment scheme can be built without knowing the exact topology as shown in the following lemma, which can be proved by examining the topology in Figure 2 .
Lemma 6: Under the assumption that MPI processes with consecutive ranks are assigned to processors (or cores) in each SMP node, logical ring pattern,
Experiments
Based on the algorithms described in the previous section, we implement all-reduce routines in two forms. For high-end SMP/multi-core clusters, we implement a stand-alone all-reduce routine based on Lemma 6. For clusters with a physical tree topology, we develop a routine generator that takes the tree topology information as input and automatically produces an all-reduce routine that uses the topology specific algorithm for the topology. The routine generator reads a topology description file that describes the connectivity among machines and switches, determines the order of the machines that realizes a contention-free ring, and produces the code that realizes the operations using the contention-free ring communication The latest OPENMPI 1.2.5 [20] and MPICH 171.132.149.166p1 [10] are installed on this cluster.
The performance of the algorithms is measured using the Mpptest approach [11] . We compare the proposed algorithms with native MPI implementations on these clusters. In addition, the traditional butterfly-like algorithm [22, 23, 27] , denote as butterfly, that is theoretically both latency and bandwidth optimal when the network contention is not a problem is also compared. For high-end SMP/multi-core clusters, we also compare the proposed scheme with two SMP specific implementations. One SMP specific implementation is based on algorithms recently developed for SMP clusters [26] . The implementation, denoted as SMP-binomial, has four logical phases: (1) an intra-node reduce operation for each SMP node using a binomial tree, (2) an inter-node reduce operation with a binomial tree to obtain the reduction results in one node, (3) an inter-node broadcast operation with a binomial tree to distribute the results to each SMP node, and (4) an intra-node broadcast to distribute the results to each processor. While this algorithm does not cause network contention, the inter-node communication is not bandwidth optimal: there exists a node that communicates O(log(N ) * msize) data. We enhance this algorithm by using the butterfly-like algorithm to perform inter-node all-reduce operation. This algorithm is denoted as SMP-butterfly. Notice that the main competitor of the proposed algorithm is butterfly whose main problem is that the communication pattern cannot be embedded in an SMP cluster without causing link contention. SMP-butterfly can be considered as an improvement over butterfly: eliminating the network contention by grouping intra-node communications together.
NCSA Teragrid IA-64 Linux cluster results for our bandwidth optimal algorithm to be more efficient is around 2KB on this cluster. As the data size increases, the performance difference is more significant: the network contention problem is more severe as the data size increases. This experiment shows that the break-even point for the proposed ring-based algorithm is 256KB when N = 128.
When N is larger, however, the break-even point can be much larger since the number of communication rounds in the ring-based algorithm is O(N ).
Draco cluster results Figure 4 shows the results on the Draco cluster. The native MPI library also uses butterfly when the data size is larger than 32KB. In this cluster, the trend for the relative performance of SMP-binomial, SMP-butterfly, and the proposed algorithm is similar to that in the NCSA cluster. The proposed algorithm performs better than native (butterfly) significantly for data sizes 64KB, 128KB, 256KB, 2MB and 4MB. However, for data sizes 512KB and 1MB, the performance of the proposed algorithm is slightly worse. We believe this is mainly due to the interaction between memory references and network contention, which results in the abnormal performance results for the 512KB and 1MB data points. Nonetheless, the figure shows the trend of the relative performance of native (butterfly) and the proposed algorithm:
the proposed algorithm is more effective as the data size becomes larger. 
Ethernet switched cluster results
For the Ethernet switched cluster, we show the results on 32-node clusters with the topology shown in Figure 5 . We performed experiments on other topologies with multiple switches and the trend is similar. Figure 6 show the results for the Ethernet cluster. On this cluster, the proposed algorithm is more efficient than both Open MPI and MPICH2 when the data size is larger than 256KB. Hence, the threshold value of msize N for the proposed algorithm to be more efficient is 256KB 32 = 8KB, which is much larger than the 2KB in the high-end NCSA cluster. This reflects the fact that on an Ethernet switched cluster with Gigabit switches, the communication start-up overhead is much larger than that in high-end clusters. Thus, even
for reasonably large data sizes (e.g. 128KB), it is still more important to reduce the communication start-up overheads. However, when the data size is larger, network contention and bandwidth efficiency become a problem and our bandwidth optimal algorithm is more efficient.
n0 n1 n15 n16 n17 n31 S0 S1 The all-reduce operation has been extensive studied. The one-item all-reduce operation has been studied under different names such as census function [2] , global combine [3, 5, 6, 27] , and gossip [16] . The lower bound for the communication time under various communication models has been established [2, 3, 5, 16] . In [16] , it is shown that to complete a one-item allreduce operation under the telephone model, at least 2N −4 connections must be established when N > 4. In [2, 3, 5] , the lower bounds for the number of rounds of communications and for the number of data item to be communicated in sequence in various postal models are established. However, to the best of our knowledge, the lower bound on the total number of data items to be communicated to complete the operation has not been established.
The all-reduce operation is one of the collective operations supported in the MPI standard [18] , and thus, all MPI libraries support this operation. Many efficient platform independent algorithms for this operation have been proposed [22, 23, 25, 27] . In [22] , Rabenseifner
proposed to realize the all-reduce operation by a reduce-scatter operation followed by an all-gather operation and gave various algorithms for the reduce-scatter and all-gather operations. The butterfly-like algorithm has been developed some times ago [22, 27] and has been extended to handle non-power-of-two numbers of processes [23] . Various architecture specific all-reduce schemes have also been developed [1, 4, 12, 17, 26] . An all-reduce algorithm was designed for BlueGene/L systems in [1] . In [12] , an all-reduce scheme that takes advantage of remote DMA (RDMA) capability was developed for VIA-based clusters. The work in [17] investigated an adaptive all-reduce algorithm in an InfiniBand cluster that deals with the situation when not all nodes arrive at the call site at the same time. A study on the all-reduce operation over WAN can be found in [4] . All-reduce algorithms were developed specifically for SMP clusters in [26] . Among all these algorithms, the butterfly-like algorithm [22, 23, 27] is widely used. The limitation of this algorithm is that it is difficult to realize the butterfly communication pattern without incurring network contention in contemporary SMP/multi-core clusters. Many all-gather [7, 24, 25] and reduce-scatter [13, 24, 25] algorithms, which are parts of the proposed all-reduce algorithm, have been developed. The technique for finding contention-free logical ring on tree topologies was developed in [7] . In this paper, we use the existing results in various papers (e.g. [7, 22, 25] ) as the components of our proposed algorithm. Our contribution lies in showing that putting these existing components in the particular way yields a contention-free bandwidth optimal all-reduce algorithm for the tree topology. Like other architecture dependent collective algorithms [8, 7, 15, 21] that work well in some situations, the proposed scheme can be used in advanced communication systems [9, 14, 28] .
Conclusions
We investigate efficient implementations of the all-reduce operation with large data sizes under the assumption that the reduce operator is both associative and commutative. We derive a theoretical lower bound on the communication time of this operation and develop a bandwidth optimal all-reduce algorithm on tree topologies. This algorithm only requires tree connectivity to achieve bandwidth optimality and can be applied to contemporary clusters. We demonstrate the effectiveness of the proposed algorithm on various contemporary clusters, including high-end clusters with SMP and/or multi-core nodes connected by high-speed interconnects, and low-end Ethernet switched clusters. While our algorithm is contention-free and bandwidth optimal, it is not optimal in the latency term: the number of communication rounds is proportional to the number of processes.

{'paper_id': '7433454', 'title': 'Bandwidth Optimal All-reduce Algorithms for Clusters of Workstations', 's2_url': 'https://api.semanticscholar.org/CorpusID:7433454', 'source': 'S2ORC'}