Distributed computing: overview#

For multi-GPU and multi-node systems, Ginkgo provides distributed Matrix and Vector types under gko::experimental::distributed. This page introduces the abstraction, the host-first assembly pattern, when to choose distributed over single-device, and the communicator. Per-topic detail (partitioning, communication, distributed-aware solvers) lives in the sub-pages.

When to go distributed#

Distributed is the right choice when a single device’s memory is no longer sufficient or when a multi-node MPI job demands a single matrix view spanning ranks:

  • Memory pressure. A GPU typically becomes the bottleneck around \(10^7\)–\(10^8\) unknowns. Once the matrix and vectors no longer fit in one device’s VRAM, distributing across multiple GPUs (or CPU ranks) is the natural next step.

  • Multi-node MPI runs. If your simulation already uses MPI for domain decomposition, Ginkgo’s distributed types let you express the linear algebra at the same granularity as your physics.

  • Single-device otherwise. If the problem fits comfortably on one GPU, stay single-device. Distributed is more code, introduces communication overhead, and has more failure modes.

The distributed types#

Two primary types live under gko::experimental::distributed:

  • Matrix<V, L, G>: a distributed sparse matrix. V is the value type, L is the local index type (typically gko::int32), and G is the global index type (typically gko::int64).

  • Vector<V>: a distributed dense vector or multi-vector.

Both inherit from LinOp, so they compose with solvers and preconditioners using exactly the same patterns as single-device objects. A CG solver doesn’t know whether the matrix it operates on is distributed; it just calls apply.

The host-first assembly pattern#

Construction is host-first by convention: assemble on a host executor, then transfer to the device executor. read_distributed runs on the executor the matrix was created on (it clones the partition there first), so assembling on the host and cloning to the device afterwards is the portable pattern:

namespace dist = gko::experimental::distributed;

auto host = gko::ReferenceExecutor::create();
auto exec = gko::CudaExecutor::create(0, host);
auto comm = gko::experimental::mpi::communicator(MPI_COMM_WORLD);

// Each rank collects entries of the global 4 x 4 matrix; rows owned by
// other ranks are discarded by read_distributed.
gko::matrix_data<double, gko::int64> data{gko::dim<2>{4, 4}};
data.nonzeros.emplace_back(0, 0, 2.0);  // (global row, global column, value)

// Partition: ranges {0, 2, 4} give rows 0-1 to rank 0 and rows 2-3 to rank 1.
auto partition = gko::share(
    dist::Partition<gko::int32, gko::int64>::build_from_contiguous(
        host, gko::array<gko::int64>(host, {0, 2, 4})));

// Assemble the distributed matrix on the host.
auto mat_host = dist::Matrix<double, gko::int32, gko::int64>::create(host, comm);
mat_host->read_distributed(data, partition);

// Clone to the device for computation.
auto mat = mat_host->clone(exec);

read_distributed takes the partition as a std::shared_ptr (hence gko::share) and keeps the entries of the rows this rank owns. Once assembled, clone(exec) copies the local blocks to the device in one step.

Distributed Vector assembly follows the same pattern:

auto vec_host = dist::Vector<double>::create(host, comm, global_size, local_size);
// Fill local entries ...
auto vec = vec_host->clone(exec);

The communicator#

The distributed Matrix, Vector and RowGatherer hold a gko::experimental::mpi::communicator:

auto comm = mat->get_communicator();
int rank  = comm.rank();
int size  = comm.size();
MPI_Comm raw = comm.get();   // for direct MPI use if needed

Rank 0 is the leader by convention. No rank holds global state; each rank owns only its local part of the matrix and vector data. Global operations (dot products, norms) reduce across ranks internally using MPI_Allreduce.

Block-structured view across ranks#

Conceptually, the global matrix is permuted by the row partition so that the rows owned by each rank form a contiguous block. Within that block-row, the columns split by ownership (per the column partition, by default the same as the row partition). This gives every rank two pieces of the matrix:

  • Local matrix (the diagonal block): entries whose row is owned by this rank and whose column is also owned by this rank. Across all ranks, these pieces sit on the block-diagonal of the permuted global matrix; hence the name. Both row and column indices are local (0..n_local − 1) and the dimensions are \(n_\text{local rows} \times n_\text{local cols}\). This is the bulk of the work for well-partitioned matrices.

  • Non-local matrix (the off-diagonal blocks): entries whose row is owned by this rank but whose column is not owned by this rank. These sit on the off-diagonal blocks of the permuted global matrix and drive the halo exchange during apply. The storage is compressed: only the remote columns that this rank’s local rows actually reference are kept, and those columns are renumbered into a compact \(0..K-1\) local indexing via the matrix’s index_map. So its dimensions are \(n_\text{local rows} \times K\), where \(K\) is the number of distinct remote global columns this rank touches, not the total number of remote columns. The original global column IDs are tracked separately in index_map::get_remote_global_idxs() and used to unpack the values received in the halo exchange.

Note

“Diagonal” here refers to the block structure of the partitioned global matrix, not to the diagonal of the matrix in the linear-algebra sense. Ginkgo’s API calls this piece the local matrix: an arbitrary sparse sub-block of \(A\) that this rank owns end-to-end. The non-local matrix is similarly a sparse block; “off-diagonal” describes its position in the partitioned global matrix, not anything about its values.

A worked example#

Take a 4 × 4 matrix on 2 ranks. Two partitions are involved, and they can be different functions, though by default Ginkgo uses the row partition for both:

  • Row partition: assigns each global row index to one rank.

  • Column partition: assigns each global column index to one rank.

Saying “rank \(R\) owns row \(i\)” means the row partition maps \(i\) to \(R\); saying “rank \(R\) owns column \(j\)” means the column partition maps \(j\) to \(R\). Ownership is a property of the partition, not of which rank physically stores something; those are related but distinct questions, and we will see how, below.

For this example, let both partitions assign indices {0, 1} to rank 0 and {2, 3} to rank 1. Suppose the global sparsity pattern is:

                       column partition
                ┌── cols owned by ──┬── cols owned by ──┐
                │      rank 0       │      rank 1       │
                │  col 0    col 1   │  col 2    col 3   │
                ├───────────────────┼───────────────────┤
                │                   │                   │
   row 0  ───┐  │   *        *      │   *        .      │
   row 1     │  │   *        *      │   *        .      │   ← rows owned by rank 0
             │  │                   │                   │
             │  ├───────────────────┼───────────────────┤
             │  │                   │                   │
   row 2     │  │   .        .      │   *        *      │   ← rows owned by rank 1
   row 3  ───┘  │   *        .      │   *        *      │
                └───────────────────┴───────────────────┘
                  ↑                    ↑
                  │                    │
            cols 0,1 belong to    cols 2,3 belong to
            rank 0 (per the       rank 1 (per the
            column partition,     column partition,
            globally)             globally)

The four quadrants of the matrix are defined by crossing row ownership with column ownership. Each quadrant has a name based on the row owner and on whether the column owner matches:

                       cols owned by R0     cols owned by R1
                       ┌──────────────────┬──────────────────┐
   rows owned by R0    │   R0's diagonal  │ R0's off-diagonal│   ←─ stored on rank 0
                       │  (R0 owns both   │ (R0 owns rows,   │
                       │   rows and cols) │  R1 owns cols)   │
                       ├──────────────────┼──────────────────┤
   rows owned by R1    │R1's off-diagonal │  R1's diagonal   │   ←─ stored on rank 1
                       │ (R1 owns rows,   │ (R1 owns both    │
                       │  R0 owns cols)   │  rows and cols)  │
                       └──────────────────┴──────────────────┘

Where each block is stored is determined by the row owner alone: every rank holds the two blocks of the global matrix that lie in its own row range. So:

  • Rank 0 stores its top-left quadrant (its local matrix) and its top-right quadrant (its non-local matrix). The cols 2, 3 block lives on rank 0 because rows 0, 1 live on rank 0, even though rank 1 is the one that owns those columns.

  • Rank 1 stores the bottom-left quadrant (its non-local matrix) and the bottom-right quadrant (its local matrix). Symmetric story.

Plugging in the concrete sparsity pattern above, the per-rank storage is:

Rank 0 (holds rows 0, 1):
                                          index_map for the non-local cols:
  local matrix         non-local            local col 0  ↔  global col 2
  (2 × 2)              matrix (2 × K)       (col 3 is not touched — dropped)
  ┌────────────┐       ┌─────┐
  │  *    *    │       │  *  │              K = 1: only one of the two remote
  │  *    *    │       │  *  │              columns is actually referenced.
  └────────────┘       └─────┘

Rank 1 (holds rows 2, 3):
                                          index_map for the non-local cols:
  local matrix         non-local            local col 0  ↔  global col 0
  (2 × 2)              matrix (2 × K)       (col 1 is not touched — dropped)
  ┌────────────┐       ┌─────┐
  │  *    *    │       │  .  │              K = 1: again, only one of the two
  │  *    *    │       │  *  │              remote columns is referenced.
  └────────────┘       └─────┘

Note that get_non_local_matrix() on rank 0 returns a 2 × 1 sparse block, not a 2 × 2 block: the remote column the rank doesn’t touch (global col 3) is simply not stored. The mapping from the compressed local column to its global index lives in index_map::get_remote_global_idxs() and is used during the halo exchange to know which entries of \(x\) this rank should expect from which other rank.

Access the parts directly if needed:

auto local_mat     = mat->get_local_matrix();      // shared_ptr<const LinOp>: columns owned by this rank
auto non_local_mat = mat->get_non_local_matrix();  // columns owned by other ranks (compressed columns)

// Vectors expose the rank-local Dense:
auto local_x = x->get_local_vector();             // matrix::Dense<V> with rank-owned rows only

Generic code that accepts both distributed and plain vectors takes the local block with a dynamic_cast:

const gko::LinOp* b = nullptr;  // a dist::Vector<double> or a matrix::Dense<double>
const gko::matrix::Dense<double>* local_b = nullptr;
if (auto dist_b = dynamic_cast<const dist::Vector<double>*>(b)) {
    local_b = dist_b->get_local_vector();
} else {
    local_b = gko::as<gko::matrix::Dense<double>>(b);
}

Solver implementations apply the same idea through gko::experimental::precision_dispatch_real_complex_distributed, which dispatches on distributed and non-distributed inputs alike.

Distributed sub-pages#

Distributed

See also