This page looks best with JavaScript enabled

MPI Communication Primitives and Their Use in Python Programming

 ·  ☕ 3 min read

1. What MPI Is

MPI, the Message Passing Interface, is a communication protocol used for parallel computing.

MPI provides a set of standardized interfaces for transferring data between different compute nodes, and is widely used in scientific computing, machine learning, deep learning, and other fields.

MPI has multiple implementations; the common ones are MPICH and OpenMPI. MPICH is led by Argonne National Laboratory and is the basis for various commercially customized versions, so MPICH should not simply be regarded as a single product but as a family of derived versions, such as MVAPICH and Intel MPI. OpenMPI is a version jointly developed by several research institutions (including UTK, IU, Cisco, NVIDIA, and others).

For a concrete choice, refer to:

  • Open MPI

Suited to most Linux clusters, especially scenarios that need high-performance network support (such as InfiniBand, RoCE) and GPU support. Commonly used in HPC and deep learning clusters.

  • MPICH

A general-purpose, highly compatible option for Linux systems, suitable for standard MPI applications. If a system wants to migrate between different MPI implementations, MPICH is the safe choice.

  • MVAPICH

Based on MPICH, optimized specifically for high-performance networks (such as InfiniBand and RDMA), suited to scientific computing tasks and GPU tasks with high network bandwidth demands.

  • Intel MPI

Optimized specifically for Intel hardware, suited to Intel processors and Intel Omni-Path networks, and supports mainstream Linux systems.

2. MPI Communication Primitives

2.1 Point-to-Point, P2P

A way for one process to communicate with another specified process.

  • send
1
2
3
4
5
6
7

MPI_Send(
    void* data,
    int count,
    MPI_Datatype datatype,
    int destination,
    int tag,
    MPI_Comm communicator)

Send a specified amount of data to the specified process.

  • receive
1
2
3
4
5
6
7
8

MPI_Recv(
    void* data,
    int count,
    MPI_Datatype datatype,
    int source,
    int tag,
    MPI_Comm communicator,
    MPI_Status* status)

Receive a specified amount of data from the specified process.

2.2 Collective Communication, CC

A way for one process to communicate with all processes.

  • barrier

Wait for all processes to reach a certain point.

1

MPI_Barrier(MPI_Comm communicator)

After process 0 calls MPI_Barrier at time T1, it must wait for all processes to reach the MPI_Barrier call before they can all continue executing together.

  • broadcast
1
2
3
4
5
6

MPI_Bcast(
    void* data,
    int count,
    MPI_Datatype datatype,
    int root,
    MPI_Comm communicator)

The root process, process 0, sends one copy of the data to all processes.

  • scatter
1
2
3
4
5
6
7
8
9

MPI_Scatter(
    void* send_data,
    int send_count,
    MPI_Datatype send_datatype,
    void* recv_data,
    int recv_count,
    MPI_Datatype recv_datatype,
    int root,
    MPI_Comm communicator)

Unlike broadcast, which sends a complete copy of the data to all processes, scatter splits the data into multiple parts and sends them to different processes.

  • gather
1
2
3
4
5
6
7
8
9

MPI_Gather(
    void* send_data,
    int send_count,
    MPI_Datatype send_datatype,
    void* recv_data,
    int recv_count,
    MPI_Datatype recv_datatype,
    int root,
    MPI_Comm communicator)

The opposite of scatter, gather receives multiple pieces of data into a single process.

  • allgather
1
2
3
4
5
6
7
8

MPI_Allgather(
    void* send_data,
    int send_count,
    MPI_Datatype send_datatype,
    void* recv_data,
    int recv_count,
    MPI_Datatype recv_datatype,
    MPI_Comm communicator)

allgather does not need a root to be specified; all processes receive the data of every other process.

  • reduce
1
2
3
4
5
6
7
8

MPI_Reduce(
    void* send_data,
    void* recv_data,
    int count,
    MPI_Datatype datatype,
    MPI_Op op,
    int root,
    MPI_Comm communicator)

When performing a reduce operation, an op must be specified; this operation is called a reduction. In the figure above, the reduction operation is a sum.

  • allreduce
1
2
3
4
5
6
7

MPI_Allreduce(
    void* send_data,
    void* recv_data,
    int count,
    MPI_Datatype datatype,
    MPI_Op op,
    MPI_Comm communicator)

Unlike reduce, allreduce does not need a root to be specified, and all processes receive the reduced result.

3. Installing MPICH\OpenMPI

To install MPICH, run

1

apt-get install mpich -y

To install OpenMPI, run

1

apt-get install openmpi-bin -y

You will get compilation commands and run commands; whether you install MPICH or OpenMPI, you get a similar set of command-line tools.

3.1 MPI Compilation Commands

  • mpicc

Compile MPI programs with the C compiler, ensuring the compiled program can use the MPI communication library.

  • mpic++, mpiCC, mpicxx

Compile MPI programs with the C++ compiler.

  • mpif77, mpif90, mpifort

Compile MPI programs with different Fortran compilers.

3.2 MPI Run Commands

  • mpirun

Run MPI programs and manage the startup of MPI processes.

  • mpiexec

Similar to mpirun, but more compliant with the MPI standard.

  • mpiexec.xxx, mpirun.xxx

MPI program managers for specific environments and specific tools.

4. MPI Program Example

4.1 Writing an MPI Python Program

1

vim /data/mpi.py

Save the following code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from mpi4py import MPI
import numpy as np
import socket

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

# 获取当前进程所在主机的主机名
hostname = socket.gethostname()

# 每个进程生成一个随机数并包装成 numpy 数组
local_data = np.random.random(1)  # 包装成一个 numpy 数组
print(f"Process {rank} on host {hostname} has local data: {local_data[0]}")

# 使用 gather 收集所有进程的数据
all_data = comm.gather(local_data[0], root=0)

# 创建用于 Allreduce 的缓冲区
global_data = np.zeros(1, dtype='d')  # 初始化全局数据为零，类型为 double
comm.Allreduce(local_data, global_data, op=MPI.SUM)

# 只在 rank 0 上输出结果
if rank == 0:
    print(f"All process data: {all_data}")
    print(f"The sum of all data is: {global_data[0]}")

In some collective communication scenarios, you sometimes want to do something special for one particular process; in that case you use rank to check whether the current process’s number matches the condition and then perform the operation.

4.2 Configuring Passwordless SSH

  • Configure /etc/hosts

Edit the /etc/hosts file and add mappings from hostname to IP address.

1
2
3

x.x.x.x host-1
x.x.x.x host-2
x.x.x.x host-3
  • Generate an SSH key
1

ssh-keygen -t rsa
  • Configure mutual passwordless access between hosts
1
2
3

scp /root/.ssh/id_rsa /root/.ssh/id_rsa.pub host-1:/root/.ssh/
scp /root/.ssh/id_rsa /root/.ssh/id_rsa.pub host-2:/root/.ssh/
scp /root/.ssh/id_rsa /root/.ssh/id_rsa.pub host-3:/root/.ssh/

4.3 Installing Dependencies

  • Install MPI

All hosts should install the same MPI implementation and version.

1

opscli shell -i host-1,host-2,host-3 --content "apt-get install openmpi-bin libopenmpi-dev  -y"

Here you need to install one extra dependency package, libopenmpi-dev, which is used to compile and install mpi4py.

  • Install the mpi4py and numpy dependencies
1

opscli shell -i host-1,host-2,host-3 --content "pip install mpi4py numpy"
  • Copy the mpi.py file to every host
1
2

scp /data/mpi.py host-2:/data/
scp /data/mpi.py host-3:/data/

4.4 Creating the Hostfile

1
2
3
4
5

cat > hostfile <<EOF
host-1 slots=1
host-2 slots=1
host-3 slots=1
EOF

Here slots means the maximum number of processes each host can start. In general, slots is the number of CPU cores.

Besides using a hostfile to specify hosts, you can also use the -host parameter to specify hosts and their slots values.

1
2

-host host-1 -host host-2 -host host-3
-host host-1:1 -host host-2:1

host-1:1 means host-1 can only start one process; the number after the colon is the maximum number of processes.

4.5 Running the MPI Program

  • Multiple hosts, multiple processes
1
2
3
4
5
6
7

mpirun --allow-run-as-root -np 3 -hostfile hostfile python3 /data/mpi.py

Process 2 on host host-3 has local data: 0.8237177424830892
Process 0 on host host-1 has local data: 0.4049368708804896
Process 1 on host host-2 has local data: 0.055557219335660935
All process data: [np.float64(0.4049368708804896), np.float64(0.055557219335660935), np.float64(0.8237177424830892)]
The sum of all data is: 1.2842118326992398
  • Single host, multiple processes
1
2
3
4
5
6
7
8

mpirun --allow-run-as-root -np 4 -host host-2:4 python3 /data/mpi.py

Process 1 on host host-2 has local data: 0.57806331374296
Process 0 on host host-2 has local data: 0.10289882095170111
Process 2 on host host-2 has local data: 0.7618555892517183
Process 3 on host host-2 has local data: 0.810980561742528
All process data: [0.10289882095170111, 0.57806331374296, 0.7618555892517183, 0.810980561742528]
The sum of all data is: 2.2537982856889074
  • mpi assigns processes to each host in order, giving each host up to its slots count

Make the number of available slots far larger than the number of processes and observe MPI’s process allocation strategy.

1
2
3
4
5
6
7
8

mpirun --allow-run-as-root -np 4 -host host-1:2 -host host-2:4 -host host-3:8 python3 /data/mpi.py

Process 3 on host host-2 has local data: 0.2001460877733756
Process 0 on host host-1 has local data: 0.6559694967423054
Process 2 on host host-2 has local data: 0.43367558731572886
Process 1 on host host-1 has local data: 0.8121368124082778
All process data: [np.float64(0.6559694967423054), np.float64(0.8121368124082778), np.float64(0.43367558731572886), np.float64(0.2001460877733756)]
The sum of all data is: 2.101927984239688

After many runs, the result was always that host-1 was assigned 2 processes, host-2 was assigned 2 processes, and host-3 was assigned none. This shows that MPI fills each host in the order the hosts are provided, giving each host as many processes as it can take.

5. References

微信公众号
WRITTEN BY
微信公众号

See Also