D R A F T

Document for a Standard Message-Passing Interface

Message Passing Interface Forum

May 30, 1995

This work was supported in part by ARPA and NSF under grant ASC-9310330, the

National Science Foundation Science and Technology Center Cooperative Agreement No. CCR-8809615, and by the Commission of the European Community

through Esprit project P6643.

This is the result of a LaTeX run of a draft of a single chapter of the MPIF Final Report document.

Chapter 1

Collective Communication

1.1 Introduction and Overview

Collective communication is defined as communication that involves a group of processes. The functions of this type provided by MPI are the following:

• Barrier synchronization across all group members (Sec. 1.3).

• Broadcast from one member to all members of a group (Sec. 1.4). This is shown in figure 1.1.

• Gather data from all group members to one member (Sec. 1.5). This is shown in figure 1.1.

• Scatter data from one member to all members of a group (Sec. 1.6). This is shown in figure 1.1.

• A variation on Gather where all members of the group receive the result (Sec. 1.7). This is shown as “allgather” in figure 1.1.

• Scatter/Gather data from all members to all members of a group (also called complete exchange or all-to-all) (Sec. 1.8). This is shown as “alltoall” in figure 1.1.

• Global reduction operations such as sum, max, min, or user-defined functions, where the result is returned to all group members and a variation where the result is returned to only one member (Sec. 1.9).

• A combined reduction and scatter operation (Sec. 1.10).

• Scan across all members of a group (also called prefix) (Sec. 1.11).

A collective operation is executed by having all processes in the group call the communication routine, with matching arguments. The syntax and semantics of the collective operations are defined to be consistent with the syntax and semantics of the point-to-point operations. Thus, general datatypes are allowed and must match between sending and receiving processes as specified in Chapter ??. One of the key arguments is a communicator that defines the group of participating processes and provides a context for the operation. Several collective routines such as broadcast and gather have a single originating or receiving process. Such processes are called the root. Some arguments in the collective functions are

2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Figure 1.1: Collective move functions illustrated for a group of six processes. In each case, each row of boxes represents data locations in one process. Thus, in the broadcast, initially just the first process contains the data A_{0}, but after the broadcast all processes contain it.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

specified as “significant only at root,” and are ignored for all participants except the root. The reader is referred to Chapter ?? for information concerning communication buffers, general datatypes and type matching rules, and to Chapter ?? for information on how to define groups and create communicators.

The type-matching conditions for the collective operations are more strict than the corresponding conditions between sender and receiver in point-to-point. Namely, for collective operations, the amount of data sent must exactly match the amount of data specified by the receiver. Distinct type maps (the layout in memory, see Sec. ??) between sender and receiver are still allowed.

Collective routine calls can (but are not required to) return as soon as their participation in the collective communication is complete. The completion of a call indicates that the caller is now free to access locations in the communication buffer. It does not indicate that other processes in the group have completed or even started the operation (unless otherwise indicated in the description of the operation). Thus, a collective communication call may, or may not, have the effect of synchronizing all calling processes. This statement excludes, of course, the barrier function.

Collective communication calls may use the same communicators as point-to-point communication; MPI guarantees that messages generated on behalf of collective communication calls will not be confused with messages generated by point-to-point communication. A more detailed discussion of correct use of collective routines is found in Sec. 1.12.

Rationale. The equal-data restriction (on type matching) was made so as to avoid the complexity of providing a facility analogous to the status argument of MPI RECV for discovering the amount of data sent. Some of the collective routines would require an array of status values. The statements about synchronization are made so as to allow a variety of implementations of the collective functions. The collective operations do not accept a message tag argument. If future revisions of MPI define non-blocking collective functions, then tags (or a similar mechanism) will need to be added so as to allow the dis-ambiguation of multiple, pending, collective operations. (End of rationale.)

Advice to users. It is dangerous to rely on synchronization side-effects of the collective operations for program correctness. For example, even though a particular implementation may provide a broadcast routine with a side-effect of synchronization, the standard does not require this, and a program that relies on this will not be portable. On the other hand, a correct, portable program must allow for the fact that a collective call may be synchronizing. Though one cannot rely on any synchronization side-effect, one must program so as to allow it. These issues are discussed further in Sec. 1.12. (End of advice to users.)

Advice to implementors. While vendors may write optimized collective routines matched to their architectures, a complete library of the collective communication routines can be written entirely using the MPI point-to-point communication functions and a few auxiliary functions. If implementing on top of point-to-point, a hidden, special communicator must be created for the collective operation so as to avoid interference with any on-going point-to-point communication at the time of the collective call. This is discussed further in Sec. 1.12. (End of advice to implementors.)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

1.2 Communicator argument

The key concept of the collective functions is to have a “group” of participating processes. The routines do not have a group identifier as an explicit argument. Instead, there is a communicator argument. For the purposes of this chapter, a communicator can be thought of as a group identifier linked with a context. An inter-communicator, that is, a communicator that spans two groups, is not allowed as an argument to a collective function.

1.3 Barrier synchronization

MPI BARRIER( comm )

IN  comm  communicator (handle)

int MPI Barrier(MPI Comm comm )
MPI BARRIER(COMM, IERROR)
    INTEGER COMM, IERROR

MPI BARRIER blocks the caller until all group members have called it. The call returns at any process only after all group members have entered the call.

1.4 Broadcast

MPI BCAST( buffer, count, datatype, root, comm )

INOUT buffer starting address of buffer (choice)

IN  count  number of entries in buffer (integer)
IN  datatype  data type of buffer (handle)
IN  root  rank of broadcast root (integer)
IN  comm  communicator (handle)

int MPI Bcast(void* buffer, int count, MPI Datatype datatype, int root,
             MPI Comm comm )
MPI BCAST(BUFFER, COUNT, DATATYPE, ROOT, COMM, IERROR)
    <type> BUFFER(*)
    INTEGER COUNT, DATATYPE, ROOT, COMM, IERROR

MPI BCAST broadcasts a message from the process with rank root to all processes of the group, itself included. It is called by all members of group using the same arguments for comm, root. On return, the contents of root’s communication buffer has been copied to all processes.

General, derived datatypes are allowed for datatype. The type signature of count, datatype on any process must be equal to the type signature of count, datatype at the root.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

This implies that the amount of data sent must be equal to the amount received, pairwise between each process and the root. MPI BCAST and all other data-movement collective routines make this restriction. Distinct type maps between sender and receiver are still allowed.

1.4.1 Example using MPI BCAST

Example 1.1 Broadcast 100 ints from process 0 to every process in the group.

    MPI_Comm comm;
    int array[100];
    int root=0;
    ...
    MPI_Bcast( array, 100, MPI_INT, root, comm);

As in many of our example code fragments, we assume that some of the variables (such as comm in the above) have been assigned appropriate values.

1.5 Gather

MPI GATHER( sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype, root, comm)

IN  sendbuf  starting address of send buffer (choice)
IN  sendcount  number of elements in send buffer (integer)
IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice, significant only at

root)

IN  recvcount  number of elements for any single receive (integer, sig-

nificant only at root)

IN  recvtype  data type of recv buffer elements (significant only at

root) (handle)

IN  root  rank of receiving process (integer)
IN  comm  communicator (handle)

int MPI Gather(void* sendbuf, int sendcount, MPI Datatype sendtype,
             void* recvbuf, int recvcount, MPI Datatype recvtype, int root,
             MPI Comm comm)
MPI GATHER(SENDBUF, SENDCOUNT, SENDTYPE, RECVBUF, RECVCOUNT, RECVTYPE,
             ROOT, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNT, SENDTYPE, RECVCOUNT, RECVTYPE, ROOT, COMM, IERROR

Each process (root process included) sends the contents of its send buffer to the root process. The root process receives the messages and stores them in rank order. The outcome

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

is as if each of the n processes in the group (including the root process) had executed a call to

     MPI Send(sendbuf, sendcount, sendtype, root, ...),

and the root had executed n calls to

     MPI Recv(recvbuf + i · recvcount · extent(recvtype), recvcount, recvtype, i, ...),

where extent(recvtype) is the type extent obtained from a call to MPI Type extent().

An alternative description is that the n messages sent by the processes in the group are concatenated in rank order, and the resulting message is received by the root as if by a call to MPI RECV(recvbuf, recvcount·n, recvtype, ...).

The receive buffer is ignored for all non-root processes.

General, derived datatypes are allowed for both sendtype and recvtype. The type signature of sendcount, sendtype on process i must be equal to the type signature of recvcount, recvtype at the root. This implies that the amount of data sent must be equal to the amount of data received, pairwise between each process and the root. Distinct type maps between sender and receiver are still allowed.

All arguments to the function are significant on process root, while on other processes, only arguments sendbuf, sendcount, sendtype, root, comm are significant. The arguments root and comm must have identical values on all processes.

The specification of counts and types should not cause any location on the root to be written more than once. Such a call is erroneous.

Note that the recvcount argument at the root indicates the number of items it receives from each process, not the total number of items it receives.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

MPI GATHERV( sendbuf, sendcount, sendtype, recvbuf, recvcounts, displs, recvtype, root, comm)

IN  sendbuf  starting address of send buffer (choice)
IN  sendcount  number of elements in send buffer (integer)
IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice, significant only at

root)

IN  recvcounts  integer array (of length group size) containing the num-

ber of elements that are received from each process (significant only at root)

IN  displs  integer array (of length group size). Entry i specifies

the displacement relative to recvbuf at which to place the incoming data from process i (significant only at root)

IN  recvtype  data type of recv buffer elements (significant only at

root) (handle)

IN  root  rank of receiving process (integer)
IN  comm  communicator (handle)

int MPI Gatherv(void* sendbuf, int sendcount, MPI Datatype sendtype,
             void* recvbuf, int *recvcounts, int *displs,
             MPI Datatype recvtype, int root, MPI Comm comm)
MPI GATHERV(SENDBUF, SENDCOUNT, SENDTYPE, RECVBUF, RECVCOUNTS, DISPLS,
             RECVTYPE, ROOT, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNT, SENDTYPE, RECVCOUNTS(*), DISPLS(*), RECVTYPE, ROOT,
    COMM, IERROR

MPI GATHERV extends the functionality of MPI GATHER by allowing a varying count of data from each process, since recvcounts is now an array. It also allows more flexibility as to where the data is placed on the root, by providing the new argument, displs.

The outcome is as if each process, including the root process, sends a message to the root,

     MPI Send(sendbuf, sendcount, sendtype, root, ...),

and the root executes n receives,

     MPI Recv(recvbuf + disp[i] · extent(recvtype), recvcounts[i], recvtype, i, ...).

Messages are placed in the receive buffer of the root process in rank order, that is, the data sent from process j is placed in the jth portion of the receive buffer recvbuf on process root. The jth portion of recvbuf begins at offset displs[j] elements (in terms of recvtype) into recvbuf.

The receive buffer is ignored for all non-root processes.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Figure 1.2: The root process gathers 100 ints from each process in the group.

The type signature implied by sendcount, sendtype on process i must be equal to the type signature implied by recvcounts[i], recvtype at the root. This implies that the amount of data sent must be equal to the amount of data received, pairwise between each process and the root. Distinct type maps between sender and receiver are still allowed, as illustrated in Example 1.6.

All arguments to the function are significant on process root, while on other processes, only arguments sendbuf, sendcount, sendtype, root, comm are significant. The arguments root and comm must have identical values on all processes.

The specification of counts, types, and displacements should not cause any location on the root to be written more than once. Such a call is erroneous.

1.5.1 Examples using MPI GATHER, MPI GATHERV

Example 1.2 Gather 100 ints from every process in group to root. See figure 1.2.

MPI_Comm comm;
int gsize,sendarray[100];
int root, *rbuf;
...
MPI_Comm_size( comm, &gsize);
rbuf = (int *)malloc(gsize*100*sizeof(int));
MPI_Gather( sendarray, 100, MPI_INT, rbuf, 100, MPI_INT, root, comm);

Example 1.3 Previous example modified – only the root allocates memory for the receive buffer.

MPI_Comm comm;
int gsize,sendarray[100];
int root, myrank, *rbuf;
...
MPI_Comm_rank( comm, myrank);
if ( myrank == root) {
   MPI_Comm_size( comm, &gsize);
   rbuf = (int *)malloc(gsize*100*sizeof(int));
   }
MPI_Gather( sendarray, 100, MPI_INT, rbuf, 100, MPI_INT, root, comm);

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Example 1.4 Do the same as the previous example, but use a derived datatype. Note that the type cannot be the entire set of gsize*100 ints since type matching is defined pairwise between the root and each process in the gather.

MPI_Comm comm;
int gsize,sendarray[100];
int root, *rbuf;
MPI_Datatype rtype;
...
MPI_Comm_size( comm, &gsize);
MPI_Type_contiguous( 100, MPI_INT, &rtype );
MPI_Type_commit( &rtype );
rbuf = (int *)malloc(gsize*100*sizeof(int));
MPI_Gather( sendarray, 100, MPI_INT, rbuf, 1, rtype, root, comm);

Example 1.5 Now have each process send 100 ints to root, but place each set (of 100) stride ints apart at receiving end. Use MPI GATHERV and the displs argument to achieve this effect. Assume stride ≥ 100. See figure 1.3.

MPI_Comm comm;
int gsize,sendarray[100];
int root, *rbuf, stride;
int *displs,i,*rcounts;
...
MPI_Comm_size( comm, &gsize);
rbuf = (int *)malloc(gsize*stride*sizeof(int));
displs = (int *)malloc(gsize*sizeof(int));
rcounts = (int *)malloc(gsize*sizeof(int));
for (i=0; i<gsize; ++i) {
    displs[i] = i*stride;
    rcounts[i] = 100;
}
MPI_Gatherv( sendarray, 100, MPI_INT, rbuf, rcounts, displs, MPI_INT,
                                                           root, comm);

Note that the program is erroneous if stride < 100.

Example 1.6 Same as Example 1.5 on the receiving side, but send the 100 ints from the 0th column of a 100×150 int array, in C. See figure 1.4.

MPI_Comm comm;
int gsize,sendarray[100][150];
int root, *rbuf, stride;
MPI_Datatype stype;
int *displs,i,*rcounts;

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Figure 1.3: The root process gathers 100 ints from each process in the group, each set is placed stride ints apart.

Figure 1.4: The root process gathers column 0 of a 100×150 C array, and each set is placed stride ints apart.

...
MPI_Comm_size( comm, &gsize);
rbuf = (int *)malloc(gsize*stride*sizeof(int));
displs = (int *)malloc(gsize*sizeof(int));
rcounts = (int *)malloc(gsize*sizeof(int));
for (i=0; i<gsize; ++i) {
    displs[i] = i*stride;
    rcounts[i] = 100;
}
/* Create datatype for 1 column of array
 */
MPI_Type_vector( 100, 1, 150, MPI_INT, &stype);
MPI_Type_commit( &stype );
MPI_Gatherv( sendarray, 1, stype, rbuf, rcounts, displs, MPI_INT,
                                                         root, comm);

Example 1.7 Process i sends (100-i) ints from the ith column of a 100 × 150 int array, in C. It is received into a buffer with stride, as in the previous two examples. See figure 1.5.

MPI_Comm comm;
int gsize,sendarray[100][150],*sptr;
int root, *rbuf, stride, myrank;
MPI_Datatype stype;

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

Figure 1.5: The root process gathers 100-i ints from column i of a 100×150 C array, and each set is placed stride ints apart.

int *displs,i,*rcounts;
...
MPI_Comm_size( comm, &gsize);
MPI_Comm_rank( comm, &myrank );
rbuf = (int *)malloc(gsize*stride*sizeof(int));
displs = (int *)malloc(gsize*sizeof(int));
rcounts = (int *)malloc(gsize*sizeof(int));
for (i=0; i<gsize; ++i) {
    displs[i] = i*stride;
    rcounts[i] = 100-i;     /* note change from previous example */
}
/* Create datatype for the column we are sending
 */
MPI_Type_vector( 100-myrank, 1, 150, MPI_INT, &stype);
MPI_Type_commit( &stype );
/* sptr is the address of start of "myrank" column
 */
sptr = &sendarray[0][myrank];
MPI_Gatherv( sptr, 1, stype, rbuf, rcounts, displs, MPI_INT,
root, comm);

Note that a different amount of data is received from each process.

Example 1.8 Same as Example 1.7, but done in a different way at the sending end. We create a datatype that causes the correct striding at the sending end so that that we read a column of a C array. A similar thing was done in Example ??, Section ??.

MPI_Comm comm;
int gsize,sendarray[100][150],*sptr;
int root, *rbuf, stride, myrank, disp[2], blocklen[2];
MPI_Datatype stype,type[2];
int *displs,i,*rcounts;

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

...
MPI_Comm_size( comm, &gsize);
MPI_Comm_rank( comm, &myrank );
rbuf = (int *)malloc(gsize*stride*sizeof(int));
displs = (int *)malloc(gsize*sizeof(int));
rcounts = (int *)malloc(gsize*sizeof(int));
for (i=0; i<gsize; ++i) {
    displs[i] = i*stride;
    rcounts[i] = 100-i;
}
/* Create datatype for one int, with extent of entire row
 */
disp[0] = 0;       disp[1] = 150*sizeof(int);
type[0] = MPI_INT; type[1] = MPI_UB;
blocklen[0] = 1;   blocklen[1] = 1;
MPI_Type_struct( 2, blocklen, disp, type, &stype );
MPI_Type_commit( &stype );
sptr = &sendarray[0][myrank];
MPI_Gatherv( sptr, 100-myrank, stype, rbuf, rcounts, displs, MPI_INT,
                                                           root, comm);

Example 1.9 Same as Example 1.7 at sending side, but at receiving side we make the stride between received blocks vary from block to block. See figure 1.6.

MPI_Comm comm;
int gsize,sendarray[100][150],*sptr;
int root, *rbuf, *stride, myrank, bufsize;
MPI_Datatype stype;
int *displs,i,*rcounts,offset;
...
MPI_Comm_size( comm, &gsize);
MPI_Comm_rank( comm, &myrank );
stride = (int *)malloc(gsize*sizeof(int));
...
/* stride[i] for i = 0 to gsize-1 is set somehow
 */
/* set up displs and rcounts vectors first
 */
displs = (int *)malloc(gsize*sizeof(int));
rcounts = (int *)malloc(gsize*sizeof(int));
offset = 0;
for (i=0; i<gsize; ++i) {
    displs[i] = offset;

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Figure 1.6: The root process gathers 100-i ints from column i of a 100×150 C array, and each set is placed stride[i] ints apart (a varying stride).

    offset += stride[i];
    rcounts[i] = 100-i;
}
/* the required buffer size for rbuf is now easily obtained
 */
bufsize = displs[gsize-1]+rcounts[gsize-1];
rbuf = (int *)malloc(bufsize*sizeof(int));
/* Create datatype for the column we are sending
 */
MPI_Type_vector( 100-myrank, 1, 150, MPI_INT, &stype);
MPI_Type_commit( &stype );
sptr = &sendarray[0][myrank];
MPI_Gatherv( sptr, 1, stype, rbuf, rcounts, displs, MPI_INT,
                                                    root, comm);

Example 1.10 Process i sends num ints from the ith column of a 100 × 150 int array, in C. The complicating factor is that the various values of num are not known to root, so a separate gather must first be run to find these out. The data is placed contiguously at the receiving end.

MPI_Comm comm;
int gsize,sendarray[100][150],*sptr;
int root, *rbuf, stride, myrank, disp[2], blocklen[2];
MPI_Datatype stype,types[2];
int *displs,i,*rcounts,num;
...
MPI_Comm_size( comm, &gsize);
MPI_Comm_rank( comm, &myrank );
/* First, gather nums to root
 */
rcounts = (int *)malloc(gsize*sizeof(int));
MPI_Gather( &num, 1, MPI_INT, rcounts, 1, MPI_INT, root, comm);

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

/* root now has correct rcounts, using these we set displs[] so
 * that data is placed contiguously (or concatenated) at receive end
 */
displs = (int *)malloc(gsize*sizeof(int));
displs[0] = 0;
for (i=1; i<gsize; ++i) {
    displs[i] = displs[i-1]+rcounts[i-1];
}
/* And, create receive buffer
 */
rbuf = (int *)malloc(gsize*(displs[gsize-1]+rcounts[gsize-1])
 *sizeof(int));
/* Create datatype for one int, with extent of entire row
 */
disp[0] = 0;       disp[1] = 150*sizeof(int);
type[0] = MPI_INT; type[1] = MPI_UB;
blocklen[0] = 1;   blocklen[1] = 1;
MPI_Type_struct( 2, blocklen, disp, type, &stype );
MPI_Type_commit( &stype );
sptr = &sendarray[0][myrank];
MPI_Gatherv( sptr, num, stype, rbuf, rcounts, displs, MPI_INT,
                                                           root, comm);

1.6 Scatter

MPI SCATTER( sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype, root, comm)

IN  sendbuf  address of send buffer (choice, significant only at root)
IN  sendcount  number of elements sent to each process (integer, sig-

nificant only at root)

IN  sendtype  data type of send buffer elements (significant only at

root) (handle)

OUT  recvbuf  address of receive buffer (choice)
IN  recvcount  number of elements in receive buffer (integer)
IN  recvtype  data type of receive buffer elements (handle)
IN  root  rank of sending process (integer)
IN  comm  communicator (handle)

int MPI Scatter(void* sendbuf, int sendcount, MPI Datatype sendtype,
         void* recvbuf, int recvcount, MPI Datatype recvtype, int root,
         MPI Comm comm)
MPI SCATTER(SENDBUF, SENDCOUNT, SENDTYPE, RECVBUF, RECVCOUNT, RECVTYPE,
         ROOT, COMM, IERROR)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNT, SENDTYPE, RECVCOUNT, RECVTYPE, ROOT, COMM, IERROR

MPI SCATTER is the inverse operation to MPI GATHER. The outcome is as if the root executed n send operations,

     MPI Send(sendbuf + i · sendcount · extent(sendtype), sendcount, sendtype, i, ...),

and each process executed a receive,

     MPI Recv(recvbuf, recvcount, recvtype, i, ...).

An alternative description is that the root sends a message with MPI Send(sendbuf, sendcount·n, sendtype, ...). This message is split into n equal segments, the ith segment is sent to the ith process in the group, and each process receives this message as above.

The send buffer is ignored for all non-root processes.

The type signature associated with sendcount, sendtype at the root must be equal to the type signature associated with recvcount, recvtype at all processes (however, the type maps may be different). This implies that the amount of data sent must be equal to the amount of data received, pairwise between each process and the root. Distinct type maps between sender and receiver are still allowed.

All arguments to the function are significant on process root, while on other processes, only arguments recvbuf, recvcount, recvtype, root, comm are significant. The arguments root and comm must have identical values on all processes.

The specification of counts and types should not cause any location on the root to be read more than once.

Rationale. Though not needed, the last restriction is imposed so as to achieve symmetry with MPI GATHER, where the corresponding restriction (a multiple-write restriction) is necessary. (End of rationale.)

MPI SCATTERV( sendbuf, sendcounts, displs, sendtype, recvbuf, recvcount, recvtype, root, comm)

IN  sendbuf  address of send buffer (choice, significant only at root)
IN  sendcounts  integer array (of length group size) specifying the num-

ber of elements to send to each processor

IN displs integer array (of length group size). Entry i specifies the displacement (relative to sendbuf from which to take the outgoing data to process i

IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice)
IN  recvcount  number of elements in receive buffer (integer)
IN  recvtype  data type of receive buffer elements (handle)
IN  root  rank of sending process (integer)
IN  comm  communicator (handle)

int MPI Scatterv(void* sendbuf, int *sendcounts, int *displs,
             MPI Datatype sendtype, void* recvbuf, int recvcount,

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

             MPI Datatype recvtype, int root, MPI Comm comm)
MPI SCATTERV(SENDBUF, SENDCOUNTS, DISPLS, SENDTYPE, RECVBUF, RECVCOUNT,
             RECVTYPE, ROOT, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNTS(*), DISPLS(*), SENDTYPE, RECVCOUNT, RECVTYPE, ROOT,
    COMM, IERROR

MPI SCATTERV is the inverse operation to MPI GATHERV.

MPI SCATTERV extends the functionality of MPI SCATTER by allowing a varying count of data to be sent to each process, since sendcounts is now an array. It also allows more flexibility as to where the data is taken from on the root, by providing the new argument, displs.

The outcome is as if the root executed n send operations,

     MPI Send(sendbuf + displs[i] · extent(sendtype), sendcounts[i], sendtype, i, ...),

and each process executed a receive,

     MPI Recv(recvbuf, recvcount, recvtype, i, ...).

The send buffer is ignored for all non-root processes.

The type signature implied by sendcount[i], sendtype at the root must be equal to the type signature implied by recvcount, recvtype at process i (however, the type maps may be different). This implies that the amount of data sent must be equal to the amount of data received, pairwise between each process and the root. Distinct type maps between sender and receiver are still allowed.

All arguments to the function are significant on process root, while on other processes, only arguments recvbuf, recvcount, recvtype, root, comm are significant. The arguments root and comm must have identical values on all processes.

The specification of counts, types, and displacements should not cause any location on the root to be read more than once.

1.6.1 Examples using MPI SCATTER, MPI SCATTERV

Example 1.11 The reverse of Example 1.2. Scatter sets of 100 ints from the root to each process in the group. See figure 1.7.

    MPI_Comm comm;
    int gsize,*sendbuf;
    int root, rbuf[100];
    ...
    MPI_Comm_size( comm, &gsize);
    sendbuf = (int *)malloc(gsize*100*sizeof(int));
    ...
    MPI_Scatter( sendbuf, 100, MPI_INT, rbuf, 100, MPI_INT, root, comm);

Example 1.12 The reverse of Example 1.5. The root process scatters sets of 100 ints to the other processes, but the sets of 100 are stride ints apart in the sending buffer. Requires use of MPI SCATTERV. Assume stride ≥ 100. See figure 1.8.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

Figure 1.7: The root process scatters sets of 100 ints to each process in the group.

Figure 1.8: The root process scatters sets of 100 ints, moving by stride ints from send to send in the scatter.

MPI_Comm comm;
int gsize,*sendbuf;
int root, rbuf[100], i, *displs, *scounts;
...
MPI_Comm_size( comm, &gsize);
sendbuf = (int *)malloc(gsize*stride*sizeof(int));
...
displs = (int *)malloc(gsize*sizeof(int));
scounts = (int *)malloc(gsize*sizeof(int));
for (i=0; i<gsize; ++i) {
    displs[i] = i*stride;
    scounts[i] = 100;
}
MPI_Scatterv( sendbuf, scounts, displs, MPI_INT, rbuf, 100, MPI_INT,
                                                          root, comm);

Example 1.13 The reverse of Example 1.9. We have a varying stride between blocks at sending (root) side, at the receiving side we receive into the ith column of a 100×150 C array. See figure 1.9.

MPI_Comm comm;
int gsize,recvarray[100][150],*rptr;
int root, *sendbuf, myrank, bufsize, *stride;

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Figure 1.9: The root scatters blocks of 100-i ints into column i of a 100×150 C array. At the sending side, the blocks are stride[i] ints apart.

MPI_Datatype rtype;
int i, *displs, *scounts, offset;
...
MPI_Comm_size( comm, &gsize);
MPI_Comm_rank( comm, &myrank );
stride = (int *)malloc(gsize*sizeof(int));
...
/* stride[i] for i = 0 to gsize-1 is set somehow
 * sendbuf comes from elsewhere
 */
...
displs = (int *)malloc(gsize*sizeof(int));
scounts = (int *)malloc(gsize*sizeof(int));
offset = 0;
for (i=0; i<gsize; ++i) {
    displs[i] = offset;
    offset += stride[i];
    scounts[i] = 100 - i;
}
/* Create datatype for the column we are receiving
 */
MPI_Type_vector( 100-myrank, 1, 150, MPI_INT, &rtype);
MPI_Type_commit( &rtype );
rptr = &recvarray[0][myrank];
MPI_Scatterv( sendbuf, scounts, displs, MPI_INT, rptr, 1, rtype,
                                                        root, comm);

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

1.7 Gather-to-all

MPI ALLGATHER( sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype, comm)

IN  sendbuf  starting address of send buffer (choice)
IN  sendcount  number of elements in send buffer (integer)
IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice)
IN  recvcount  number of elements received from any process (inte-

ger)

IN  recvtype  data type of receive buffer elements (handle)
IN  comm  communicator (handle)

int MPI Allgather(void* sendbuf, int sendcount, MPI Datatype sendtype,
             void* recvbuf, int recvcount, MPI Datatype recvtype,
             MPI Comm comm)
MPI ALLGATHER(SENDBUF, SENDCOUNT, SENDTYPE, RECVBUF, RECVCOUNT, RECVTYPE,
             COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNT, SENDTYPE, RECVCOUNT, RECVTYPE, COMM, IERROR

MPI ALLGATHER can be thought of as MPI GATHER, but where all processes receive the result, instead of just the root. The jth block of data sent from each process is received by every process and placed in the jth block of the buffer recvbuf.

The type signature associated with sendcount, sendtype, at a process must be equal to the type signature associated with recvcount, recvtype at any other process.

The outcome of a call to MPI ALLGATHER(...) is as if all processes executed n calls to

   MPI_GATHER(sendbuf,sendcount,sendtype,recvbuf,recvcount,
                                                 recvtype,root,comm),

for root = 0 , ..., n-1. The rules for correct usage of MPI ALLGATHER are easily found from the corresponding rules for MPI GATHER.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

MPI ALLGATHERV( sendbuf, sendcount, sendtype, recvbuf, recvcounts, displs, recvtype, comm)

IN  sendbuf  starting address of send buffer (choice)
IN  sendcount  number of elements in send buffer (integer)
IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice)
IN  recvcounts  integer array (of length group size) containing the num-

ber of elements that are received from each process

IN  displs  integer array (of length group size). Entry i specifies

the displacement (relative to recvbuf) at which to place the incoming data from process i

IN  recvtype  data type of receive buffer elements (handle)
IN  comm  communicator (handle)

int MPI Allgatherv(void* sendbuf, int sendcount, MPI Datatype sendtype,
             void* recvbuf, int *recvcounts, int *displs,
             MPI Datatype recvtype, MPI Comm comm)
MPI ALLGATHERV(SENDBUF, SENDCOUNT, SENDTYPE, RECVBUF, RECVCOUNTS, DISPLS,
             RECVTYPE, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNT, SENDTYPE, RECVCOUNTS(*), DISPLS(*), RECVTYPE, COMM,
    IERROR

MPI ALLGATHERV can be thought of as MPI GATHERV, but where all processes receive the result, instead of just the root. The jth block of data sent from each process is received by every process and placed in the jth block of the buffer recvbuf. These blocks need not all be the same size.

The type signature associated with sendcount, sendtype, at process j must be equal to the type signature associated with recvcounts[j], recvtype at any other process.

The outcome is as if all processes executed calls to

    MPI_GATHERV(sendbuf,sendcount,sendtype,recvbuf,recvcounts,displs,
                                                   recvtype,root,comm),

for root = 0 , ..., n-1. The rules for correct usage of MPI ALLGATHERV are easily found from the corresponding rules for MPI GATHERV.

1.7.1 Examples using MPI ALLGATHER, MPI ALLGATHERV

Example 1.14 The all-gather version of Example 1.2. Using MPI ALLGATHER, we will gather 100 ints from every process in the group to every process.

    MPI_Comm comm;
    int gsize,sendarray[100];
    int *rbuf;
    ...

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

    MPI_Comm_size( comm, &gsize);
    rbuf = (int *)malloc(gsize*100*sizeof(int));
    MPI_Allgather( sendarray, 100, MPI_INT, rbuf, 100, MPI_INT, comm);

After the call, every process has the group-wide concatenation of the sets of data.

1.8 All-to-All Scatter/Gather

MPI ALLTOALL(sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype, comm)

IN  sendbuf  starting address of send buffer (choice)
IN  sendcount  number of elements sent to each process (integer)
IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice)
IN  recvcount  number of elements received from any process (inte-

ger)

IN  recvtype  data type of receive buffer elements (handle)
IN  comm  communicator (handle)

int MPI Alltoall(void* sendbuf, int sendcount, MPI Datatype sendtype,
             void* recvbuf, int recvcount, MPI Datatype recvtype,
             MPI Comm comm)
MPI ALLTOALL(SENDBUF, SENDCOUNT, SENDTYPE, RECVBUF, RECVCOUNT, RECVTYPE,
             COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNT, SENDTYPE, RECVCOUNT, RECVTYPE, COMM, IERROR

MPI ALLTOALL is an extension of MPI ALLGATHER to the case where each process sends distinct data to each of the receivers. The jth block sent from process i is received by process j and is placed in the ith block of recvbuf.

The type signature associated with sendcount, sendtype, at a process must be equal to the type signature associated with recvcount, recvtype at any other process. This implies that the amount of data sent must be equal to the amount of data received, pairwise between every pair of processes. As usual, however, the type maps may be different.

The outcome is as if each process executed a send to each process (itself included) with a call to,

     MPI Send(sendbuf + i · sendcount · extent(sendtype), sendcount, sendtype, i, ...),

and a receive from every other process with a call to,

     MPI Recv(recvbuf + i · recvcount · extent(recvtype), recvcount, i, ...).

All arguments on all processes are significant. The argument comm must have identical values on all processes.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

MPI ALLTOALLV(sendbuf, sendcounts, sdispls, sendtype, recvbuf, recvcounts, rdispls, recvtype, comm)

IN  sendbuf  starting address of send buffer (choice)
IN  sendcounts  integer array equal to the group size specifying the

number of elements to send to each processor

IN  sdispls  integer array (of length group size). Entry j specifies

the displacement (relative to sendbuf from which to take the outgoing data destined for process j

IN  sendtype  data type of send buffer elements (handle)
OUT  recvbuf  address of receive buffer (choice)
IN  recvcounts  integer array equal to the group size specifying the

number of elements that can be received from each processor

IN  rdispls  integer array (of length group size). Entry i specifies

the displacement (relative to recvbuf at which to place the incoming data from process i

IN  recvtype  data type of receive buffer elements (handle)
IN  comm  communicator (handle)

int MPI Alltoallv(void* sendbuf, int *sendcounts, int *sdispls,
             MPI Datatype sendtype, void* recvbuf, int *recvcounts,
             int *rdispls, MPI Datatype recvtype, MPI Comm comm)
MPI ALLTOALLV(SENDBUF, SENDCOUNTS, SDISPLS, SENDTYPE, RECVBUF, RECVCOUNTS,
             RDISPLS, RECVTYPE, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER SENDCOUNTS(*), SDISPLS(*), SENDTYPE, RECVCOUNTS(*), RDISPLS(*),
    RECVTYPE, COMM, IERROR

MPI ALLTOALLV adds flexibility to MPI ALLTOALL in that the location of data for the send is specified by sdispls and the location of the placement of the data on the receive side is specified by rdispls.

The jth block sent from process i is received by process j and is placed in the ith block of recvbuf. These blocks need not all have the same size.

The type signature associated with sendcount[j], sendtype at process i must be equal to the type signature associated with recvcount[i], recvtype at process j. This implies that the amount of data sent must be equal to the amount of data received, pairwise between every pair of processes. Distinct type maps between sender and receiver are still allowed.

The outcome is as if each process sent a message to every other process with,

     MPI Send(sendbuf + displs[i] · extent(sendtype), sendcounts[i], sendtype, i, ...),

and received a message from every other process with a call to

     MPI Recv(recvbuf + displs[i] · extent(recvtype), recvcounts[i], recvtype, i, ...).

All arguments on all processes are significant. The argument comm must have identical values on all processes.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

Rationale. The definitions of MPI ALLTOALL and MPI ALLTOALLV give as much flexibility as one would achieve by specifying n independent, point-to-point communications, with two exceptions: all messages use the same datatype, and messages are scattered from (or gathered to) sequential storage. (End of rationale.)

Advice to implementors. Although the discussion of collective communication in terms of point-to-point operation implies that each message is transferred directly from sender to receiver, implementations may use a tree communication pattern. Messages can be forwarded by intermediate nodes where they are split (for scatter) or concatenated (for gather), if this is more efficient. (End of advice to implementors.)

1.9 Global Reduction Operations

The functions in this section perform a global reduce operation (such as sum, max, logical AND, etc.) across all the members of a group. The reduction operation can be either one of a predefined list of operations, or a user-defined operation. The global reduction functions come in several flavors: a reduce that returns the result of the reduction at one node, an all-reduce that returns this result at all nodes, and a scan (parallel prefix) operation. In addition, a reduce-scatter operation combines the functionality of a reduce and of a scatter operation.

1.9.1 Reduce

MPI REDUCE( sendbuf, recvbuf, count, datatype, op, root, comm)

IN  sendbuf  address of send buffer (choice)
OUT  recvbuf  address of receive buffer (choice, significant only at

root)

IN  count  number of elements in send buffer (integer)
IN  datatype  data type of elements of send buffer (handle)
IN  op  reduce operation (handle)
IN  root  rank of root process (integer)
IN  comm  communicator (handle)

int MPI Reduce(void* sendbuf, void* recvbuf, int count,
             MPI Datatype datatype, MPI Op op, int root, MPI Comm comm)
MPI REDUCE(SENDBUF, RECVBUF, COUNT, DATATYPE, OP, ROOT, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER COUNT, DATATYPE, OP, ROOT, COMM, IERROR

MPI REDUCE combines the elements provided in the input buffer of each process in the group, using the operation op, and returns the combined value in the output buffer of the process with rank root. The input buffer is defined by the arguments sendbuf, count and datatype; the output buffer is defined by the arguments recvbuf, count and datatype; both have the same number of elements, with the same type. The routine is called by all

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

group members using the same arguments for count, datatype, op, root and comm. Thus, all processes provide input buffers and output buffers of the same length, with elements of the same type. Each process can provide one element, or a sequence of elements, in which case the combine operation is executed element-wise on each entry of the sequence. For example, if the operation is MPI MAX and the send buffer contains two elements that are floating point numbers (count = 2 and datatype = MPI FLOAT), then recvbuf(1) = global max(sendbuf(1)) and recvbuf(2) = global max(sendbuf(2)).

Sec. 1.9.2, lists the set of predefined operations provided by MPI. That section also enumerates the datatypes each operation can be applied to. In addition, users may define their own operations that can be overloaded to operate on several datatypes, either basic or derived. This is further explained in Sec. 1.9.4.

The operation op is always assumed to be associative. All predefined operations are also assumed to be commutative. Users may define operations that are assumed to be associative, but not commutative. The “canonical” evaluation order of a reduction is determined by the ranks of the processes in the group. However, the implementation can take advantage of associativity, or associativity and commutativity in order to change the order of evaluation. This may change the result of the reduction for operations that are not strictly associative and commutative, such as floating point addition.

Advice to implementors. It is strongly recommended that MPI REDUCE be implemented so that the same result be obtained whenever the function is applied on the same arguments, appearing in the same order. Note that this may prevent optimizations that take advantage of the physical location of processors. (End of advice to implementors.)

The datatype argument of MPI REDUCE must be compatible with op. Predefined operators work only with the MPI types listed in Sec. 1.9.2 and Sec. 1.9.3. User-defined operators may operate on general, derived datatypes. In this case, each argument that the reduce operation is applied to is one element described by such a datatype, which may contain several basic values. This is further explained in Section 1.9.4.

1.9.2 Predefined reduce operations

The following predefined operations are supplied for MPI REDUCE and related functions MPI ALLREDUCE, MPI REDUCE SCATTER, and MPI SCAN. These operations are invoked by placing the following in op.

Name Meaning

MPI MAX maximum MPI MIN minimum MPI SUM sum MPI PROD product MPI LAND logical and MPI BAND bit-wise and MPI LOR logical or MPI BOR bit-wise or MPI LXOR logical xor

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

MPI BXOR bit-wise xor

MPI MAXLOC max value and location

MPI MINLOC min value and location

The two operations MPI MINLOC and MPI MAXLOC are discussed separately in Sec. 1.9.3. For the other predefined operations, we enumerate below the allowed combinations of op and datatype arguments. First, define groups of MPI basic datatypes in the following way.

C integer:  MPI INT,  MPI LONG,  MPI SHORT,

MPI UNSIGNED SHORT, MPI UNSIGNED, MPI UNSIGNED LONG

Fortran integer: MPI INTEGER

Floating point: MPI FLOAT, MPI DOUBLE, MPI REAL,

MPI DOUBLE PRECISION, MPI LONG DOUBLE

Logical: MPI LOGICAL

Complex: MPI COMPLEX

Byte: MPI BYTE

Now, the valid datatypes for each option is specified below.

Op Allowed Types

⊤
MPI MAX, MPI MIN  C integer, Fortran integer, Floating point  ⊥

MPI SUM, MPI PROD C integer, Fortran integer, Floating point, Complex MPI LAND, MPI LOR, MPI LXOR C integer, Logical MPI BAND, MPI BOR, MPI BXOR C integer, Fortran integer, Byte

Example 1.15 A routine that computes the dot product of two vectors that are distributed across a group of processes and returns the answer at node zero.

SUBROUTINE PAR_BLAS1(m, a, b, c, comm)
REAL a(m), b(m)       ! local slice of array
REAL c                ! result (at node zero)
REAL sum
INTEGER m, comm, i, ierr
! local sum
sum = 0.0
DO i = 1, m
   sum = sum + a(i)*b(i)
END DO
! global sum
CALL MPI_REDUCE(sum, c, 1, MPI_REAL, MPI_SUM, 0, comm, ierr)
RETURN

Example 1.16 A routine that computes the product of a vector and an array that are distributed across a group of processes and returns the answer at node zero.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

SUBROUTINE PAR_BLAS2(m, n, a, b, c, comm)
REAL a(m), b(m,n)    ! local slice of array
REAL c(n)            ! result
REAL sum(n)
INTEGER n, comm, i, j, ierr
! local sum
DO j= 1, n
  sum(j) = 0.0
  DO i = 1, m
    sum(j) = sum(j) + a(i)*b(i,j)
  END DO
END DO
! global sum
CALL MPI_REDUCE(sum, c, n, MPI_REAL, MPI_SUM, 0, comm, ierr)
! return result at node zero (and garbage at the other nodes)
RETURN

1.9.3 MINLOC and MAXLOC

The operator MPI MINLOC is used to compute a global minimum and also an index attached to the minimum value. MPI MAXLOC similarly computes a global maximum and index. One application of these is to compute a global minimum (maximum) and the rank of the process containing this value.

The operation that defines MPI MAXLOC is:

( \begin{matrix}u \\ i \end{matrix} ) ∘ ( \begin{matrix}v \\ j \end{matrix} ) = ( \begin{matrix}w \\ k \end{matrix} )

where

w = max(u, v)

and

k = \{ \begin{matrix}i & if u > v \\ min(i, j) & if u = v \\ j & if u < v \end{matrix}

MPI MINLOC is defined similarly:

( \begin{matrix}u \\ i \end{matrix} ) ∘ ( \begin{matrix}v \\ j \end{matrix} ) = ( \begin{matrix}w \\ k \end{matrix} )

where

w = min(u, v)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

and

k = \{ \begin{matrix}i & if u < v \\ min(i, j) & if u = v \\ j & if u > v \end{matrix}

Both operations are associative and commutative. Note that if MPI MAXLOC is applied to reduce a sequence of pairs (u_{0}, 0), (u_{1}, 1), …, (u_{n−1}, n − 1), then the value returned is (u, r), where u = max_{i} u_{i} and r is the index of the first global maximum in the sequence. Thus, if each process supplies a value and its rank within the group, then a reduce operation with op = MPI MAXLOC will return the maximum value and the rank of the first process with that value. Similarly, MPI MINLOC can be used to return a minimum and its index. More generally, MPI MINLOC computes a lexicographic minimum, where elements are ordered according to the first component of each pair, and ties are resolved according to the second component.

The reduce operation is defined to operate on arguments that consist of a pair: value and index. For both Fortran and C, types are provided to describe the pair. The potentially mixed-type nature of such arguments is a problem in Fortran. The problem is circumvented, for Fortran, by having the MPI-provided type consist of a pair of the same type as value, and coercing the index to this type also. In C, the MPI-provided pair type has distinct types and the index is an int.

In order to use MPI MINLOC and MPI MAXLOC in a reduce operation, one must provide a datatype argument that represents a pair (value and index). MPI provides seven such predefined datatypes. The operations MPI MAXLOC and MPI MINLOC can be used with each of the following datatypes. ⊤

Fortran:

Name Description

MPI 2REAL pair of REALs

MPI 2DOUBLE PRECISION pair of DOUBLE PRECISION variables MPI 2INTEGER pair of INTEGERs

⊥

C:

Name Description

MPI FLOAT INT float and int

MPI DOUBLE INT double and int

MPI LONG INT long and int

MPI 2INT pair of int

MPI SHORT INT short and int

MPI LONG DOUBLE INT long double and int

The datatype MPI 2REAL is as if defined by the following (see Section ??).

MPI_TYPE_CONTIGUOUS(2, MPI_REAL, MPI_2REAL)

Similar statements apply for MPI 2INTEGER, MPI 2DOUBLE PRECISION, and MPI 2INT.

The datatype MPI FLOAT INT is as if defined by the following sequence of instructions.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

type[0] = MPI_FLOAT
type[1] = MPI_INT
disp[0] = 0
disp[1] = sizeof(float)
block[0] = 1
block[1] = 1
MPI_TYPE_STRUCT(2, block, disp, type, MPI_FLOAT_INT)

Similar statements apply for MPI LONG INT and MPI DOUBLE INT.

Example 1.17 Each process has an array of 30 doubles, in C. For each of the 30 locations, compute the value and rank of the process containing the largest value.

...
/* each process has an array of 30 double: ain[30]
 */
double ain[30], aout[30];
int  ind[30];
struct {
    double val;
    int   rank;
} in[30], out[30];
int i, myrank, root;
MPI_Comm_rank(MPI_COMM_WORLD, &myrank);
for (i=0; i<30; ++i) {
    in[i].val = ain[i];
    in[i].rank = myrank;
}
MPI_Reduce( in, out, 30, MPI_DOUBLE_INT, MPI_MAXLOC, root, comm );
/* At this point, the answer resides on process root
 */
if (myrank == root) {
    /* read ranks out
     */
    for (i=0; i<30; ++i) {
        aout[i] = out[i].val;
        ind[i] = out[i].rank;
    }
}

Example 1.18 Same example, in Fortran.

...
! each process has an array of 30 double: ain(30)
DOUBLE PRECISION ain(30), aout(30)
INTEGER ind(30);
DOUBLE PRECISION in(2,30), out(2,30)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

    INTEGER i, myrank, root, ierr;
    MPI_COMM_RANK(MPI_COMM_WORLD, myrank);
    DO I=1, 30
        in(1,i) = ain(i)
        in(2,i) = myrank    ! myrank is coerced to a double
    END DO
    MPI_REDUCE( in, out, 30, MPI_2DOUBLE_PRECISION, MPI_MAXLOC, root,
                                                          comm, ierr );
    ! At this point, the answer resides on process root
    IF (myrank .EQ. root) THEN
        ! read ranks out
        DO I= 1, 30
            aout(i) = out(1,i)
            ind(i) = out(2,i)  ! rank is coerced back to an integer
        END DO
    END IF

Example 1.19 Each process has a non-empty array of values. Find the minimum global value, the rank of the process that holds it and its index on this process.

#define  LEN   1000
float val[LEN];        /* local array of values */
int count;             /* local number of values */
int myrank, minrank, minindex;
float minval;
struct {
    float value;
    int   index;
} in, out;
    /* local minloc */
in.value = val[0];
in.index = 0;
for (i=1; i < count; i++)
    if (in.value > val[i]) {
        in.value = val[i];
        in.index = i;
    }
    /* global minloc */
MPI_Comm_rank(MPI_COMM_WORLD, &myrank);
in.index = myrank*LEN + in.index;
MPI_Reduce( in, out, 1, MPI_FLOAT_INT, MPI_MINLOC, root, comm );

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

    /* At this point, the answer resides on process root
     */
if (myrank == root) {
    /* read answer out
     */
    minval = out.value;
    minrank = out.index / LEN;
    minindex = out.index % LEN;
}

Rationale. The definition of MPI MINLOC and MPI MAXLOC given here has the advantage that it does not require any special-case handling of these two operations: they are handled like any other reduce operation. A programmer can provide his or her own definition of MPI MAXLOC and MPI MINLOC, if so desired. The disadvantage is that values and indices have to be first interleaved, and that indices and values have to be coerced to the same type, in Fortran. (End of rationale.)

1.9.4 User-Defined Operations

MPI OP CREATE( function, commute, op)

IN  function  user defined function (function)
IN  commute  true if commutative; false otherwise.
OUT  op  operation (handle)

int MPI Op create(MPI User function *function, int commute, MPI Op *op)
MPI OP CREATE( FUNCTION, COMMUTE, OP, IERROR)
    EXTERNAL FUNCTION
    LOGICAL COMMUTE
    INTEGER OP, IERROR

MPI OP CREATE binds a user-defined global operation to an op handle that can subsequently be used in MPI REDUCE, MPI ALLREDUCE, MPI REDUCE SCATTER, and MPI SCAN. The user-defined operation is assumed to be associative. If commute = true, then the operation should be both commutative and associative. If commute = false,

⊤ then the order of operands is fixed and is defined to be in ascending, process rank order, beginning with process zero. The order of evaluation can be changed, talking advantage of the associativity of the operation. If commute = true then the order of evaluation can be

⊥ changed, taking advantage of commutativity and associativity.

function is the user-defined function, which must have the following four arguments: invec, inoutvec, len and datatype.

The ANSI-C prototype for the function is the following.

typedef void MPI_User_function( void *invec, void *inoutvec, int *len,
                                                MPI_Datatype *datatype);

The Fortran declaration of the user-defined function appears below.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

FUNCTION USER_FUNCTION( INVEC(*), INOUTVEC(*), LEN, TYPE)
<type> INVEC(LEN), INOUTVEC(LEN)
INTEGER LEN, TYPE

The datatype argument is a handle to the data type that was passed into the call to MPI REDUCE. The user reduce function should be written such that the following holds: Let u[0], ... , u[len-1] be the len elements in the communication buffer described by the arguments invec, len and datatype when the function is invoked; let v[0], ... , v[len-1] be len elements in the communication buffer described by the arguments inoutvec, len and datatype when the function is invoked; let w[0], ... , w[len-1] be len elements in the communication buffer described by the arguments inoutvec, len and datatype when the function returns; then w[i] = u[i]∘v[i], for i=0 , ... , len-1, where ∘ is the reduce operation that the function computes.

Informally, we can think of invec and inoutvec as arrays of len elements that function is combining. The result of the reduction over-writes values in inoutvec, hence the name. Each invocation of the function results in the pointwise evaluation of the reduce operator on len elements: I.e, the function returns in inoutvec[i] the value invec[i] ∘ inoutvec[i], for i = 0, …, count − 1, where ∘ is the combining operation computed by the function.

Rationale. The len argument allows MPI REDUCE to avoid calling the function for each element in the input buffer. Rather, the system can choose to apply the function to chunks of input. In C, it is passed in as a reference for reasons of compatibility with Fortran.

By internally comparing the value of the datatype argument to known, global handles, it is possible to overload the use of a single user-defined function for several, different data types. (End of rationale.)

General datatypes may be passed to the user function. However, use of datatypes that are not contiguous is likely to lead to inefficiencies.

No MPI communication function may be called inside the user function. MPI ABORT may be called inside the function in case of an error.

Advice to users. Suppose one defines a library of user-defined reduce functions that are overloaded: the datatype argument is used to select the right execution path at each invocation, according to the types of the operands. The user-defined reduce function cannot “decode” the datatype argument that it is passed, and cannot identify, by itself, the correspondence between the datatype handles and the datatype they represent. This correspondence was established when the datatypes were created. Before the library is used, a library initialization preamble must be executed. This preamble code will define the datatypes that are used by the library, and store handles to these datatypes in global, static variables that are shared by the user code and the library code.

The Fortran version of MPI REDUCE will invoke a user-defined reduce function using the Fortran calling conventions and will pass a Fortran-type datatype argument; the C version will use C calling convention and the C representation of a datatype handle. Users who plan to mix languages should define their reduction functions accordingly. (End of advice to users.)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Advice to implementors. We outline below a naive and inefficient implementation of MPI REDUCE.

if (rank > 0) {
    RECV(tempbuf, count, datatype, rank-1,...)
    User_reduce( tempbuf, sendbuf, count, datatype)
}
if (rank < groupsize-1) {
    SEND( sendbuf, count, datatype, rank+1, ...)
}
/* answer now resides in process groupsize-1 ... now send to root
 */
if (rank == groupsize-1) {
    SEND( sendbuf, count, datatype, root, ...)
}
if (rank == root) {
    RECV(recvbuf, count, datatype, groupsize-1,...)
}

The reduction computation proceeds, sequentially, from process 0 to process groupsize-1. This order is chosen so as to respect the order of a possibly non-commutative operator defined by the function User reduce(). A more efficient implementation is achieved by taking advantage of associativity and using a logarithmic tree reduction. Commutativity can be used to advantage, for those cases in which the commute argument to MPI OP CREATE is true. Also, the amount of temporary buffer required can be reduced, and communication can be pipelined with computation, by transferring and reducing the elements in chunks of size len <count.

The predefined reduce operations can be implemented as a library of user-defined operations. However, better performance might be achieved if MPI REDUCE handles these functions as a special case. (End of advice to implementors.)

MPI OP FREE( op)

IN  op  operation (handle)

int MPI op free( MPI Op *op)
MPI OP FREE( OP, IERROR)
INTEGER OP, IERROR

Marks a user-defined reduction operation for deallocation and sets op to MPI OP NULL.

Example of User-defined Reduce

It is time for an example of user-defined reduction.

Example 1.20 Compute the product of an array of complex numbers, in C.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

typedef struct {
double real,imag;
} Complex;
/* the user-defined function
*/
void myProd( Complex *in, Complex *inout, int *len, MPI_Datatype *dptr )
{
int i;
Complex c;
for (i=0; i< *len; ++i) {
    c.real = inout->real*in->real -
               inout->imag*in->imag;
    c.imag = inout->real*in->imag +
               inout->imag*in->real;
    *inout = c;
    in++; inout++;
}
}
/* and, to call it...
*/
...
/* each process has an array of 100 Complexes
 */
Complex a[100], answer[100];
MPI_Op myOp;
MPI_Datatype ctype;
/* explain to MPI how type Complex is defined
 */
MPI_Type_contiguous( 2, MPI_DOUBLE, &ctype );
MPI_Type_commit( &ctype );
/* create the complex-product user-op
 */
MPI_Op_create( myProd, True, &myOp );
MPI_Reduce( a, answer, 100, ctype, myOp, root, comm );
/* At this point, the answer, which consists of 100 Complexes,
 * resides on process root
 */

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

1.9.5 All-Reduce

MPI includes variants of each of the reduce operations where the result is returned to all processes in the group. MPI requires that all processes participating in these operations receive identical results.

MPI ALLREDUCE( sendbuf, recvbuf, count, datatype, op, comm)

IN  sendbuf  starting address of send buffer (choice)
OUT  recvbuf  starting address of receive buffer (choice)
IN  count  number of elements in send buffer (integer)
IN  datatype  data type of elements of send buffer (handle)
IN  op  operation (handle)
IN  comm  communicator (handle)

int MPI Allreduce(void* sendbuf, void* recvbuf, int count,
             MPI Datatype datatype, MPI Op op, MPI Comm comm)
MPI ALLREDUCE(SENDBUF, RECVBUF, COUNT, DATATYPE, OP, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER COUNT, DATATYPE, OP, COMM, IERROR

Same as MPI REDUCE except that the result appears in the receive buffer of all the group members.

Advice to implementors. The all-reduce operations can be implemented as a reduce, followed by a broadcast. However, a direct implementation can lead to better performance. (End of advice to implementors.)

Example 1.21 A routine that computes the product of a vector and an array that are distributed across a group of processes and returns the answer at all nodes (see also Example 1.16).

SUBROUTINE PAR_BLAS2(m, n, a, b, c, comm)
REAL a(m), b(m,n)    ! local slice of array
REAL c(n)            ! result
REAL sum(n)
INTEGER n, comm, i, j, ierr
! local sum
DO j= 1, n
  sum(j) = 0.0
  DO i = 1, m
    sum(j) = sum(j) + a(i)*b(i,j)
  END DO
END DO
! global sum

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

CALL MPI_ALLREDUCE(sum, c, n, MPI_REAL, MPI_SUM, 0, comm, ierr)
! return result at all nodes
RETURN

1.10 Reduce-Scatter

MPI includes variants of each of the reduce operations where the result is scattered to all processes in the group on return.

MPI REDUCE SCATTER( sendbuf, recvbuf, recvcounts, datatype, op, comm)

IN  sendbuf  starting address of send buffer (choice)
OUT  recvbuf  starting address of receive buffer (choice)
IN  recvcounts  integer array specifying the number of elements in re-

sult distributed to each process. Array must be identical on all calling processes.

IN  datatype  data type of elements of input buffer (handle)
IN  op  operation (handle)
IN  comm  communicator (handle)

int MPI Reduce scatter(void* sendbuf, void* recvbuf, int *recvcounts,
             MPI Datatype datatype, MPI Op op, MPI Comm comm)
MPI REDUCE SCATTER(SENDBUF, RECVBUF, RECVCOUNTS, DATATYPE, OP, COMM,
             IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER RECVCOUNTS(*), DATATYPE, OP, COMM, IERROR

MPI REDUCE SCATTER first does an element-wise reduction on vector of count = ∑i recvcounts[i] elements in the send buffer defined by sendbuf, count and datatype. Next, the resulting vector of results is split into n disjoint segments, where n is the number of members in the group. Segment i contains recvcounts[i] elements. The ith segment is sent to process i and stored in the receive buffer defined by recvbuf, recvcounts[i] and datatype.

Advice to implementors. The MPI REDUCE SCATTER routine is functionally equivalent to: A MPI REDUCE operation function with count equal to the sum of recvcounts[i] followed by MPI SCATTERV with sendcounts equal to recvcounts. However, a direct implementation may run faster. (End of advice to implementors.)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

1.11 Scan

MPI SCAN( sendbuf, recvbuf, count, datatype, op, comm )

IN  sendbuf  starting address of send buffer (choice)
OUT  recvbuf  starting address of receive buffer (choice)
IN  count  number of elements in input buffer (integer)
IN  datatype  data type of elements of input buffer (handle)
IN  op  operation (handle)
IN  comm  communicator (handle)

int MPI Scan(void* sendbuf, void* recvbuf, int count,
             MPI Datatype datatype, MPI Op op, MPI Comm comm )
MPI SCAN(SENDBUF, RECVBUF, COUNT, DATATYPE, OP, COMM, IERROR)
    <type> SENDBUF(*), RECVBUF(*)
    INTEGER COUNT, DATATYPE, OP, COMM, IERROR

MPI SCAN is used to perform a prefix reduction on data distributed across the group. The operation returns, in the receive buffer of the process with rank i, the reduction of the values in the send buffers of processes with ranks 0,...,i (inclusive). The type of operations supported, their semantics, and the constraints on send and receive buffers are as for MPI REDUCE.

Rationale. We have defined an inclusive scan, that is, the prefix reduction on process i includes the data from process i. An alternative is to define scan in an exclusive manner, where the result on i only includes data up to i-1. Both definitions are useful. The latter has some advantages: the inclusive scan can always be computed from the exclusive scan with no additional communication; for non-invertible operations such as max and min, communication is required to compute the exclusive scan from the inclusive scan. There is, however, a complication with exclusive scan since one must define the “unit” element for the reduction in this case. That is, one must explicitly say what occurs for process 0. This was thought to be complex for user-defined operations and hence, the exclusive scan was dropped. (End of rationale.)

1.11.1 Example using MPI SCAN

Example 1.22 This example uses a user-defined operation to produce a segmented scan. A segmented scan takes, as input, a set of values and a set of logicals, and the logicals delineate the various segments of the scan. For example:

\begin{matrix}values & v_{1} & v_{2} & v_{3} & v_{4} & v_{5} & v_{6} & v_{7} & v_{8} \\ logicals & 0 & 0 & 1 & 1 & 1 & 0 & 0 & 1 \\ result & v_{1} & v_{1} + v_{2} & v_{3} & v_{3} + v_{4} & v_{3} + v_{4} + v_{5} & v_{6} & v_{6} + v_{7} & v_{8} \end{matrix}

The operator that produces this effect is,

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

( \begin{matrix}u \\ i \end{matrix} ) ∘ ( \begin{matrix}v \\ j \end{matrix} ) = ( \begin{matrix}w \\ j \end{matrix} ) ,

where,

w = \{ \begin{matrix}u + v & if i = j \\ v & if i ≠ j \end{matrix} .

Note that this is a non-commutative operator. C code that implements it is given below.

typedef struct {
double val;
int log;
} SegScanPair;
/* the user-defined function
*/
void segScan( SegScanPair *in, SegScanPair *inout, int *len,
MPI_Datatype *dptr )
{
int i;
SegScanPair c;
for (i=0; i< *len; ++i) {
    if ( in->log == inout->log )
        c.val = in->val + inout->val;
    else
        c.val = inout->val;
    c.log = inout->log;
    *inout = c;
    in++; inout++;
}
}

Note that the inout argument to the user-defined function corresponds to the righthand operand of the operator. When using this operator, we must be careful to specify that it is non-commutative, as in the following.

int i,base;
SeqScanPair  a, answer;
MPI_Op       myOp;
MPI_Datatype type[2] = {MPI_DOUBLE, MPI_INT};
MPI_Aint     disp[2];
int          blocklen[2] = { 1, 1};
MPI_Datatype sspair;
/* explain to MPI how type SegScanPair is defined
 */

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

MPI_Address( a, disp);
MPI_Address( a.log, disp+1);
base = disp[0];
for (i=0; i<2; ++i) disp[i] -= base;
MPI_Type_struct( 2, blocklen, disp, type, &sspair );
MPI_Type_commit( &sspair );
/* create the segmented-scan user-op
 */
MPI_Op_create( segScan, False, &myOp );
...
MPI_Scan( a, answer, 1, sspair, myOp, root, comm );

1.12 Correctness

A correct, portable program must invoke collective communications so that deadlock will not occur, whether collective communications are synchronizing or not. The following examples illustrate dangerous use of collective routines.

Example 1.23 The following is erroneous.

switch(rank) {
case 0:
    MPI_Bcast(buf1, count, type, 0, comm);
    MPI_Bcast(buf2, count, type, 1, comm);
    break;
case 1:
    MPI_Bcast(buf2, count, type, 1, comm);
    MPI_Bcast(buf1, count, type, 0, comm);
    break;
}

We assume that the group of comm is {0,1}. Two processes execute two broadcast operations in reverse order. If the operation is synchronizing then a deadlock will occur.

Collective operations must be executed in the same order at all members of the communication group.

Example 1.24 The following is erroneous.

switch(rank) {
case 0:
    MPI_Bcast(buf1, count, type, 0, comm0);
    MPI_Bcast(buf2, count, type, 2, comm2);
    break;
case 1:
    MPI_Bcast(buf1, count, type, 1, comm1);
    MPI_Bcast(buf2, count, type, 0, comm0);
    break;
case 2:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

        MPI_Bcast(buf1, count, type, 2, comm2);
        MPI_Bcast(buf2, count, type, 1, comm1);
        break;
}

Assume that the group of comm0 is {0,1}, of comm1 is {1, 2} and of comm2 is {2,0}. If the broadcast is a synchronizing operation, then there is a cyclic dependency: the broadcast in comm2 completes only after the broadcast in comm0; the broadcast in comm0 completes only after the broadcast in comm1; and the broadcast in comm1 completes only after the broadcast in comm2. Thus, the code will deadlock.

Collective operations must be executed in an order so that no cyclic dependences occur.

Example 1.25 The following is erroneous.

switch(rank) {
    case 0:
        MPI_Bcast(buf1, count, type, 0, comm);
        MPI_Send(buf2, count, type, 1, tag, comm);
        break;
    case 1:
        MPI_Recv(buf2, count, type, 0, tag, comm);
        MPI_Bcast(buf1, count, type, 0, comm);
        break;
}

Process zero executes a broadcast, followed by a blocking send operation. Process one first executes a blocking receive that matches the send, followed by broadcast call that matches the broadcast of process zero. This program may deadlock. The broadcast call on process zero may block until process one executes the matching broadcast call, so that the send is not executed. Process one will definitely block on the receive and so, in this case, never executes the broadcast.

The relative order of execution of collective operations and point-to-point operations should be such, so that even if the collective operations and the point-to-point operations are synchronizing, no deadlock will occur.

Example 1.26 A correct, but non-deterministic program.

switch(rank) {
    case 0:
        MPI_Bcast(buf1, count, type, 0, comm);
        MPI_Send(buf2, count, type, 1, tag, comm);
        break;
    case 1:
        MPI_Recv(buf2, count, type, MPI_ANY_SOURCE, tag, comm);
        MPI_Bcast(buf1, count, type, 0, comm);
        MPI_Recv(buf2, count, type, MPI_ANY_SOURCE, tag, comm);
        break;
    case 2:
        MPI_Send(buf2, count, type, 1, tag, comm);

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

Figure 1.10: A race condition causes non-deterministic matching of sends and receives. One cannot rely on synchronization from a broadcast to make the program deterministic.

        MPI_Bcast(buf1, count, type, 0, comm);
        break;
}

All three processes participate in a broadcast. Process 0 sends a message to process 1 after the broadcast, and process 2 sends a message to process 1 before the broadcast. Process 1 receives before and after the broadcast, with a wildcard source argument.

Two possible executions of this program, with different matchings of sends and receives, are illustrated in figure 1.10. Note that the second execution has the peculiar effect that a send executed after the broadcast is received at another node before the broadcast. This example illustrates the fact that one should not rely on collective communication functions to have particular synchronization effects. A program that works correctly only when the first execution occurs (only when broadcast is synchronizing) is erroneous.

Finally, in multithreaded implementations, one can have more than one, concurrently executing, collective communication call at a process. In these situations, it is the user’s responsibility to ensure that the same communicator is not used concurrently by two different collective communication calls at the same process.

Advice to implementors. Assume that broadcast is implemented using point-to-point MPI communication. Suppose the following two rules are followed.

1. All receives specify their source explicitly (no wildcards).

2. Each process sends all messages that pertain to one collective call before sending any message that pertain to a subsequent collective call.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

Then, messages belonging to successive broadcasts cannot be confused, as the order of point-to-point messages is preserved.

It is the implementor’s responsibility to ensure that point-to-point messages are not confused with collective messages. One way to accomplish this is, whenever a communicator is created, to also create a “hidden communicator” for collective communication. One could achieve a similar effect more cheaply, for example, by using a hidden tag or context bit to indicate whether the communicator is used for point-to-point or collective communication. (End of advice to implementors.)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47