MPI & interconnects

MPI & interconnects#

Threads scale you within one machine. To use many machines — a cluster — you need MPI, the distributed-memory model from the primer. This chapter introduces MPI from zero, shows the two ways qc-rs uses it, and covers the part that trips everyone up on a real cluster: the interconnect (the network between machines) and how to select it.

What MPI is#

MPI (Message Passing Interface) is the standard for distributed-memory parallelism. An MPI job launches several processes — called ranks, numbered 0, 1, 2, … — possibly on different machines. Each rank has its own private memory and runs the same program; to share data, ranks send messages to one another over the network. This is the primer’s distributed-memory model made concrete:

  • Ranks are the unit of work (one rank ≈ one CPU core group, often one per NUMA domain).

  • Each rank holds only part of the big data (e.g. its slab of the integrals), so the whole problem can be far larger than one machine’s memory.

  • Ranks coordinate with collectives (all ranks combine a value — a sum, a broadcast) and point-to-point messages.

qc-rs uses MPI to distribute the memory-hungry integral/Fock work across ranks so a calculation too big for one node fits across several.

Two ways to run MPI in qc-rs#

1. run(nmpi=..., hosts=...) — qc-rs relaunches for you#

The easy path: ask .run() for ranks, and qc-rs relaunches itself under mpirun in a subprocess:

import qc
m = qc.chk.new(atom="...", ao="cc-pvtz")
m = m.scf(ref="r").run(nthread=8, nmpi=4, hosts="nodeA:2,nodeB:2")

  • nmpi = number of ranks (nmpi=1, the default, runs in-process — no MPI).

  • nthread = cores per rank (required when nmpi>1).

  • hosts = where the ranks go: None (all local), "nodeA:2,nodeB:2" (explicit slots), or "nodeA,nodeB" (even split).

nmpi>1 prints a parallel-plan banner — total ranks, threads per rank, nodes, and a per-node table showing the budget ranks × nthread against the node’s core count, flagging oversubscription (⚠). Oversubscription is an error by default (the parallel.strict IOP; set False to only warn). Keep ranks_per_node × nthread ≤ cores_per_node, exactly as the threads chapter warned.

2. External mpirun + comm= — you launch, qc-rs joins#

The HPC path: you launch the job with mpirun/mpiexec yourself and hand qc-rs the communicator:

import mpi4py
mpi4py.rc.thread_level = "serialized"     # REQUIRED for the UCX PML — see caveats
from mpi4py import MPI
import qc

m = qc.chk.new(atom="...", ao="cc-pvtz")
m = m.scf(ref="r").run(nthread=8, comm=MPI.COMM_WORLD, log_rank="root")

Run it with mpirun -np 4 python script.py. With a comm, nmpi is ignored (the launcher already set the ranks). log_rank controls which ranks print: "root" (default, rank 0 only), "all", "gather", or a list — this keeps a 128-rank job from flooding the output.

Tip

Which to use nmpi=/hosts= is convenient for a laptop or an interactive node. The comm= path is standard on a cluster with a scheduler (PBS/Slurm), where the job script already runs under mpirun with the right host list. Cross-host nmpi= runs additionally need a shared filesystem for the relaunch temp files.

The interconnect: why the network matters enormously#

On a cluster the ranks talk over a network, and its speed can dominate a communication-heavy step. Two kinds matter:

  • Ethernet / TCP — ordinary networking. Ubiquitous but slow (~0.1 GB/s on 1 GbE), and MPI’s auto-selection sometimes silently falls back to it.

  • InfiniBand (IB) — a high-performance interconnect with RDMA (remote direct memory access). Roughly 35× the bandwidth of 1 GbE (measured ~3.9 GB/s here), and the right choice whenever ranks exchange large arrays.

Pick the transport explicitly — leaving it to auto-selection can hang or quietly use slow Ethernet. The three OpenMPI recipes below are the ones that work in practice:

# (1) InfiniBand (UCX) — the fast cross-node path (~35x TCP)
mpirun --mca pml ucx \
  -x UCX_TLS=rc_verbs,ud_verbs,sm,self -x UCX_NET_DEVICES=mlx4_0:1 \
  -x PATH -x LD_LIBRARY_PATH -x MKL_THREADING_LAYER \
  --host nodeA:128,nodeB:128 -np <N> python script.py

# (2) TCP fallback (no IB / IB down)
mpirun --mca pml ob1 --mca btl tcp,sm,self --mca btl_tcp_if_include eno1 \
  -x PATH -x LD_LIBRARY_PATH -x MKL_THREADING_LAYER \
  --host nodeA:128,nodeB:128 -np <N> python script.py

# (3) Shared-memory only (single node; also the safe path when the network is broken)
mpirun --mca pml ob1 --mca btl self,sm -np <N> python script.py

Check the link first: cat /sys/class/infiniband/*/ports/*/state should read ACTIVE.

The caveats that will otherwise hang you#

These are the non-obvious failures a real cluster throws — each documented from hard experience:

  • mpi4py must init at MPI_THREAD_SERIALIZED for the UCX PML. mpi4py defaults to THREAD_MULTIPLE, which this UCX cannot select (“PML UCX could not be selected”). Put import mpi4py; mpi4py.rc.thread_level = "serialized" before from mpi4py import MPI. qc-rs only calls MPI from the rank’s main thread, so serialized is enough.

  • UCX RC needs the ud auxiliary transport. UCX_TLS=rc_verbs,sm,self (no ud_verbs) fails with “no auxiliary transport … Destination is unreachable”. Always include ud_verbs.

  • Forward the environment with -x. Remote ranks do not inherit your login shell, so pass -x PATH -x LD_LIBRARY_PATH -x MKL_THREADING_LAYER (keep MKL_THREADING_LAYER=GNU).

  • /tmp is node-local. Cross-node scripts and scratch must live on a shared filesystem ($HOME / /home1), not /tmp.

  • Bind ranks to NUMA nodes when packing several ranks per host: --map-by numa:PE=<threads> --bind-to core. The default --bind-to none thrashes and looks like a hang.

  • InfiniBand needs a high memlock ulimit (RDMA pins memory). Batch jobs often inherit a low cap (ibv_reg_mr ... Cannot allocate memory); the admin fix raises LimitMEMLOCK on the compute nodes.

qc-rs supports OpenMPI, MPICH, and MVAPICH (all can drive InfiniBand via UCX); rebuild mpi4py against whichever you use. The exhaustive per-cluster recipes live in the README’s HPC appendix.

Under the hood: distributed memory done right#

qc-rs does not naively broadcast whole matrices. Its distributed backends build on a one-sided RMA (remote direct memory access) layer: each rank fills its own slab in place and serves it to others with no copy, streams one remote block at a time instead of reconstructing a whole factor, and reduces the small aux intermediate rather than the big tensor. The payoff is that per-rank memory scales as total / nranks — which is how RI-JK and RI-MP2 break the “one node’s RAM” wall. You do not manage any of this; it is why the same eri="ri-ram" calculation that needs one big node can instead spread across several small ones.

Exercise 18

  1. Your 2-node MPI job runs, but no faster than a single node, and top shows the network barely used. Name two likely causes from this chapter.

  2. Why can an MPI run tackle a molecule whose integrals do not fit in one machine’s RAM, while adding threads cannot?

  3. A cluster job hangs at start with no error. You suspect the interconnect. What is the first single-line check, and what is the safest transport to fall back to?

CPUs — threaded and distributed — are one road to speed. The other is the GPU.