Mpi

MPI Agent Skill

Agent-optimized skill for MPI (Message Passing Interface) parallel computing.

Quick Reference

mpi4py - Python Bindings

from mpi4py import MPI

# Initialize
comm = MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

# Point-to-point (Python objects)
if rank == 0:
    data = {'key': 'value', 'number': 42}
    comm.send(data, dest=1, tag=11)
elif rank == 1:
    data = comm.recv(source=0, tag=11)

# NumPy arrays (fast)
import numpy as np
if rank == 0:
    data = np.arange(100, dtype='float64')
    comm.Send(data, dest=1, tag=13)
elif rank == 1:
    data = np.empty(100, dtype='float64')
    comm.Recv(data, source=0, tag=13)

# Collective operations
data = np.arange(10, dtype='float64') * rank
total = np.empty(10, dtype='float64')
comm.Allreduce(data, total, op=MPI.SUM)

# Broadcast
if rank == 0:
    data = {'a': 1, 'b': 2}
else:
    data = None
data = comm.bcast(data, root=0)

Launching MPI Programs

# Basic execution
mpirun -n 4 python script.py
mpiexec -np 8 ./mpi_program

# Hostfile
mpirun -np 16 --hostfile hosts.txt python script.py

# Process mapping
mpirun -np 8 --map-by node python script.py  # 1 per node
mpirun -np 8 --map-by core python script.py  # 1 per core
mpirun -np 8 --bind-to core python script.py  # Bind to cores

# SLURM integration
srun -n 16 python mpi_script.py

Common Patterns

Point-to-Point Communication

from mpi4py import MPI
import numpy as np

comm = MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

# Send/Recv (blocking)
if rank == 0:
    data = np.random.randn(100)
    comm.Send(data, dest=1, tag=0)
elif rank == 1:
    data = np.empty(100)
    comm.Recv(data, source=0, tag=0)

# Isend/Irecv (non-blocking)
if rank == 0:
    data = np.random.randn(100)
    req = comm.Isend(data, dest=1, tag=0)
    req.wait()
elif rank == 1:
    data = np.empty(100)
    req = comm.Irecv(data, source=0, tag=0)
    req.wait()

# Sendrecv (exchange)
if rank == 0:
    send_data = np.ones(10)
    recv_data = np.empty(10)
    comm.Sendrecv(send_data, dest=1, recvbuf=recv_data, source=1)

Collective Operations

# Broadcast
if rank == 0:
    data = np.arange(100, dtype='float64')
else:
    data = np.empty(100, dtype='float64')
comm.Bcast(data, root=0)

# Scatter
if rank == 0:
    data = np.arange(100, dtype='float64')
else:
    data = None
local_data = np.empty(100 // size, dtype='float64')
comm.Scatter(data, local_data, root=0)

# Gather
local_data = np.arange(10, dtype='float64') * rank
if rank == 0:
    global_data = np.empty(10 * size, dtype='float64')
else:
    global_data = None
comm.Gather(local_data, global_data, root=0)

# Allgather
local_data = np.arange(10, dtype='float64') * rank
global_data = np.empty(10 * size, dtype='float64')
comm.Allgather(local_data, global_data)

# Reduce
local_sum = np.array([rank], dtype='int')
global_sum = np.empty(1, dtype='int')
comm.Reduce(local_sum, global_sum, op=MPI.SUM, root=0)

# Allreduce
local_data = np.ones(10) * rank
global_data = np.empty(10)
comm.Allreduce(local_data, global_data, op=MPI.SUM)

Reduction Operations

# Available operations
MPI.SUM      # Sum
MPI.PROD     # Product
MPI.MAX      # Maximum
MPI.MIN      # Minimum
MPI.LAND     # Logical AND
MPI.LOR      # Logical OR
MPI.BAND     # Bitwise AND
MPI.BOR      # Bitwise OR
MPI.MAXLOC   # Max and location
MPI.MINLOC   # Min and location

# Custom reduction
def custom_op(x, y, datatype):
    return x * 2 + y

op = MPI.Op.Create(custom_op, commute=True)
result = comm.reduce(data, op=op, root=0)
op.Free()

Process Topology

Cartesian Topology

# Create 2D grid
dims = MPI.Compute_dims(size, [0, 0])  # Auto-determine dims
cart_comm = comm.Create_cart(dims, periods=[False, False])

# Get coordinates
coords = cart_comm.Get_coords(rank)

# Get neighbors
left, right = cart_comm.Shift(0, 1)
up, down = cart_comm.Shift(1, 1)

# Exchange with neighbors
send_data = np.ones(10) * rank
recv_data = np.empty(10)
cart_comm.Sendrecv(send_data, dest=right, recvbuf=recv_data, source=left)

Communicator Splitting

# Split by color
color = rank // 2
new_comm = comm.Split(color, key=rank)

# Split by type
new_comm = comm.Split_type(MPI.COMM_TYPE_SHARED)

Parallel I/O

MPI-IO

from mpi4py import MPI
import numpy as np

comm = MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

# Write parallel file
amode = MPI.MODE_WRONLY | MPI.MODE_CREATE
fh = MPI.File.Open(comm, "output.dat", amode)

# Each rank writes its chunk
offset = rank * 100 * 8  # 100 doubles per rank
data = np.arange(100, dtype='float64') * rank
fh.Write_at(offset, data)
fh.Close()

# Read parallel file
amode = MPI.MODE_RDONLY
fh = MPI.File.Open(comm, "output.dat", amode)
data = np.empty(100, dtype='float64')
fh.Read_at(offset, data)
fh.Close()

Collective I/O

# Collective write
fh = MPI.File.Open(comm, "output.dat", MPI.MODE_WRONLY | MPI.MODE_CREATE)

# Define file view
offset = rank * 100 * 8
fh.Set_view(offset, MPI.DOUBLE)

# Collective write
data = np.arange(100, dtype='float64') * rank
fh.Write_all(data)
fh.Close()

Performance Patterns

Non-Blocking Communication

# Overlap communication and computation
requests = []

# Post receives
for src in range(size):
    if src != rank:
        buf = np.empty(100, dtype='float64')
        req = comm.Irecv(buf, source=src, tag=0)
        requests.append((req, buf))

# Post sends
send_data = np.ones(100, dtype='float64') * rank
for dest in range(size):
    if dest != rank:
        req = comm.Isend(send_data, dest=dest, tag=0)
        requests.append((req, None))

# Do computation while communication happens
local_result = expensive_computation()

# Wait for communication
for req, buf in requests:
    req.wait()

Buffering

# Buffered send (immediate return)
data = np.random.randn(1000)
comm.Bsend(data, dest=1, tag=0)

# Attach buffer for Bsend
buffer_size = 10000 * MPI.DOUBLE.size + MPI.BSEND_OVERHEAD
buffer = bytearray(buffer_size)
MPI.Attach_buffer(buffer)

# Use Bsend...

# Detach buffer
MPI.Detach_buffer()

GPU-Aware MPI

CUDA Integration

# With cupy
import cupy as cp
from mpi4py import MPI

comm = MPI.COMM_WORLD
rank = comm.Get_rank()

# GPU array
gpu_data = cp.arange(100, dtype='float64') * rank

# Direct GPU-to-GPU transfer (if MPI is CUDA-aware)
if rank == 0:
    comm.Send(gpu_data, dest=1, tag=0)
elif rank == 1:
    recv_data = cp.empty(100, dtype='float64')
    comm.Recv(recv_data, source=0, tag=0)

# Collective with GPU arrays
result = cp.empty(100, dtype='float64')
comm.Allreduce(gpu_data, result, op=MPI.SUM)

Error Handling

# Check MPI errors
try:
    comm.send(data, dest=invalid_rank)
except MPI.Exception as e:
    print(f"MPI Error: {e}")

# Error handler
def my_error_handler(comm, error_code):
    print(f"MPI Error on rank {comm.Get_rank()}: {error_code}")

comm.Set_errhandler(MPI.ERRORS_RETURN)

Timing and Profiling

# MPI timer (wall clock)
start = MPI.Wtime()

# Work...
result = computation()

end = MPI.Wtime()
elapsed = end - start

# Global timing
local_time = elapsed
max_time = comm.reduce(local_time, op=MPI.MAX, root=0)

if rank == 0:
    print(f"Max time: {max_time} seconds")

Integration Patterns

NumPy Integration

# Automatic datatype detection
data = np.array([1, 2, 3], dtype=np.int32)
comm.Send(data, dest=1)  # MPI.INT automatically detected

# Manual datatype
comm.Send([data, MPI.INT], dest=1)

SLURM Integration

#!/bin/bash
#SBATCH -N 4
#SBATCH --ntasks-per-node=8
#SBATCH --time=01:00:00

# mpirun auto-detects SLURM
mpirun python mpi_script.py

# Or use srun directly
srun python mpi_script.py

Hybrid MPI + OpenMP

from mpi4py import MPI
import os

# Set OpenMP threads
os.environ['OMP_NUM_THREADS'] = '4'

comm = MPI.COMM_WORLD
rank = comm.Get_rank()

# MPI between nodes, OpenMP within nodes
# Use MPI.COMM_TYPE_SHARED for node-local communication

Debugging

# Verbose output
mpirun -n 4 --verbose python script.py

# Display map
mpirun -n 4 --display-map python script.py

# Attach debugger
mpirun -n 4 --debug python script.py

# Valgrind
mpirun -n 2 valgrind --leak-check=full python script.py

References

Detailed documentation in references/:

• mpi-standard.md: MPI standard, concepts, and specifications
• point-to-point.md: Send/Recv operations, tags, buffering
• collective-operations.md: Broadcast, scatter, gather, reduce
• parallel-io.md: MPI-IO, collective I/O, file views
• execution-deployment.md: mpirun, hostfiles, SLURM, process mapping

Official Documentation

• MPI Forum Standard
• Open MPI Documentation
• MPICH Documentation
• mpi4py Documentation

Best Practices

1. Use collective operations: More efficient than point-to-point loops
2. Non-blocking communication: Overlap communication and computation
3. Match send/recv counts: Avoid deadlocks
4. Use MPI-IO: Parallel file access
5. Profile communication: Identify bottlenecks
6. Uppercase for NumPy: Send/Recv for arrays
7. Check buffer sizes: Prevent overflows
8. Use appropriate communicators: Split for subsystems
9. GPU-aware MPI: Direct GPU transfers when available
10. Test at small scale: Debug before large runs

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: dtunai
• Source: dtunai/agent-skills-for-compute
• License: MIT

Install and usage instructions live in the source repository linked above.