Obsidian + 🪴 Quartz 4.0

HPC - Lecture 6 - Message Passing

8 min read

HPC - Lecture 6 - Message Passing

Cluster Model and Distributed Memory

A cluster is a set of computing nodes, each equipped with its own local memory, interconnected via a network or a switch. This architecture represents a distributed-memory MIMD (Multiple Instruction, Multiple Data) system.

  • Memory Access: The processing units (CPUs) within a node can access their own local memory but cannot directly access the memories of remote nodes.
  • Node Composition: Multi-core CPUs inside each node implement shared-memory MIMD parallelism locally, while the cluster as a whole operates on distributed memory.
  • Historical Context: The first clusters emerged in the late 1980s and early 1990s as a cost-effective alternative to monolithic supercomputers.

The Message-Passing Model

In this model, a program running on a single node is defined as a process. A parallel program is composed of a set of processes that cooperate by exchanging data through an explicit communication network.

  • Address Space: Each process can only access data in its own private address space.
  • Explicit Data Transfer: If a process requires data owned by another, it must perform an explicit transfer.
  • Process Identification: Each process is uniquely identified by an integer (rank), typically ranging from to , where is the total number of processes.
  • SPMD (Single Program Multiple Data): The standard programming approach where the same executable runs concurrently across all processes, but operates on different data based on the process ID.

Basic Communication Operations

To enable data transfer, programming languages are extended with two fundamental operations:

  1. send(data, dest): Dispatches data to the process with identifier dest.
  2. recv(data, source): Receives data from the process with identifier source.

Global Summation Strategies

A common problem in distributed systems is computing a global sum from partial results stored in each process.

Strategy 0: Centralized Collection

Each process sends its partial result to a master process (Process 0), which performs all additions.

  • Steps: We need communication steps to complete.
  • Observation: Only Process 0 owns the final result.

Centralize Collection Algorithm:

if (myid=0) then
	Stot = S
	for i =1 to NP-1
		recv(remres,i)   // proc 0 receives into remres
		Stot = Stot + remres
	endfor
else
	send(S,0)    // other processes send
endif

Strategy 1: Ring Sum

Processes are arranged in a logical ring. Each process sends a value to the next and receives a value from the previous in every step.

  • Steps: communication steps.
  • Observation: After completion, all processes own the total result.

Ring Sum Algorithm

Stot = S;
token = S;
for (i = 1; i < NP; i++) {
    send(token, (myid + 1) % NP);
    recv(token, (myid + NP - 1) % NP);
    Stot = Stot + token;
}
  • myid process identifier
  • NP number of processes
  • S partial result of the process
  • Stot total result
  • token value received and passed by each process

Strategy 2: Cascade Sum

Each process has a partner with which it exchanges data. Processes exchange data with partners at increasing distances (1, 2, 4, …).

  • Steps: communication steps.
  • Observation: All processes own the total result. Highly efficient for large clusters.

SPMD Algorithm for Cascade Sum:

q = myid
dist = 1
Stot = S
 
for i=1 to log(NP) //number of steps
	r = q%2   //remainder
	if (r == 0) then
		send(Stot, myid+dist)
		recv(token, myid+dist)
	else      // if odd
		send(Stot, myid-dist)
		recv(token, myid-dist)
	endif
	Stot = Stot + token //preparenext iteration
	q = q/2
	dist = dist * 2
  • q is the quotient of division by 2 and r is the remainder of division by 2
  • dist is the distance between partner processes.

Performance Considerations

The communication phases in these algorithms do not exploit available parallelism and are often the primary cause of the decay in Speed-up and Efficiency.

Amdahl’s Law

The maximum speed-up is limited by the sequential fraction of the program :

Generalized Amdahl’s Law

For multi-level parallelism:

MPI: Message Passing Interface

MPI is a standard that defines function interfaces to ensure portability across different distributed systems. It is a library, not a new language, primarily used with C, C++, and Fortran.

There are several implementations e.g.:

  • Open Source like MPICH and OpenMPI
  • Proprietary like Mellanox ScalableMPI and Intel MPI

Developing applications with MPI

The system runs multiple copies of the same executable program (SPMD model). Multiple processes can also run on the same cluster node. In any case, each process has its own address space.

  • Context: A group of processes that can communicate with each other.
  • Communicator: A data structure (MPI_Comm) associated with a context.
  • Default Communicator: MPI_COMM_WORLD includes all processes launched at the start.

Within each context, processes are uniquely identified by an integer (rank), which may be different in each context to which they belong.

If is the number of processes belonging to a context, the rank is an integer between and .

Fundamental MPI Functions

All MPI programs must be structured with the following initialization and finalization calls:

Initialization and Termination:

  • int MPI_Init(int *argc, char **argv): Initializes the MPI environment.
  • int MPI_Finalize(void): Terminates the MPI environment.

Both return 0 if OK, otherwise an error code. These functions create and destroy the default communication environment, in particular:

  • the communicator
  • the number of processes and the rank
  • the communication data structures

Environment Inquiry:

  • int MPI_Comm_size(MPI_Comm comm, int *size): Returns the total number of processes in the communicator.
  • int MPI_Comm_rank(MPI_Comm comm, int *rank): Returns the rank of the calling process.

Point-to-Point Communication

In an MPI program, cooperation among processes takes place through explicit communication operations. The most elementary communication operation consists of:

  • a sender process (send)
  • a receiver process (recv)

MPI_Send

int MPI_Send(void *data, int count, MPI_Datatype datatype, int dest, int tag, MPI_Comm comm)
  • data: Pointer to the buffer.
  • count: Number of elements.
  • datatype: e.g., MPI_INT, MPI_FLOAT, MPI_DOUBLE, MPI_CHAR.
  • dest: Rank of destination.
  • tag: Message identifier to distinguish different types of messages.

MPI_Recv

int MPI_Recv(void *data, int count, MPI_Datatype datatype, int source, int tag, MPI_Comm comm, MPI_Status *status)
  • source: Rank of sender.
  • status: Contains info like the actual number of items received.

Practical Implementation

Compiling and Running

MPI programs are compiled using mpicc (a wrapper for gcc) and executed using mpiexec or mpirun.

  • Compiling: mpicc source.c -o executable
  • Running: mpiexec -np <num_proc> ./executable

First example: hello world

#include <stdio.h>
#include <mpi.h>
 
int main(int argc, char *argv[]) {
	int nproc, myid;
	
	MPI_Init(&argc, &argv)
	
	MPI_Comm_size(MPI_COMM_WORLD, &nproc);
	MPI_Comm_rank(MPI_COMM_WORLD, &myid);
	
	printf("Hello from %d of %d processes\n", myid, nproc);
	
	MPI_Finalize();
}

Second Example: Only two processes

#include <stdio.h> 
#include <mpi.h> 
 
int main (int argc, char *argv[ ]) {
 int nproc, myid; MPI_Status status; /* MPI type defined in mpi.h */ 
 float a[2]; /* array to send */ 
 
 MPI_Init(&argc, &argv); 
 MPI_Comm_size(MPI_COMM_WORLD, &nproc); 
 MPI_Comm_rank(MPI_COMM_WORLD, &myid); 
 
 if (myid == 0) { 
	 a[0] = 1; 
	 a[1] = 2; 
	 MPI_Send(a, 2, MPI_FLOAT, 1, 10, MPI_COMM_WORLD); 
 } else { 
	 MPI_Recv(a, 2, MPI_FLOAT, 0, 10, MPI_COMM_WORLD, &status);
	  printf("%d: a[0]=%f a[1]=%f\n", myid, a[0], a[1]); } 
	  MPI_Finalize(); 
  
}

Error codes

All MPI functions are of type int and return an error code to check whether the instruction was executed correctly.

6 basic functions

The MPI standard defines more than 100 functions, but with these 6 basic functions, it is possible to write most applications for distributed-memory MIMD systems:

  • MPI_INIT
  • MPI_FINALIZE
  • MPI_Comm_rank
  • MPI_Comm_size
  • MPI_Send
  • MPI_Recv

All MPI type and function declarations are in the file mpi.h.

Slurm Batch Script Example

On high-performance clusters (e.g., PurpleJeans), jobs are submitted via batch scripts:

#!/bin/bash
#SBATCH --job-name=hello
#SBATCH --nodes=1
#SBATCH --ntasks-per-node=4
#SBATCH --partition=xhicpu
 
ulimit -s 10240 
module load ompi-4.1.0-gcc-8.3.1 
mpicc -o hello hello.c 
mpirun ./hello

See PurpleJeans Intro.

Exercise: Global Sum Implementation

Task: Write an MPI function int sum(int *A, int N) that computes the global sum using the Ring Sum strategy.

Initialization Logic: If and , each process initializes its local array A as: The function must ensure that at the end, every process prints the same global sum .

Graph View