Collective operations and communicators

Collective operations

A collective operation is performed by all processes of a communicator: each process calls the same function with consistent arguments. Collectives are simpler and usually faster than loops with MPI_Send/MPI_Recv: the implementation chooses an algorithm for the number of processes and the data size. The main operations are shown in Fig. 12.5 and Table 12.7.

Figure 12.5. MPI collective operations

Table 12.7. Collective operations

Operation	Action
MPI_Bcast	broadcast the root’s data to all processes
MPI_Scatter	split the root’s array into equal parts: part $i$ goes to process $i$
MPI_Gather	gather equal parts from all processes into the root’s array
MPI_Scatterv, MPI_Gatherv	the same for parts of different sizes: arrays of counts counts and displacements displs
MPI_Allgather	gather the parts in all processes
MPI_Alltoall	each process sends part $i$ to process $i$ (“transpose”)
MPI_Reduce	combine values with an operation (MPI_SUM, MPI_PROD, MPI_MIN, MPI_MAX, MPI_LAND, MPI_MINLOC …) at the root
MPI_Allreduce	the same with the result in all processes
MPI_Scan	prefix reduction: process $i$ receives the result for processes $0\dots i$
MPI_Barrier	synchronization without data

// Root 0 has size · 2 numbers; each process receives 2.
std::vector<int> send, recv(size), mine(2);
if (rank == 0)
    for (int i = 0; i < size * 2; ++i) send.push_back(i * 10);
MPI_Scatter(send.data(), 2, MPI_INT, mine.data(), 2, MPI_INT, 0,
            MPI_COMM_WORLD);
int local = mine[0] + mine[1];
MPI_Gather(&local, 1, MPI_INT, recv.data(), 1, MPI_INT, 0,
           MPI_COMM_WORLD);                // on rank 0: 10 50 90 130

In MPI_Scatter and MPI_Gather, the element count is given per process, and the root’s array has size parts. The MPI_IN_PLACE value in place of the send buffer means “the data is already in the result buffer”: MPI_Allreduce(MPI_IN_PLACE, &sum, 1, MPI_DOUBLE, MPI_SUM, comm) replaces the local sum with the global one.

Rules for collective operations:

• all processes of the communicator call them in the same order; if one process skips a call (for example, because of if (rank != 0)), the program hangs;
• collective operations have no tags and do not mix with point-to-point operations;
• except for MPI_Barrier, they do not necessarily synchronize processes: the root of MPI_Bcast may return before the others receive the data;
• reductions on floating-point numbers combine values in an order that depends on the number of processes, so the last digits of a sum differ (as with OpenMP reductions).

Algorithms. A naive MPI_Bcast implemented as sequential MPI_Send calls from the root takes $p-1$ transfers. Implementations use a binomial tree: at each step, every process that already has the data passes it to another, and all $p$ processes receive the data in ${\log }_{2}p$ steps. For large arrays, MPI_Allreduce runs as a ring: the array is split into $p$ parts that travel around the ring so that each process transfers only about $2n$ elements regardless of $p$. Open MPI chooses the algorithm automatically (the coll/tuned component).

The MPI-3 standard added nonblocking collective operations such as MPI_Ibcast and MPI_Iallreduce, which return an MPI_Request and complete with MPI_Wait, and MPI-4.0 added persistent collectives for exchanges that repeat in a loop.

Derived types, communicators, and topologies

Derived datatypes

To send a structure or non-contiguous array elements in one message, you create a derived datatype: a description of how the data is laid out in memory. The type is created, registered with MPI_Type_commit, and freed after use with MPI_Type_free:

• MPI_Type_contiguous(n, old, &t) describes $n$ contiguous elements;
• MPI_Type_vector(count, blocklen, stride, old, &t) describes count blocks of blocklen elements with stride stride; this is how a column of a row-major matrix is described;
• MPI_Type_create_struct describes a structure with fields of different types.

struct Sale
{
    int shop;          // store number
    double amount;     // amount, UAH
    char date[11];     // "2026-09-18"
};

int lengths[3] = {1, 1, 11};
MPI_Aint offsets[3] = {offsetof(Sale, shop), offsetof(Sale, amount),
                       offsetof(Sale, date)};
MPI_Datatype types[3] = {MPI_INT, MPI_DOUBLE, MPI_CHAR};
MPI_Datatype saleType;
MPI_Type_create_struct(3, lengths, offsets, types, &saleType);
MPI_Type_commit(&saleType);
MPI_Send(sales, 2, saleType, 1, 0, MPI_COMM_WORLD);   // 2 records
// …
MPI_Type_free(&saleType);

Field offsets are taken with the offsetof macro (header <cstddef>) rather than computed by hand: the compiler aligns fields, and there are 4 unused bytes between int and double. The type’s extent, returned by MPI_Type_get_extent, equals sizeof(Sale) = 32 bytes here, so an array of records is transferred correctly. Structures with std::string or std::vector cannot be sent this way: they contain pointers into the process’s memory.

New communicators

The MPI_Comm_split(comm, color, key, &newcomm) function divides processes into groups: processes with the same color end up in one new communicator, and key sets the order of ranks in it. Collective operations in the new communicator involve only its processes.

int color = rank % 2;                     // even and odd ranks
MPI_Comm half;
MPI_Comm_split(MPI_COMM_WORLD, color, rank, &half);
int halfRank, halfSize, sum = 0;
MPI_Comm_rank(half, &halfRank);
MPI_Comm_size(half, &halfSize);
MPI_Allreduce(&rank, &sum, 1, MPI_INT, MPI_SUM, half);
std::println("world {} -> group {}: rank {} of {}, sum {}",
             rank, color, halfRank, halfSize, sum);
MPI_Comm_free(&half);

world 0 -> group 0: rank 0 of 2, sum 2
world 1 -> group 1: rank 0 of 2, sum 4
world 2 -> group 0: rank 1 of 2, sum 2
world 3 -> group 1: rank 1 of 2, sum 4

This is how, for example, computations along the rows and columns of a process grid, or separate groups for independent subtasks, are organized. The MPI_Comm_split_type function with MPI_COMM_TYPE_SHARED creates a communicator of processes on the same node, which is useful for hybrid programs.

Cartesian topologies

For grid problems, it is convenient to arrange processes in a Cartesian topology, a grid of dimension 1, 2, or 3 (Fig. 12.6):

int dims[2] = {0, 0}, periods[2] = {1, 1};  // torus: edges wrap around
MPI_Dims_create(size, 2, dims);             // 4 → 2 × 2, 8 → 4 × 2
MPI_Comm grid;
MPI_Cart_create(MPI_COMM_WORLD, 2, dims, periods, 1, &grid);
int up, down, left, right;
MPI_Cart_shift(grid, 0, 1, &up, &down);     // neighbors along rows
MPI_Cart_shift(grid, 1, 1, &left, &right);  // neighbors along columns

MPI_Dims_create picks the most “square” grid possible, MPI_Cart_create creates a communicator with the topology (the reorder = 1 parameter lets the library renumber processes, so the rank is taken from the new communicator), MPI_Cart_coords returns a process’s coordinates, and MPI_Cart_shift returns the ranks of the neighbors in a given dimension. At the edge of a non-periodic grid, the neighbor equals MPI_PROC_NULL: an exchange with it does nothing, so boundary processes need no special conditions.

Figure 12.6. Domain decomposition and halo exchange

Domain decomposition divides a grid into blocks, one per process. Computing the boundary nodes of a block requires values from neighboring blocks, so each block has a frame of halo cells (halo, ghost cells) with copies of the neighbors’ boundary nodes. Before each step, processes exchange halos (MPI_Sendrecv with each neighbor) and then compute their own nodes independently. The amount of communication is proportional to the block’s perimeter, and the amount of computation to its area, so a two-dimensional decomposition beats strips when there are many processes. An example of a Cartesian topology for the Game of Life is given in the lab.

Parallel algorithms with message passing

Most MPI programs follow a few typical schemes (Table 12.8).

Table 12.8. Typical schemes of MPI parallel algorithms

Scheme	Implementation with MPI
integration, Monte Carlo	each process handles its share of steps or random points (a generator with a separate seed per rank), and MPI_Reduce sums the results; only a few numbers are exchanged
manager–worker (master–worker)	rank 0 hands out tasks one at a time, and a worker receives the next one after replying (MPI_ANY_SOURCE, “work”/“stop” tags); dynamic distribution for uneven tasks
matrix multiplication	MPI_Bcast of matrix $B$, MPI_Scatter of rows of $A$, local multiplication, MPI_Gather of rows of $C$; for large matrices, blocks on a process grid (Cannon’s algorithm)
odd–even sort	each process sorts its part; $p$ phases of exchange with a neighbor (MPI_Sendrecv of the parts), and after merging, the lower process keeps the smaller elements
Jacobi method, cellular automata	domain decomposition, halo exchange with neighbors at every step, MPI_Allreduce for the error norm

The common rule is to minimize the number and volume of messages. Every message has a fixed latency, so one message of 1000 numbers is much cheaper than 1000 messages of one number each; data needed by everyone is distributed with a collective operation rather than from each process separately. The Mandelbrot set in the lab is an example of the manager–worker scheme, and the Jacobi method with decomposition is shown in the “Program examples” section.