Collective Buffering and Two-Phase I/O for Parallel Checkpoints

A checkpoint writes the state of every rank to stable storage. On a cluster this is the phase where a job that spent an hour computing spends minutes transferring, and the naive implementation of it is the one most codes use first: each rank calls fopen on a file named after its rank, writes its own block, and closes it. It works, it requires no library support, and it degrades badly as the process count rises.

The degradation is not primarily about bandwidth. It is about how many independent streams the file system can absorb before the metadata operations and the small writes overwhelm the storage servers’ ability to coalesce them.

Why per-rank files stop working

A parallel file system distributes each file across a set of storage targets, typically by striping fixed-size chunks round-robin. A file of one megabyte with a one-megabyte stripe unit lands entirely on one target, so the write goes to one server while all the others are idle. A rank that writes its own file therefore produces a single-target write, and the aggregate throughput of the job is the throughput of one target multiplied by nothing.

Scaling the stripe unit down helps one file but hurts another. A stripe unit of 64 KB spreads small files widely, and then each write is smaller than the stripe unit, so the client must perform a read-modify-write of a partial block or the server must handle partial writes. Both cost more than a full-unit transfer.

The second problem is stream count. Metadata operations open, close, and create a file, and each is a network round trip to a server that may be shared by thousands of clients. A job with a hundred thousand ranks creates a hundred thousand files and performs a hundred thousand creates and opens within a short window. The metadata service becomes the bottleneck long before the data path does, and the checkpoint time is dominated by operations that transfer no payload at all.

The third problem is that the job now has a hundred thousand files to manage, to move, and to clean up, and the number of files is not a property the application wants to expose.

One file, many views

The alternative is a single file with each rank writing a disjoint region of it. This is what MPI-IO is for: the interface separates the file from the view each rank has of it, so a rank can write a block of a global array without knowing where the other blocks are.

The mechanism is a file view, set with MPI_File_set_view. The view is a displacement, an elementary datatype, and a filetype. The elementary datatype is the unit the implementation works in, usually a byte type; the filetype describes the pattern of etypes that this rank’s view maps onto the file. A rank whose block is a contiguous run of doubles at offset rank * n * 8 uses a filetype of n contiguous etypes, which requires repeating the type n times or using a contiguous type constructor.

For a three-dimensional decomposition the filetype is not contiguous. A rank owns a block of an (N, N, N) array laid out in row-major order, and its block consists of nz slices, each of which is ny rows of nx contiguous elements separated by a stride of N. The filetype is a nested constructor: a vector of ny contiguous runs of nx doubles with a stride of N, nested inside a vector of nz such slabs with a stride of N * N. The implementation then knows exactly which file offsets the rank will touch, and it can generate the read or write without any application-level packing.

#include <mpi.h>

/* Create the filetype for one block of an N^3 grid decomposed into
   nx x ny x nz points, starting at (ox, oy, oz), row-major order. */
MPI_Datatype block_type(long N, long nx, long ny, long nz)
{
    MPI_Datatype row, slab;
    MPI_Type_contiguous(nx, MPI_DOUBLE, &row);           /* one row */
    MPI_Type_create_hvector(ny, 1, (MPI_Aint)N * sizeof(double), row, &slab);
    MPI_Datatype blk;
    MPI_Type_create_hvector(nz, 1, (MPI_Aint)N * N * sizeof(double), slab, &blk);
    MPI_Type_commit(&blk);
    MPI_Type_free(&row);
    MPI_Type_free(&slab);
    return blk;
}

/* Displacement of the block origin in bytes. */
MPI_Aint block_offset(long N, long ox, long oy, long oz)
{
    return (MPI_Aint)((oz * N * N + oy * N + ox) * sizeof(double));
}

The filetype is a description, not a data movement. A single MPI_File_write_at_all call with this type can describe a write that touches nz * ny separate contiguous runs, without the application copying anything into a contiguous buffer.

Two-phase I/O

Describing the pattern does not by itself make the access efficient. The implementation still has to move the data from each rank’s memory into the file, and the access pattern of the collective write is the union of all ranks’ views.

ROMIO’s approach is two phases. In the first, the ranks shuffle data among themselves so that each rank ends up holding a contiguous chunk of the file, aligned to the file system’s stripe unit. In the second, each rank issues one large write to the file for the region it now owns. The shuffle is a message-passing operation over the same interconnect the application uses; the write is a set of large, stripe-aligned requests that the file system handles efficiently.

The gain comes from replacing many small, unaligned writes with a few large ones. A rank whose block spans several stripe units would otherwise issue a write per unit, and each write is a separate request with its own overhead. After the shuffle, each rank writes whole units, and the number of requests falls by roughly the ratio of the chunk size to the request size.

The shuffle has a cost. It performs an all-to-all exchange, so its time is bounded below by the same bisection bandwidth limit that applies to any global permutation. It also needs buffer space to hold the chunk, and the buffer size is a tuning parameter: too small and the write requests are again too small, too large and the memory is unavailable on a node that is already holding the simulation state.

The implementation chooses between this and a simpler direct path based on the access pattern and the file system. For an access pattern that is already contiguous and stripe-aligned per rank, the shuffle buys nothing and the direct write is used. For a pattern that is strided, the shuffle is essential. The choice is not exposed as a flag in the portable interface; it is a property of the implementation, which is why the same code can perform differently on two machines with the same hardware.

Consistency, atomicity, and the cost of being careful

Parallel I/O has to reconcile two models. The file is a shared object, and POSIX defines the result of concurrent accesses in terms of atomic operations and write visibility. The message-passing program has no shared memory and no shared open file description, and the ranks may not even be able to see the file if the underlying system is not POSIX-compliant.

The MPI-IO standard addresses this with a set of consistency and atomicity levels. The default is that the program is responsible for avoiding conflicting accesses: two ranks must not write to overlapping regions without intervening synchronization, and the result of doing so is not defined. The relaxed mode removes some of the file pointer sharing requirements and can be faster because the implementation need not check for conflicts. The atomic mode requires that overlapping accesses behave as if they occurred in some order, which forces the implementation to serialize them and costs performance.

The practical rules follow. Every rank must use a distinct view, or the writes conflict. Collective operations must be called by all ranks in the communicator, in the same order, with the same arguments where the interface requires them; the collective write can then be optimized globally. Barriers are needed between a write and a read of the same region by another rank, because the file system may buffer, and a barrier in MPI does not flush the file system’s cache. The flush operation, MPI_File_sync, is what makes the data visible, and it is a separate, expensive call.

The other consistency question is between the checkpoint and the compute phase. A checkpoint that is written while the simulation continues produces an inconsistent image unless the state is frozen. The halo exchange pattern makes this concrete: if ghost cells are being updated while the interior is being written, the file contains a mixture of two iterations. Checkpoints therefore serialize against the time step, which means the write time adds directly to the run time rather than overlapping with it, unless the state is copied to a separate buffer first.

The storage hierarchy below the file system

The file system is not the last layer. A parallel file system is fronted by a set of I/O nodes, and behind them are disk arrays or, increasingly, solid-state devices and a tier of burst buffers.

A burst buffer is a layer of fast storage on the compute nodes or on a dedicated set of nodes, sized to absorb a checkpoint quickly and drain it to the parallel file system afterwards. The write to the buffer is fast because it does not traverse the file system’s metadata path or its network; the drain is slow but happens in the background. The trade is capacity against latency: the buffer holds a few checkpoints, not a whole campaign, and a job that checkpoints more often than it drains fills the buffer and blocks.

The same arithmetic governs the checkpoint interval. If a checkpoint costs c seconds and the job runs for T seconds between failures, writing every t seconds costs c/t of the runtime in overhead and loses at most t/2 seconds of work per failure on average. Minimizing the total cost gives an interval proportional to the square root of cT, which is the classical result for checkpoint placement. The formula assumes independent failures and a constant cost, and both assumptions are poor on a large machine, but the shape is right: as the checkpoint cost rises with the process count, the optimal interval lengthens, and a code that checkpoints on a fixed schedule becomes progressively less efficient.

The scaling argument applies to storage as it does to the network. The number of storage targets is fixed by the installation, so the per-rank share of write bandwidth falls as the job grows. A checkpoint that takes a minute at a thousand ranks takes longer at ten thousand even if the per-rank data shrinks, because the aggregate demand grows faster than the provisioned capacity. That is the ceiling that parallel I/O shares with every other resource in the machine, and it is the reason checkpoint cost is measured in the same scaling studies as iteration time.