Designing a parallel program means turning one large computation into smaller units of work that can run at the same time. A good design does more than create many tasks: it keeps processors busy, limits communication, preserves data locality, and maps work onto hardware in a way that reduces waiting.
A useful design framework is PCAM:
+-----------------------------+
| Big Problem |
+--------------+--------------+
|
v
+-----------------------------+
| 1. Partitioning |
| Break the problem into |
| smaller parallel tasks |
+--------------+--------------+
|
v
+-----------------------------+
| 2. Communication |
| Identify what data must be |
| exchanged between tasks |
+--------------+--------------+
|
v
+-----------------------------+
| 3. Agglomeration |
| Combine tasks where useful |
| to reduce overhead |
+--------------+--------------+
|
v
+-----------------------------+
| 4. Mapping |
| Assign tasks to processors |
| or hardware resources |
+-----------------------------+Partitioning divides the problem into smaller pieces that can run concurrently. The goal is to expose enough parallel work while avoiding unnecessary dependencies between tasks.
+-----------------------+
| Big Problem |
+-----------+-----------+
|
+---------+---------+
| | |
+----+----+ +--+--+ +----+----+
| Part 1 | |Part2| | Part 3 |
+---------+ +-----+ +---------+I. Balance the workload among tasks
II. Minimize dependencies between tasks
III. Expose enough concurrency
Domain decomposition splits the data space into chunks. Each worker receives a portion of the data and performs the same or similar computation on that portion.
This works well when computation is mostly local, such as image processing, simulations, matrix operations, and grid-based numerical methods.
Example: blurring a 4000 × 4000 image
+-----------------------+-----------------------+
| Tile 1 | Tile 2 |
| Worker 1 | Worker 2 |
+-----------------------+-----------------------+
| Tile 3 | Tile 4 |
| Worker 3 | Worker 4 |
+-----------------------+-----------------------+Why it works: most pixels depend only on nearby pixels, so most computation can run independently. Only the tile boundaries require communication.
Functional decomposition splits the work by type of operation rather than by data region. It is often used to build pipelines.
Example: web request pipeline
Request
|
v
+--------+ +-------+ +-----------+ +--------+
| Auth | --> | Data | --> | Transform | --> | Render |
+--------+ +-------+ +-----------+ +--------+Each stage can scale independently. For example, if database access becomes the bottleneck, you can allocate more workers to the Data stage.
Functional decomposition is useful when:
In parallel programming, cyclic decomposition usually means distributing tasks or loop iterations in a round-robin pattern across workers.
Instead of giving each worker one large contiguous block, cyclic decomposition assigns item i to worker i mod P, where P is the number of workers.
Example: 12 loop iterations on 4 workers
| Iteration | 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 |
| Worker | 0 | 1 | 2 | 3 | 0 | 1 | 2 | 3 | 0 | 1 | 2 | 3 |
Worker 0: iterations 0, 4, 8
Worker 1: iterations 1, 5, 9
Worker 2: iterations 2, 6, 10
Worker 3: iterations 3, 7, 11When cyclic decomposition helps:
Tradeoff: cyclic decomposition often improves load balance, but it can reduce locality because each worker touches data spread across the whole domain.
Block decomposition divides arrays or matrices into contiguous blocks. Each worker receives one or more blocks and operates on them as units.
This is common in matrix algorithms, stencil computations, and cache-aware numerical methods.
Let
$$ A = \begin{pmatrix} 1 & 2 & 9 & 8 \\ 0 & 3 & 7 & 6 \\ 4 & 5 & 0 & 0 \\ 4 & 5 & 0 & 0 \end{pmatrix} = \begin{pmatrix} A_{11} & A_{12} \\ A_{21} & A_{22} \end{pmatrix} $$
where
$$ A_{11}=\begin{pmatrix}1 & 2 \ 0 & 3\end{pmatrix}, \quad A_{12}=\begin{pmatrix}9 & 8 \ 7 & 6\end{pmatrix}, \quad A_{21}=\begin{pmatrix}4 & 5 \ 4 & 5\end{pmatrix}, \quad A_{22}=\begin{pmatrix}0 & 0 \ 0 & 0\end{pmatrix}. $$
Why blocks help:
C = A B, then $C_{11}=A_{11}B_{11}+A_{12}B_{21}$ and similarly for the other blocks. This lets the program reuse optimized kernels on smaller pieces.Related strategy: block-cyclic decomposition
Block-cyclic decomposition combines block and cyclic ideas. It assigns blocks round-robin across workers, which helps balance load while still preserving some locality.
Communication is the exchange of data between tasks. It is necessary when one task needs another task’s results. The best parallel designs minimize communication without sacrificing correctness.
I. Minimize communication volume and frequency
II. Preserve locality
III. Overlap communication with computation
Local communication occurs between tasks on the same processor, socket, or node. It usually uses shared memory and is faster than network communication.
Example: two cores on the same CPU socket share data through memory.
+-----------+ +-----------+
| Core 1 | <-> | Core 2 |
| Task A | | Task B |
+-----------+ +-----------+
\ /
\ /
+-------------+
| Shared Mem |
+-------------+Global communication occurs between tasks on different nodes or physical machines. It usually travels over a network and is more expensive than local communication.
Example: MPI ranks running on separate servers exchange boundary data.
+-------------+ Network +-------------+
| Node 1 | <-------------------> | Node 2 |
| Task A | | Task B |
+-------------+ +-------------+Point-to-point communication sends data directly from one task to another.
+---------+ message +----------+
| Sender | --------------> | Receiver |
+---------+ +----------+Common examples include:
Send / RecvIsend / IrecvCollective communication involves a group of tasks. Common collective operations include broadcast, scatter, gather, reduce, and all-reduce.
Scatter: one task sends different pieces of data to different tasks.
Scatter:
+-----------------------+
| ABC |
+-----------+-----------+
|
+------------+------------+
| | |
+---v---+ +---v---+ +---v---+
| A | | B | | C |
+-------+ +-------+ +-------+Broadcast: one task sends the same data to all tasks.
Broadcast:
+-----------------------+
| ABC |
+-----------+-----------+
|
+------------+------------+
| | |
+---v---+ +---v---+ +---v---+
| ABC | | ABC | | ABC |
+-------+ +-------+ +-------+Reduce: many tasks contribute values, and one result is produced.
Worker values: 4, 7, 2, 9
Operation: sum
Result: 22All-reduce: every task receives the reduced result. This is common in simulations and machine learning training.
Blocking communication waits until the operation is complete before the task continues.
Nonblocking communication starts an operation and allows the task to continue while the message is in progress.
Example pattern:
Overhead is time spent managing communication rather than doing the main computation. It includes message setup, synchronization, copying, packing, unpacking, and scheduling.
Latency is the time required for a message to begin reaching its destination. It matters most when messages are small and frequent.
Bandwidth is the amount of data that can be transferred per second. It matters most when messages are large.
Contention occurs when multiple tasks compete for the same memory bus, network link, file system, or synchronization point.
Agglomeration combines smaller tasks into larger tasks after the initial partitioning step. The purpose is to reduce overhead, improve locality, and create tasks that better match the target hardware.
I. Improve computational granularity
II. Minimize communication
III. Fit the memory hierarchy
Granularity describes how much computation occurs per communication event.
$$ \text{Granularity} = \frac{\text{Computation}}{\text{Communication}} $$
Fine-grained parallelism uses many small tasks.
Advantages:
Disadvantages:
Coarse-grained parallelism uses fewer, larger tasks.
Advantages:
Disadvantages:
The best granularity depends on the problem, hardware, and runtime system. A good design usually starts with many logical tasks, then agglomerates them until communication and scheduling overhead are acceptable.
Mapping assigns agglomerated tasks to hardware resources such as cores, sockets, nodes, GPUs, or clusters. Good mapping improves load balance and reduces communication distance.
I. Balance computational load across processors
II. Minimize communication overhead
III. Minimize total execution time
Tasks are assigned before execution and remain fixed.
Best for: predictable workloads with uniform task sizes.
Tasks are assigned during execution based on current load.
Best for: irregular workloads where task cost is hard to predict.
Tasks are mapped in levels: for example, first to nodes, then sockets, then cores.
Best for: multi-node systems with NUMA domains, GPUs, or complex network topology.
Tasks that communicate frequently are grouped and placed near each other.
Best for: graph algorithms, stencil computations, and workflows with known communication patterns.
Represent tasks as graph nodes and communication as graph edges. Then partition the graph so that heavily connected tasks stay together and edge cuts are minimized.
Best for: irregular meshes, sparse matrix computations, and graph analytics.
Problem: simulate heat diffusion on a 4096 × 4096 grid for 5,000 time steps using a 5-point stencil in double precision.
The program uses two row-major arrays, A[nx][ny] and B[nx][ny], which alternate roles each step. This is often called a ping-pong buffer pattern.
A typical 5-point stencil update is:
$$ B[i,j] = 0.25 \times (A[i-1,j] + A[i+1,j] + A[i,j-1] + A[i,j+1]) $$
Periodic checkpoints are written to HDF5 files.
Goal: split the domain into equal, independent tasks.
Method: use an 8 × 8 domain decomposition, giving 64 blocks.
T(0,0) through T(7,7) or numbered T0 through T63 in row-major order.+---------------------------------------------------------------+
| 4096 x 4096 Grid |
+---------------------------------------------------------------+
| Split into 8 x 8 = 64 blocks |
| Each block = 512 x 512 cells |
+---------------------------------------------------------------+
Example partition view:
+----+----+----+----+----+----+----+----+
| T0 | T1 | T2 | T3 | T4 | T5 | T6 | T7 |
+----+----+----+----+----+----+----+----+
| T8 | T9 |T10 |T11 |T12 |T13 |T14 |T15 |
+----+----+----+----+----+----+----+----+
|T16 |T17 |T18 |T19 |T20 |T21 |T22 |T23 |
+----+----+----+----+----+----+----+----+
| ... continues until T63 ... |
+---------------------------------------+Goal: exchange only the boundary data needed for each stencil step.
Method: use nonblocking point-to-point halo exchanges and occasional collectives.
For each time step:
$$ 512 \times 1 \times 8\text{ bytes} = 4096\text{ bytes} = 4\text{ KB} $$
Allreduce computes a global residual using one double value.Example: one 512 x 512 block
North halo
vvvvvvvvvvvvvvv
+---------------------+
| N N N N N N N N N N |
| |
W | Interior work | E
e | | a
s | | s
t | S S S S S S S S S S | t
+---------------------+
^^^^^^^^^^^^^^^
South halo
Each step:
- Exchange north, south, east, and west halos.
- Compute interior cells while halo messages are in flight.
- Compute border cells after halo data arrives.Recommended overlap pattern:
1. Post Irecv for north/south/east/west halos.
2. Pack and Isend local boundary rows/columns.
3. Compute the interior region that does not need new halos.
4. Wait/Test for halo receives to complete.
5. Unpack halos and compute border cells.Goal: reduce communication overhead and MPI rank count without exceeding memory limits.
Method: merge 2 × 2 blocks into one larger tile and use OpenMP threads inside each MPI rank.
Assume the cluster has:
Before agglomeration:
After agglomeration:
For a 1024 × 1024 tile:
$$ 1024 \times 1 \times 8\text{ bytes} = 8192\text{ bytes} = 8\text{ KB per edge} $$
Four edges produce about 32 KB of halo data per rank per step.
Memory per rank:
A: 1024 × 1024 × 8 bytes ≈ 8 MB.B: 1024 × 1024 × 8 bytes ≈ 8 MB.Before: 64 MPI ranks, each 512 x 512
After: 16 MPI ranks, each 1024 x 1024
Agglomerating 2 x 2 blocks:
+----+----+ +--------+
| A | B | -> | |
+----+----+ | R | One rank owns the merged tile
| C | D | | |
+----+----+ +--------+
Effect:
- Fewer ranks: 64 -> 16
- Fewer messages
- Larger messages: 8 KB per edge instead of 4 KB
- OpenMP threads handle work inside each tileGoal: minimize cross-node traffic and NUMA penalties.
Method: use a Cartesian rank layout and core pinning.
Final configuration:
A simple 4 × 4 Cartesian rank layout is:
R0 R1 R2 R3
R4 R5 R6 R7
R8 R9 R10 R11
R12 R13 R14 R15One possible node placement is:
Node A: R0 R1 R2 R3
Node B: R4 R5 R6 R7
Node C: R8 R9 R10 R11
Node D: R12 R13 R14 R15With this placement:
Recommended OpenMP placement:
export OMP_NUM_THREADS=8
export OMP_PLACES=cores
export OMP_PROC_BIND=closeThis keeps threads close to the rank’s memory allocation and reduces NUMA overhead.
Code: heat2d.c
Expected implementation features:
A and B.MPI_Allreduce every 100 steps.mpicc -O3 -march=native -fopenmp heat2d.c -o heat2d
./heat2d --nx 512 --ny 512 --steps 10 --tile 512 --verify analyticmpirun -n 16 --map-by ppr:4:node --bind-to core --report-bindings \
./heat2d --nx 4096 --ny 4096 --steps 5000 --tile 1024OMP_NUM_THREADS=8 OMP_PLACES=cores OMP_PROC_BIND=close \
mpirun -n 16 --map-by ppr:4:node --bind-to core \
./heat2d --nx 4096 --ny 4096 --steps 5000 --tile 1024 --checkpoint 200chkpt_t<step>_tile<rank>.h5.final_temp.h5.temp_step_<t>.
| Question | Why It Matters |
| Are tasks balanced? | Prevents idle processors. |
| Is there enough parallel work? | Keeps all cores or nodes busy. |
| Are dependencies minimized? | Reduces synchronization and communication. |
| Are messages batched? | Avoids excessive latency overhead. |
| Is communication overlapped with computation? | Hides some communication cost. |
| Are frequently communicating tasks placed close together? | Improves locality and reduces network traffic. |
| Does each task fit memory/cache constraints? | Prevents paging and improves performance. |
| Has the program been profiled? | Confirms bottlenecks instead of guessing. |
perf, HPCToolkit, TAU, and MPI profilers can reveal computation hotspots and communication bottlenecks.OMP_PLACES, OMP_PROC_BIND, and MPI binding options help keep threads and ranks close to their data.
| Mistake | Why It Hurts | Better Approach |
| Creating too many tiny tasks | Scheduling and communication overhead dominate | Agglomerate tasks until granularity is reasonable |
| Ignoring halo costs | Boundary communication becomes the bottleneck | Batch halos and overlap communication with interior work |
| Using blocking communication everywhere | Ranks wait instead of computing | Use nonblocking communication for predictable exchanges |
| Mapping ranks without topology awareness | Increases cross-node and NUMA traffic | Use Cartesian layouts and binding options |
| Assuming equal data size means equal work | Some data regions may be more expensive | Measure work or use dynamic scheduling |
| Optimizing before profiling | Time is spent on the wrong bottleneck | Profile computation, communication, and I/O |