Ranks, Communicators, and the SPMD Execution Model

A cluster program starts differently from a threaded one. There is no master thread that creates workers; the runtime is given a process count and a set of hosts, and it starts that many copies of the same executable. Each copy runs main from the beginning, initializes the message-passing library, asks which copy it is, and then follows the same source text down a branch selected by the answer. That arrangement is the single program, multiple data model, and most of the constraints in a distributed-memory code follow from it rather than from the network.

The model is old enough to look obvious and specific enough to punish assumptions. Two of those assumptions cause most of the early failures: that a rank identifies a place, and that a process index is a sufficient address for a message.

One executable, many processes

The first library call is MPI_Init, which takes the addresses of the argc and argv that main received. It must precede every other call except the handful that report whether initialization has already happened, and it may be called only once in the life of a process. MPI_Finalize is the matching call at the end, and every process in the job must make it. Between those two points the process is free to call the rest of the interface.

Two accessors establish the shape of the job. MPI_Comm_rank returns an integer in the range zero to size minus one, and MPI_Comm_size returns the count. The ranks are contiguous and each appears exactly once. Nothing else is promised. In particular, the standard does not say that rank 0 is on the head node, that consecutive ranks are on the same host, or that rank order follows any cabling order. Placement is decided by the launcher and the scheduler, and a program that infers locality from rank arithmetic is relying on an implementation detail. The topology article on bisection bandwidth and fat trees covers what that placement costs when it goes wrong.

Process launch itself is outside the standard. The mpiexec command is an implementation convention rather than a standardized interface, and its flags differ between MPICH-derived and Open MPI-derived launchers. MPI 4 added sessions, which let processes establish a world communicator without sharing a common parent, but even there the mechanism that starts the processes is supplied by the environment rather than by the specification. This is deliberate: the standard describes communication between processes that already exist, so that it can run on a batch system, an interactive shell, or a dynamic process manager without change.

The executable is shared in the sense that every rank runs the same file. It is not shared in any other sense. Static data, the heap, the stack, and open file descriptors are per process. A rank that allocates a 4 GB array causes every rank to allocate 4 GB, which is why memory per node, not total cluster memory, sets the largest problem a job can hold.

What a communicator carries that a rank does not

If ranks were addresses, a message could be sent to rank 3 and that would be the end of it. The interface instead requires a communicator argument on every send, receive, and collective, and the reason is that source and tag alone are not enough to identify a message.

A communicator is an ordered group of processes together with a context. The context is an identifier that the implementation attaches to every message sent through that communicator, and matching uses the triple of communicator, source, and tag. Without the context, two independent pieces of code that both send tag 7 between the same pair of ranks would consume each other’s messages. That happens readily when a solver library and a checkpoint library are linked into one executable: each is internally consistent, and the combination is not. Giving each library a private communicator makes the two conversations disjoint by construction.

Communicators are created collectively. MPI_Comm_dup produces a new communicator with the same group and a fresh context, and every member of the old communicator must call it. MPI_Comm_split partitions the parent group: each process passes a color, processes with the same color land in the same child communicator, and a key orders them within it. Passing the undefined-color sentinel excludes a process from every result. The call is collective over the parent even for processes that exclude themselves, which surprises people who expect a filtering operation to be local.

#include <mpi.h>
#include <stdio.h>

int main(int argc, char **argv)
{
    int rank, size;
    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    /* Interpret the flat rank as a row-major 2-D process grid. */
    int p = 0;
    while (p * p < size && (p + 1) * (p + 1) <= size) p++;
    int row = rank / p, col = rank % p;

    MPI_Comm row_comm;
    MPI_Comm_split(MPI_COMM_WORLD, row, col, &row_comm);

    int row_rank, row_size;
    MPI_Comm_rank(row_comm, &row_rank);
    MPI_Comm_size(row_comm, &row_size);
    printf("world %d/%d -> grid (%d,%d), row rank %d of %d\n",
           rank, size, row, col, row_rank, row_size);

    MPI_Comm_free(&row_comm);
    MPI_Finalize();
    return 0;
}

The example also shows the usual reason to split: a two-dimensional process grid in which each row of the grid needs its own collective domain. A row-wise reduction should not involve the other rows, and giving each row a communicator expresses that directly.

Context identifiers are a bounded resource. Implementations reserve a pool of them per process, and a program that creates communicators inside a loop without freeing them will eventually fail to create another. The pool is large enough that ordinary code never notices, but a library that duplicates a communicator on every call is a latent failure. Create communicators once, during setup, and free them at teardown.

The cost of private address spaces

Nothing is shared, so every value that two ranks need to agree on must be turned into bytes and moved. A pointer is meaningless outside the process that created it, and a linked list cannot be sent without a traversal that assigns indices. This is the tax that the model charges for eliminating data races: there is no way for two ranks to write the same cache line, because there is no cache line they both can name.

The tax shows up in three places. Data structures have to be described rather than passed, which is why derived datatypes exist: they let a strided or heterogeneous layout be sent without a pack loop, and the implementation can use the network’s scatter-gather support instead of a copy. Global indexing has to be computed, since a rank’s local array indices are not the global ones; the halo exchange article on surface-to-volume ratio works through the arithmetic. And every rank pays for the runtime: an interpreter, a garbage collector, or a large statically linked library is multiplied by the number of ranks on a node.

The isolation also changes failure behaviour. A segmentation fault in one rank terminates that process, but the remaining ranks are typically blocked in a receive or a collective and do not learn about it. The job hangs until the launcher notices that a process exited and kills the rest. This is why batch systems report a failed job as a timeout as often as a signal, and why launchers watch the whole process group rather than the first exit status.

Where the model breaks

The most common structural error is putting a collective inside a branch that depends on rank. A reduction called only by rank 0 waits forever for the other members, and no amount of network tuning fixes it. The rule is exact: every process in a communicator must call the same sequence of collectives on that communicator, in the same order, with compatible arguments. The companion article on collective cost models explains why the implementations cannot detect a mismatch.

A second error is assuming that a send will complete before its matching receive is posted. Point-to-point semantics are subtle enough to deserve their own treatment, and the matching and ordering rules are where naive exchange patterns deadlock. The short version is that the standard permits a send to block until a receive is posted, so a program that relies on buffering is relying on a message size threshold it did not choose.

A third is oversubscription. When the launcher starts more ranks than there are cores, each rank still runs a progress loop that polls the network. The loops compete for the same cores, and the job slows by far more than the oversubscription factor suggests. Binding ranks to distinct cores is a launcher responsibility, and the default binding is not always what the rank count implies.

Finally, the rank count is not a tuning knob that can be changed without thought. Decompositions that divide a domain by the rank count produce different partitions at different counts, and a solver whose iteration count depends on the partition will change its answer slightly when the job size changes. That is a property of the algorithm, not of the library, and it is the reason scaling studies fix the problem size and vary the process count deliberately rather than incidentally.