arXiv:2605.29970v1 [cs.DC] 28 May 2026
Effective MPI: User-defined Datatypes and Cartesian Communicators for Zero-copy All-to-all Communication in Multidimensional Tori Jesper Larsson Träff TU Wien Faculty of Informatics Institute of Computer Engineering, Research Group Parallel Computing 191-4 Treitlstrasse 3, 5th Floor, 1040 Vienna, Austria May 2026
Abstract We present and show how to implement a non-trivial all-to-all communication algorithm for arbitrary d-dimensional tori effectively in MPI. Given a factorization of the number of processes p into d factors that can be mapped onto a d-dimensional torus, we first utilize a Cartesian communicator to split a given p-process MPI communicator into, for each MPI process, d smaller communicators spanning each of the dimensions of the torus to which the process belongs, and cache these communicators in order to avoid expensive splitting at each all-to-all operation. The all-to-all operation itself is decomposed into a sequence of d MPI Alltoall operations on the dimension-wise communicators. The non-trivial data rearrangement before and after each MPI Alltoall call is implicit only and effected by MPI derived datatypes. This makes the implementation of the algorithm formally zero-copy, meaning that no explicit process-local reordering of data blocks ever has to be performed. In order to achieve this, the algorithm employs a double-buffering scheme with modest temporary buffer requirements. By choosing the factorization of p and selecting appropriate implementations for the component MPI Alltoall operations, the presented implementation gives ample opportunities for algorithm tuning and adaptation to the particular highperformance system. A few, select experimental results show competitive performance with native MPI Alltoall implementations and illustrate problems that common MPI Alltoall implementations may have.
1
Introduction
The all-to-all personalized communication problem, in which each process in a set of processes has an individual block of data (of some given number of elements) to each of the other processes, arises naturally in many important applications, but is also the most communication intensive and expensive collective data exchange problem. The operation should therefore be used judiciously, while at the same time efficient and well-implemented algorithms are needed. All-to-all communication is standardized in MPI as the regular MPI Alltoall collective and the irregular MPI Alltoallv and MPI Alltoallw collectives [7]. We present a paradigm for potentially efficient implementations of the all-to-all operation in terms of smaller component all-to-all operations, which shows how such a collective operation can be expressed in terms of more specialized uses of the same collective operation. Such a decomposition is possible with MPI functionality and highlights how MPI can be used effectively 1
to efficiently implement non-trivial algorithms. Such implementations of MPI functionality in terms of other MPI functionality naturally poses upper bounds on the concrete performance of, in the case here, the MPI Alltoall collective in any given MPI library as so-called self-consistent performance guidelines [12]. Such performance guidelines can be automatically checked and strengthen ones confidence in the performance and quality of a given MPI library [5]. Essentially, a performance guideline states that for a given collective operation, there is no immediate better implementation in terms of MPI itself. We can view the all-to-all algorithm that we present here in three different ways. 1. As a dedicated algorithm for a torus communication network of any number of dimensions, in which communication is restricted to be between processes (or nodes of processes) that are neighbors along one of the dimensions in the torus. With d dimensions, the algorithm takes d communication rounds. 2. As an algorithm for the all-to-all problem for any given or desired factorization of the number of processes p. Viewed in this way, the algorithm can be adopted to the given communication network and partition by choice of factorization and choice of implementations for the all-to-all problems to be solved on the factors. With d factors, the algorithm takes d communication rounds. For a power of two number or processes, p = 2d , this is log2 p, and the resulting algorithm can be seen as a natural generalization of well-known hypercube algorithms. 3. As message-combining performance guideline implementations of the all-to-all operation against which the MPI library native MPI Alltoall function can be judged by comparative benchmarking [5]. All three viewpoints entail combining of blocks of elements from and to different processes. In each round different combinations of blocks are needed. We show how such composite blocks can be put together implicitly, at the corresponding component MPI Alltoall call, by expressing the sequences of blocks as derived datatypes. The communicators for the component MPI Alltoall operations are created in advance and cached so that the expensive creation of the needed subcommunicators can be amortized over many all-to-all operations. The paper has the didactic purpose of showing the effectiveness of MPI mechanisms (Cartesian communicators, communicator splitting, attribute caching and derived datatypes) for the implementation of non-toy examples.
2
Related Work
Standard all-to-all algorithms for two- and three-dimensional meshes and hypercubes are explained in [4], see also [9] and [1] for full-ported hypercubes. Message combining all-to-all algorithms like Bruck [2], and for more general circulant graphs [11], are well-known and used in many different forms, also in MPI libraries. Message-combining can reduce the number of communication rounds d down to the optimal d = ⌈log2 p⌉, but at an inherent cost of (re)sending roughly half the data blocks in each round [2]. An implementation of the Bruck algorithm using MPI derived datatypes to reorganize and pipeline data blocks was given in [17]. Multidimensional torus all-to-all algorithms with one-ported communication were presented and implemented in [10,18,19], and elsewhere, and use links along only one dimension at a time. The simple d = 2 algorithm for the BlueGene/L system in [20] attempts to use multiple links simultaneously. Two-level decompositions of the MPI Alltoall operation, also relying on MPI derived datatypes were given in [13, 16]; in the latter the decomposition in particular follows 2
the compute node structure of the given cluster and performs the all-to-all communication as concurrent intra-node all-to-all operations followed by concurrent inter-node all-to-all operations (or the other way round). The algorithm of this paper extends this implementation to arbitrary number of dimensions and deeper processor hierarchies. A somewhat similar algorithm, for two levels only, viewing the MPI all-to-all problem as a matrix transposition problem, was given in [3]. Our implementation precomputes and uses information stored with communicators to decide on the details of the algorithms to be executed. This viewpoint was discussed in more detail specifically for MPI in [14]. Some problems related to MPI Dims create that also appear in this paper were discussed and solved in [15], see also [8].
3
A d-dimensional Torus All-To-All Algorithm
Assume that a p-process(or) communication structure (network, system, artifact), say a ddimensional torus, is given and can be described as a Cartesian graph product MdD = G0 2 G1 2 . . . 2Gd−1 with graph factors Gi = (Vi , Ei ) and p = Πd−1 i=0 D[i] = |V0 | × |V1 | × . . . |Vd−1 |. D When Md is a torus with dimension orders D[i], i = 0, 1, . . . , d − 1, the factors Gi are ring communication structures, when MdD is a mesh, the factors are linear processor arrays; but the factors could be (isomorphic to) other graphs (like fully connected networks) as fitting to the actual hardware communication capabilities of the given system, as long as they, for the algorithm to be described, admit all-to-all communication. Our d-dimensional torus all-to-all algorithm decomposes the all-to-all operation into concurrent all-to-all operations on smaller communication structures along the dimensions of the torus, each isomorphic to a factor Gi of the graph factorization. On dimension i, 0 ≤ i < d, a smaller all-to-all operation on D[i] processors will exchange p blocks of elements (the blocks to the processor itself do not have to be communicated), and therefore the communication volume complexity will be proportional to both p and d roughly as pd (with a constant factor smaller than one). The algorithm works for any number of torus dimensions and for any factorization of p into d factors (in any chosen order). In particular, if p = 2d is a power of two, the resulting algorithm will be equivalent to well-known hypercube algorithms for all-to-all communication. Algorithm 1 The generic all-to-all algorithm for processor r ∈ MdD in a d-dimensional Cartesian product MdD = G0 2G1 2 . . . 2Gd−1 with Gk = (Vk , Ek ). Processor r has p input blocks in S[p], one for each processor, and receive a block from each processor in R[p]. procedure AllToAll(S[p], R[p], r ∈ MdD ) O ← rank-to-vector(r, d, D) b←0 T [b] ← S for k = 0, 1, . . . , d − 1 do
▷ Origin vector ▷ Double buffer switch
▷ Dimension-wise all-to-all ′ ▷ Define stride function σ(i) = Πi−1 k′ =0 D[k ] ′ S[σ(k)][σ(k+1)]...[σ(d−1)] [D[k]][D[k + 1]] . . . [D[d − 1]] ≡ T [b] ′ R[σ(k)][σ(k+1)]...[σ(d−1)] [D[k]][D[k + 1]] . . . [D[d − 1]] ≡ T [1 − b] D G ≡ Md [{O[0]} × · · · × {O[k − 1]} × Vk × {O[k + 1]} × · · · × {O[d − 1]}] AllToAll(S ′ , R′ , G) ▷ All-to-all on subgraph G b←1−b end for R ← T [b] end procedure
3
The all-to-all algorithm is shown as Algorithm 1 and relies heavily on a specific notation for indexing in arrays. The input and output arrays S and R are declared as one-dimensional arrays of p blocks (of elements) in linear index order such that S[i] denotes the ith block of elements to be sent to processor i and R[i] the ith block in R which is to be received from processor i. The bracket [i] can be seen as a mapping which gives the index or address of the ith element of the array to which it is applied. Multiple bracket are used, C style [6], for multidimensional arrays. A declared array T [D[d − 1]] . . . [D[0]] denotes a d-dimensional array of order p = D[0] × D[1] × . . . × D[d − 1]. Lookup with d indices T [id−1 ][id−2 ] . . . [i0 ] addresses d−3 element i = id−1 Πd−2 k=0 D[k] + id−2 Πk=0 D[k] + . . . + i0 , corresponding to a row-major organization of the elements of T . Each of the factors Πd−2 k=0 D[k] etc. is the stride used for the given dimension (here d−1). Explicit strides can be used to express other traversals of the array. In our notation T[sd−1 ][sd−2 ]...[s0 ] [id−1 ][id−2 ] . . . [i0 ] will denote the element i = id−1 sd−1 + id−2 sd−2 + . . . + i0 s0 with explicit strides si . For instance, taking si = Πd−2 j=i D[j], the d-dimensional array is handled in column-major order. See the examples at the end of this section. The all-to-all algorithm takes d successive all-to-all communication rounds, in each of which multiple, concurrent all-to-all operations are performed. The processors use two intermediate buffers T [0] and T [1] that are divided into p blocks in the same way as the result buffer R[p]; but indexed differently by each of the all-to-all operations. Let processor r ∈ MdD have coordinates (O[0], O[1], . . . , O[d − 1]). In communication round k, k = 0, 1, . . . , d − 1, processor r shall do an all-to-all exchange with the processors {O[0]} × · · · × {O[k − 1]} × Vk × {O[k + 1]} × · · · × {O[d − 1]} which are the processors of the communication structures G that are used for the all-to-all calls. For each processor, G is the subgraph of MdD restricted to these processors. The allto-all operation on the corresponding communication structures will ensure that processor r has received all blocks from the processors in its G to itself and to all processors with which r will do all-to-all operations in the following rounds. All these blocks are combined into a single composite block for processor r. The communication structures for round k each obviously have size D[k]. The number of blocks that must be sent and received (the blocks that a processor p communicates with itself not included) in round k is (D[k] − 1) D[k] , since the number of blocks per processor is p/D[k]. In the first round with k = 0, processor r has one block to each of the other processors with which it communicates and to the processors with which these will communicate in later rounds. After the round, processor r has received D[0] blocks to itself (from D[0] − 1 other processors and itself), and also all the segments of blocks to the processors with which r will later communicate now consist of D[0] blocks. In each round, the segments of blocks that are exchanged in the all-to-all communication get larger by a factor of D[k] blocks, such that after the last round, each processor will have received Πd−1 k=0 D[k] = p blocks, including the block from itself. After each round, the blocks to processor r and the processors with which r will exchange blocks in the later rounds will form consecutive segments of the T [0] and T [1] buffers with each such segment consisting of Πk−1 i=0 D[i] blocks. The segments will be put together by traversing the buffers as (d − 1 − k)-dimensional arrays in column-major order. For round k, this is expressed as ′ S[σ(k)][σ(k+1)]...[σ(d−1)] [D[k]] [D[k + 1]] . . . [D[d − 1]] {z } | {z }| Composite blocks
Segment indices
... |{z}
Consecutive Πk−q i=0 D[i] block segment
where σ(k) = Πk−1 i=0 D[i] is the stride to be used for the the corresponding index. The first index (from the left) is the number of composite blocks and therefore equal to the size D[k] (number of processors) of G. The number of indexed segments is Πd−1 i=k D[i] and the size of the segments 4
Πk−1 i=0 D[i] = σ(k). The second index enumerates the segments for the next dimension, and are therefore strided by σ(k + 1) blocks; and so on for the remaining indices up to d − 1 − k. We note that the stride for the first (leftmost) index equals the size of the consecutive segments for round k. Theorem 1. On a d-dimensional torus MdD , Algorithm 1 correctly solves the all-to-all problem in d successive rounds of concurrent all-to-all operations. There are p/D[k] concurrent all-to-all operations in round k, k = 0, 1, . . . , d − 1, each of which involves D[k] processors. The number of elements in the blocks that are sent and received in round k is likewise p/D[k]. Per processor D[k] − 1 such blocks are sent and received, and one block is copied from send to receive buffer. The total number of blocks sent and received per processor is d−1 X D[k] − 1 k=0
D[k]
p = dp −
d−1 X k=0
p D[k]
.
Proof. The complexity in number of all-to-all operations and number and sizes of blocks follows immediately from the description of the d rounds of the algorithm. For correctness, Algorithm 1 maintains the following invariant: Before round k, processor r = (O[0], . . . , O[d − 1]) has collected a block to each of the processors in Vd−1 × . . . × Vk × {O[k − 1]} × . . . × {O[0]} from each of the processors in {O[d − 1]} × . . . × {O[k]} × Vk−1 × . . . × V0
.
The invariant holds before the first iteration k = 0, since each processor has in the input buffer S[p] a block to each of the other processors. The invariant implies that after the last iteration, processor r has received a block from each of the other processors. Assume that the invariant holds before iteration k. By the all-to-all operation, each processor sends the blocks that it has for all processors in {O[0]} × . . . {O[k − 1]} × Vk × {O[k + 1]} × . . . × {O[d − 1]} and receives the blocks that the other processors have. Except for the copy operations before the first iteration of S into T [0] and after the last iteration of T [b] into R, there are no explicit copy operations. All exchanges and reordering of elements is handled implicitly by the traversal of the temporary buffers (as (d − 1 − k)dimensional arrays) and the all-to-all operations. The first and last copies can easily be avoided in a real implementation by using R as one of the intermediate buffers and making sure that this will be used as receive buffer in the last all-to-all call, and by taking the block directly from the send buffer S in the first iteration. Doing this makes the implementation formally a zero-copy implementation. Three examples: Some examples of how the algorithm works, in particular in what index order the temporary arrays are addressed will surely be helpful. Let p = 5 × 4 = 20 = D[0] × D[1] be a two-dimensional factorization of p. Algorithm 1 first performs all-to-all operations on communication structures with D[0] = 5 processors, then on communication structures with D[1] = 4 processors. The index sequences for the S ′ and R′ arrays in each round are shown in the table below. As can be seen, in each of the two rounds, all p indices 0, 1, . . . p − 1 are listed, meaning that p blocks are sent and received for each processor r = [O[0], O[1]], except of course the blocks for processor r itself. These are given by T [O[0]] for round 0 and T [O[1]] for round 1. As can also be seen, the blocks for the last round consist of consecutively indexed elements. 5
k R′ [0] R′ [1] R′ [2] R′ [3] R′ [4]
R′ [0] R′ [1] R′ [2] R′ [3]
0
′ R[1][5] [5][4] = [0, 1, 2, 3, 4][0, 5, 10, 15]
= = = = =
[0, 5, 10, 15] [1, 6, 11, 16] [2, 7, 12, 17] [3, 8, 13, 18] [4, 9, 14, 19]
1
′ [4] = [0, 5, 10, 15][0, 1, 2, 3, 4] R[5]
= = = =
[0, 1, 2, 3, 4] [5, 6, 7, 8, 9] [10, 11, 12, 13, 14] [15, 16, 17, 28, 19]
Now, let instead p = 2 × 3 × 4 = 24 = D[0] × D[1] × D[2]. The index sequences for the R′ arrays in each round are shown in the table below. As can be seen, the blocks for the last round consist of consecutively indexed elements. k R′ [0] R′ [1]
R′ [0] R′ [1] R′ [2]
R′ [0] R′ [1] R′ [2] R′ [3]
0
′ R[1][2][6] [2][3][4] = [0, 1][0, 2, 4][0, 6, 12, 18]
= =
[0, 6, 12, 18, 2, 8, 14, 20, 4, 10, 16, 22] [1, 7, 13, 19, 3, 9, 15, 21, 5, 11, 17, 23]
1
′ R[2][6] [3][4] = [0, 2, 4][0, 6, 12, 18][0, 1]
= = =
[0, 1, 6, 7, 12, 13, 18, 19] [2, 3, 8, 9, 14, 15, 20, 21] [4, 5, 10, 11, 16, 17, 22, 23]
2
′ [4] = [0, 6, 12, 18][0, 1, 2, 3, 4, 5] R[6]
= = = =
[0, 1, 2, 3, 4, 5] [6, 7, 8, 9, 10, 11] [12, 23, 14, 15, 16, 17] [18, 19, 20, 21, 22, 23]
Finally, let p = 4 × 3 × 3 × 4 = 144 = D[0] × D[1] × D[2] × D[3]. The index sequences for the R′ arrays in each round are shown in the table below. As can be seen, the blocks for the last round consist of consecutively indexed elements.
6
k R′ [0] R′ [1] R′ [2] R′ [3]
R′ [0] R′ [1] R′ [2]
R′ [0] R′ [1] R′ [2]
R′ [0] R′ [1] R′ [2] R′ [3]
4
0
′ R[1][4][12][36] [4][3][3][4] = [0, 1, 2, 3][0, 4, 8][0, 12, 24][0, 36, 72, 108]
= = = =
[0, 36, 72, 108, 12, . . . , 32, 68, 104, 140] [1, 37, 73, 109, 13, . . . , 33, 69, 104, 141] [2, 38, 74, 110, 14, . . . , 34, 70, 105, 142] [3, 39, 75, 111, 15, . . . , 35, 71, 106, 143]
1
′ R[4][12][36] [3][3][4] = [0, 4, 8][0, 12, 24][0, 36, 72, 108][0, 1, 2, 3]
= = =
[0, 1, 2, 3, 36, 37, 38, 39, . . . , 132, 133, 134, 135] [4, 5, 6, 7, 40, 41, 42, 43, . . . , 136, 137, 138, 139] [8, 9, 10, 11, 44, 45, 46, 47, . . . , 140, 141, 142, 143]
2
′ R[12][36] [3][4] = [0, 12, 24][0, 36, 72, 108][0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
= = =
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, . . . , 117, 118, 119] [12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 48, . . . , 129, 130, 131] [24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 60, . . . , 141, 142, 143]
3
′ [4] = [0, 36, 72, 108][0, 1, 2, 3, 4, 5, 6, 7, 8, . . . , 33, 34, 35] R[36]
= = = =
[0, 1, 2, 3, 4, 5, 6, 7, 8, . . . , 33, 34, 35] [36, 37, 38, 39, 40, 41, 42, 43, 44, . . . , 69, 70, 71] [72, 73, 74, 75, 76, 77, 78, 79, 80, . . . , 105, 106, 107] [108, 109, 110, 111, 112, 113, 114, 115, 116, . . . , 141, 142, 143]
The Implementation in MPI
For the implementation of Algorithm 1 in and for MPI, three details have to be handled: 1. Creation of the communication structures G. 2. Traversal of the arrays of blocks of elements. 3. The component all-to-all communications. We give the concrete, full MPI code in C which handles the three issues, showing how the non-trivial all-to-all algorithm can be effectively implemented with MPI. Listing 1: Full MPI code for (graph) factorizing a communicator by communicator splitting following a factorization of the number of processes given as an array of dimension orders. int Comm_factorize ( MPI_Comm comm , const int d , const int order [] , MPI_Comm * factorcomm ) { int rank , size ; int i , p ; torusattr * factors ; p = 1; for ( i =0; i < d ; i ++) { assert ( order [ i ] >1); p *= order [ i ]; }
7
int periodicity [ d ]; for ( i =0; i < d ; i ++) periodicity [ i ] = 1; // make Cartesian communicator with no reorder MPI_Cart_create ( comm ,d , order , periodicity ,0 , factorcomm ); int result ; MPI_Comm_compare ( comm ,* factorcomm ,& result ); assert ( result == MPI_CONGRUENT ); MPI_Comm_rank (* factorcomm ,& rank ); MPI_Comm_size (* factorcomm ,& size ); assert ( size == p ); factors = ( torusattr *) malloc ( sizeof ( torusattr )); assert ( factors != NULL ); factors - > d = d ; factors - > comm = ( MPI_Comm *) malloc ( d * sizeof ( MPI_Comm )); assert ( factors - > comm != NULL ); int origin [ d ]; MPI_Cart_coords (* factorcomm , rank ,d , origin ); for ( i =0; i < d ; i ++) { int f ; p /= order [ i ]; f = rank - origin [ i ]* p ; MPI_Comm_split (* factorcomm ,f , origin [ i ] ,& factors - > comm [ i ]); int factorsize ; MPI_Comm_size ( factors - > comm [ i ] ,& factorsize ); assert ( factorsize == order [ i ]); } MPI_ Comm_set_attr (* factorcomm , toruskey () , factors ); return MPI_SUCCESS ; }
Given a communicator of size p and a factorization of p into d factors, the code in Listing 1 shows how to create subcommunicators corresponding to the subgraphs G of Algorithm 1 to be used by the processes in the later implementation of the all-to-all algorithm. For each MPI process, for all-to-all communication round k, k = 0, 1, . . . , d − 1, a communicator consisting of the process itself and all other processes having the same coordinates in the torus except for the kth coordinate (which is between 0 and order[k]) is created by MPI Comm split. Processes that belong together are colored by the first process which has a kth coordinate of zero, and the call will compute all p/order[k] communicators (subgraphs) for round k. To associate d-dimensional coordinates with the processes, a Cartesian MPI communicator is created for convenience. The creation of the communicators which entails d MPI Comm split calls is expensive. For each process, the d communicators to which this process belongs are therefore cached with the calling communicator for later lookup when the all-to-all operation is performed. For caching, a communicator key value must be created; a possible way of doing this transparently is shown in Listing 2. Listing 2: Surrounding code for caching subcommunicators with a communicator. A key value is kept as a static lookup value that is created at first use. A delete function (beware: with collective semantics) frees created subcommunicators for the process. typedef struct { int d ; MPI_Comm * comm ; } torusattr ;
8
static int torusdel ( MPI_Comm comm , int keyval , void * attr , void * s ) { torusattr * factors = ( torusattr *) attr ; int i ; for ( i =0; i < factors - > d ; i ++) MPI_Comm_free (& factors - > comm [ i ]); free ( factors - > comm ); free ( factors ); return MPI_SUCCESS ; } static int toruskey () { static int toruskeyval = MPI_KEYVAL_INVALID ; // hidden key if ( toruskeyval == MPI_KEYVAL_INVALID ) { M P I _ C o mm_ cr ea te_ ke yv al ( MPI_COMM_NULL_COPY_FN , torusdel ,& toruskeyval , NULL ); } return toruskeyval ; }
Listing 3: MPI implementation of the d factor all-to-all algorithm. A datatype for the sequence of blocks to be communicated is created for each communication round, and freed again. int Alltoall_torus ( void * sendbuf , int sendcount , MPI_Datatype sendtype , void * recvbuf , int recvcount , MPI_Datatype recvtype , MPI_Comm comm ) { int rank , size ; int i , k ; int c , d ; MPI_Comm_rank ( comm ,& rank ); MPI_Comm_size ( comm ,& size ); MPI_Datatype block ; MPI_Aint lb , blockextent ; M PI _ T yp e_contiguous ( recvcount , recvtype ,& block ); M PI _ T yp e_get_extent ( block ,& lb ,& blockextent ); void * tempbuf ; tempbuf = ( void *) malloc ( size * blockextent ); assert ( tempbuf != NULL ); int flag ; torusattr * factors ; MPI_ Comm_get_attr ( comm , toruskey () ,& factors ,& flag ); if (! flag ) // fallback return MPI_Alltoall ( sendbuf , sendcount , sendtype , recvbuf , recvcount , recvtype , comm ); int count , stride [ factors - > d ]; count = 1; for ( i = factors - >d -1; i >=0; i - -) {
9
stride [ i ] = count ; MPI_Comm_size ( factors - > comm [ i ] ,& c ); count = c * count ; } void * in , * out ; out = sendbuf ; in = ( factors - > d %2==0) ? tempbuf : recvbuf ; count = 1; for ( k = factors - >d -1; k >=0; k - -) { // create the contiguous segment MPI_Datatype segment , tile , tiled ; M PI _ T ype_contiguous ( count , block ,& segment ); // column - major parts for ( i =0; i < k ; i ++) { M P I _T y pe _ cr ea t e_ r es iz e d ( segment ,0 , stride [ i ]* blockextent , & tile ); MPI_Type_free (& segment ); MPI_Comm_size ( factors - > comm [ i ] ,& c ); M P I_Type_contiguous (c , tile ,& segment ); MPI_Type_free (& tile ); } M P I _ T y pe _ cr e at e_ r es iz e d ( segment ,0 , count * blockextent , & tiled ); MPI_Type_free (& segment ); MPI_Type_commit (& tiled ); MPI_Alltoall ( out ,1 , tiled , in ,1 , tiled , factors - > comm [ k ]); MPI_Type_free (& tiled ); if ( out == sendbuf ) { if ( in == recvbuf ) { out = recvbuf ; in = tempbuf ; } else { out = tempbuf ; in = recvbuf ; } } else { void * s ; s = in ; in = out ; out = s ; } MPI_Comm_size ( factors - > comm [ k ] ,& c ); count = c * count ; } MPI_Type_free (& block ); free ( tempbuf ); return MPI_SUCCESS ; }
Finally, the implementation of the all-to-all algorithm in Algorithm 1 is shown as Listing 3. The factorization with communicators to be used by the calling process is looked up using the attribute key. We here assume that the communicator has indeed been factorized, if not MPI Alltoall on the full communicator is performed as fallback. The implementation goes through d iterations in each of which an MPI Alltoall call is done. The send and receive buffers in and out are traversed in the order described by the
10
stride function σ(i). These strides are precomputed into the stride[d] array, and used to create the datatype for each iteration. First, a single block of elements is described by the block type. Input and output buffers are such blocks in rank order, but the order in which these blocks are sent and received is determined by the final tiled datatype. Strides are set by resizing component datatypes with MPI Type create resized, these strided segments of blocks are listed in contiguous order by creating contiguous, but tiled types with MPI Type contiguous. As can be seen, the resulting implementation is completely zero-copy in the sense that there are no process-local explicit copying of data whatsoever. This is accomplished by taking blocks directly from the sendbuf in the first iteration and making sure that recvbuf is used as receive buffer in the last iteration. A theoretical drawback of the implementation is that O(d) time is spent creating datatypes in each iteration before the MPI Alltoall call. With the static, derived datatype mechanism of MPI, this cannot be avoided, but in a more flexible programming model, it possibly could.
5
Selected Benchmark Results
We have implemented the generic all-to-all algorithm exactly as explained and shown in the previous sections. We aim to compare it against itself with different factorizations of the number of processes p into d factors, and of course against the MPI library native MPI Alltoall operation for the same inputs. For the latter, we examine three MPI libraries that are available to us. Our test system is a small 36 × 32 processor cluster with 36 dual socket compute nodes, each with two Intel(R) Xeon(R) Gold 6130F 16-core CPUs. The nodes are interconnected via dual Intel OmniPath interconnects each with a bandwidth of 100 GBytes/s. The implementations and benchmarks were compiled with gcc 10.2.1 with the -O3 option. The MPI libraries used are OpenMPI 4.1.6, mpich 4.1.2, and IntelMPI 2021.13. Our benchmark does a simple MPI Alltoall call with preallocated and filled send and receive communication buffers in deciles that are increased by factors of 10 from 1 to 10 000 elements per MPI process. Elements are integers of MPI datatype MPI INT. For each number of elements, 40 measurement repetitions are done with 8 prior warmup measurements. The best seen completion times (of the slowest process) are plotted. Processes are synchronized before each measurement with a double MPI Barrier call. We experiment with different number of dimensions from 2 to ⌈log2 p⌉ and use MPI Dims create to do the actual factorization of the number of processes. We assume that MPI Dims create behaves according to the MPI specification [15] and returns a factorization where the factors are as close to each other as possible. This is not quite the case for the OpenMPI 4.1.6 library as can be seen in Table 1. This problem was addressed in [15] and can be solved. We run the new all-to-all implementation in a single full-system MPI process configuration with p = 36 × 32 and compare different factorizations of p against the library native MPI Alltoall. We have tried with d = 2, 3, 4 factors and ⌈log2 p⌉ = 9 factors, see Table 1 for the specific factorizations. Results with IntelMPI 2021.13 are shown in Figure 1, with OpenMPI 4.1.6 in Figure 2, and with mpich 4.1.2 in Figure 3. The different MPI libraries lead to similar conclusions, but have many specific and conspicuous differences. First, with up to, say, 100 MPI INT elements per process, the d = 2, 3 factor algorithm seems to be considerably faster than library native MPI Alltoall, in many cases by a factor of two or more. For larger number of elements, a (presumably) direct algorithm is faster. The highest possible dimensional d = ⌈log2 p⌉ round algorithm is never competitive. The OpenMPI 4.1.6 library exhibits a severe MPI Alltoall performance (guideline violation) problem for medium sized blocks from about 80 to 800 MPI INT elements, where performance unexpectedly degrades by a factor of 11
Table 1: The factorizations of the given number of processes p for different number of dimensions d as used in the experiments. The factorization is computed (presumably) in a portable way by MPI Dims create. As can be seen, the OpenMPI 4.1.6 library computes a factorization that is not in accordance with the MPI specification. d 2 3 4 ⌈log2 p⌉
IntelMPI 2021.13 mpich 4.1.2
OpenMPI 4.1.6
p = 36 × 32 p = 12 × 12 × 8 p=8×6×6×4 p=3×3×2×2×2×2×2×2×2
p = 48 × 24 p = 12 × 12 × 8 p=8×6×6×4 p=3×3×2×2×2×2×2×2×2
MPI_Alltoall, 36x32 MPI processes, Intel MPI library
Time (microseconds)
1x107
MPI_Alltoall 2-Factor 3-Factor 4-Factor log-Factor
1x106 100000 10000 1000 100
1
10
100
1000
Count (MPI_INT) per process
Figure 1: Results for the IntelMPI 2021.13 library.
12
10000
MPI_Alltoall, 36x32 MPI processes, OpenMPI library
Time (microseconds)
1x107
MPI_Alltoall 2-Factor 3-Factor 4-Factor log-Factor
1x106 100000 10000 1000 100
1
10
100
1000
10000
Count (MPI_INT) per process
Figure 2: Results for the OpenMPI 4.1.6 library.
MPI_Alltoall, 36x32 MPI processes, mpich library
Time (microseconds)
1x106
MPI_Alltoall 2-Factor 3-Factor 4-Factor log-Factor
100000
10000
1000
100
1
10
100
1000
Count (MPI_INT) per process
Figure 3: Results for the mpich 4.1.2 library.
13
10000
more than 10. Therefore, MPI libraries may have room for improvement by message-combining implementations for smaller inputs.
6
Summary
We presented an extremely compact, highly non-trivial algorithm for all-to-all communication using a factorization of the number of processes to decompose the problem into smaller component all-to-all problems. We showed how this algorithm could be effectively and transparently be implemented in MPI by using derived datatypes, subcommunicators and attribute caching on communicators. The resulting algorithm could be used as performance guideline against which to compare library native MPI Alltoall implementations, and as a template for possibly more efficient implementations in MPI libraries by careful selection of the algorithms for the component MPI Alltoall calls and by choosing a factorization and splitting of the communicators that map well to the given, physical communication system.
References [1] Dimitri P. Bertsekas, C. Özveren, George D. Stamoulis, Paul Tseng, and John N. Tsitsiklis. Optimal communication algorithms for hypercubes. Journal of Parallel and Distributed Computing, 11(4):263–275, 1991. [2] Jehoshua Bruck, Ching-Tien Ho, Schlomo Kipnis, Eli Upfal, and Derrick Weathersby. Efficient algorithms for all-to-all communications in multiport message-passing systems. IEEE Transactions on Parallel and Distributed Systems, 8(11):1143–1156, 1997. [3] George Chochia, David G. Solt, and Joshua Hursey. Applying on node aggregation methods to MPI alltoall collectives: Matrix block aggregation algorithm. In EuroMPI/USA’22: 29th European MPI Users’ Group Meeting, pages 11–17. ACM, 2022. [4] Ananth Grama, George Karypis, Vipin Kumar, and Anshul Gupta. Introduction to Parallel Computing. Addison-Wesley, second edition, 2003. [5] Sascha Hunold, Alexandra Carpen-Amarie, Felix Donatus Lübbe, and Jesper Larsson Träff. Automatic verification of self-consistent MPI performance guidelines. In Euro-Par Parallel Processing, volume 9833 of Lecture Notes in Computer Science, pages 433–446. Springer, 2016. [6] Brian W. Kernighan and Dennis M. Ritchie. The C Programming Language. Prentice-Hall, second edition, 1988. [7] MPI Forum. MPI: A Message-Passing Interface Standard. Version 4.1, November 2nd 2023. www.mpi-forum.org. [8] Christoph Niethammer and Rolf Rabenseifner. An MPI interface for application and hardware aware Cartesian topology optimization. In 26th European MPI Users’ Group Meeting (EuroMPI), pages 6:1–6:8, 2019. [9] Yousef Saad and Martin H. Schultz. Data communication in hypercubes. Journal of Parallel and Distributed Computing, 6(1):115–135, 1989.
14
[10] Young-Joo Suh and Kang G. Shin. All-to-all personalized communication in multidimensional torus and mesh networks. IEEE Transactions on Parallel and Distributed Systems, 12(1):38–59, 2001. [11] Jesper Larsson Träff. Optimal, non-pipelined reduce-scatter and allreduce algorithms with an application to all-to-all communication. ACM Transactions on Parallel Computing, 12(4):1–23, 2025. [12] Jesper Larsson Träff, William D. Gropp, and Rajeev Thakur. Self-consistent MPI performance guidelines. IEEE Transactions on Parallel and Distributed Systems, 21(5):698–709, 2010. [13] Jesper Larsson Träff and Sascha Hunold. Decomposing MPI collectives for exploiting multilane communication. In IEEE International Conference on Cluster Computing (CLUSTER), pages 270–280. IEEE Computer Society, 2020. [14] Jesper Larsson Träff, Sascha Hunold, Guillaume Mercier, and Daniel J. Holmes. MPI collective communication through a single set of interfaces: A case for orthogonality. Parallel Computing, 107, 2021. [15] Jesper Larsson Träff and Felix Donatus Lübbe. Specification guideline violations by MPI Dims create. In 22nd European MPI Users’ Group Meeting (EuroMPI), pages 19:1– 19:2. ACM, 2015. [16] Jesper Larsson Träff and Antoine Rougier. MPI collectives and datatypes for hierarchical all-to-all communication. In 21st European MPI Users’ Group Meeting (EuroMPI/ASIA), pages 27–32. ACM, 2014. [17] Jesper Larsson Träff, Antoine Rougier, and Sascha Hunold. Implementing a classic: Zerocopy all-to-all communication with MPI datatypes. In 28th ACM International Conference on Supercomputing (ICS), pages 135–144. ACM, 2014. [18] Yu-Chee Tseng and Sandeep K. S. Gupta. All-to-all personalized communication in a wormhole-routed torus. IEEE Transactions on Parallel and Distributed Systems, 7(5):498– 505, 1996. [19] Yu-Chee Tseng, Sze-Yao Ni, and Jang-Ping Sheu. Toward optimal complete exchange on wormhole-routed tori. IEEE Transactions on Computers, 48(10):1065–1082, 1999. [20] Syunji Yazaki, Haruyuki Takaue, Yuichiro Ajima, Toshiyuki Shimizu, and Hiroaki Ishihata. An efficient all-to-all communication algorithm for mesh/torus networks. In 10th IEEE International Symposium on Parallel and Distributed Processing with Applications (ISPA), pages 277–284, 2012.
15