Distributed Memory

Distributed memory is a parallelism model in which multiple processes run on different nodes (servers), communicating via message passing.

Basic Concepts

  • Processes: Independent execution units, each with its own memory

  • MPI (Message Passing Interface): Standard for communication between processes

  • Message passing: Processes communicate explicitly by sending/receiving data

  • Scalability: Can use hundreds or thousands of nodes

Characteristics:

  • Processes can be on different nodes

  • Each process has its own memory space

  • Communication is explicit (more complex)

  • Almost unlimited scalability

MPI

MPI is the most widely used standard for distributed-memory programming. On GridUnesp, we have two implementations:

  • Intel MPI (recommended, optimized for Intel hardware)

  • OpenMPI (open-source, widely used)

Example in C

mpi_hello.c
#include <mpi.h>
#include <stdio.h>
#include <string.h>

int main(int argc, char *argv[])
{
  int numtasks, rank, len, rc;
  char hostname[MPI_MAX_PROCESSOR_NAME];

  // Initialize MPI
  MPI_Init(&argc, &argv);

  // Total number of processes
  MPI_Comm_size(MPI_COMM_WORLD, &numtasks);

  // Rank (ID) of this process
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);

  // Name of the node where it is running
  MPI_Get_processor_name(hostname, &len);

  printf("Process %d of %d running on %s\n",
         rank, numtasks, hostname);

  // Finalize MPI
  MPI_Finalize();

  return 0;
}

Compiling

# With Intel MPI
module load intel/compilers
module load intel/mpi
mpiicc mpi_hello.c -o mpi_hello

# With OpenMPI
module load openmpi/4.1.5
mpicc mpi_hello.c -o mpi_hello

Submission Script

In SLURM, the number of processes is controlled by the -n (--ntasks) directive.

job_mpi.sh
#!/bin/bash
#SBATCH -J mpi_test
#SBATCH -N 4                # 4 nodes
#SBATCH -n 112              # 112 processes total
#SBATCH --ntasks-per-node=28 # 28 processes per node
#SBATCH -t 02:00:00
#SBATCH --mem-per-cpu=4G

export INPUT="input_data/ mpi_hello"
export OUTPUT="mpi_results/"

# Load the MPI module
module load openmpi/4.1.5

# Run with srun (recommended for SLURM)
job-nanny mpirun -n 112 ./mpi_hello

Process Distribution

Option 1: Specify the total number of processes

#SBATCH -n 112   # 112 processes total
# SLURM decides how to distribute among the nodes

Option 2: Specify processes per node

#SBATCH -N 4
#SBATCH --ntasks-per-node=28   # 28 processes on each node
# Total: 4 × 28 = 112 processes

Option 3: Specify minimum and maximum nodes

#SBATCH -n 112
#SBATCH -N 2-4   # Minimum 2, maximum 4 nodes
# SLURM decides the optimal distribution

Communication Between Processes

Example: Distributed Sum

mpi_sum.c
#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>

#define ARRAY_SIZE 1000000

int main(int argc, char *argv[]) {
    int rank, size;
    int *data = NULL;
    int local_sum = 0, global_sum = 0;

    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    // Process 0 initializes the data
    if (rank == 0) {
        data = (int*)malloc(ARRAY_SIZE * sizeof(int));
        for (int i = 0; i < ARRAY_SIZE; i++)
            data[i] = i + 1;
    }

    // Compute how many elements each process receives
    int chunk_size = ARRAY_SIZE / size;
    int *local_data = (int*)malloc(chunk_size * sizeof(int));

    // Distribute the data (MPI_Scatter)
    MPI_Scatter(data, chunk_size, MPI_INT,
                local_data, chunk_size, MPI_INT,
                0, MPI_COMM_WORLD);

    // Each process computes its local sum
    for (int i = 0; i < chunk_size; i++)
        local_sum += local_data[i];

    // Gather the partial sums (MPI_Reduce)
    MPI_Reduce(&local_sum, &global_sum, 1, MPI_INT,
               MPI_SUM, 0, MPI_COMM_WORLD);

    if (rank == 0) {
        printf("Total sum: %d\n", global_sum);
        free(data);
    }

    free(local_data);
    MPI_Finalize();
    return 0;
}

Hybrid: MPI + OpenMP

It is possible to combine the two models:

  • MPI for communication between nodes

  • OpenMP for parallelism within each node

job_hybrid.sh
#!/bin/bash
#SBATCH -J hybrid
#SBATCH -N 4                # 4 nodes
#SBATCH --ntasks-per-node=4  # 4 MPI processes per node
#SBATCH -c 7                 # 7 threads per MPI process
#SBATCH -t 24:00:00
#SBATCH --mem=64G

export OMP_NUM_THREADS=$SLURM_CPUS_PER_TASK
export INPUT="data.dat hybrid_program"
export OUTPUT="results/"

module load openmpi/4.1.5
job-nanny srun ./hybrid_program

Total threads: 4 nodes × 4 processes × 7 threads = 112 threads

When to Use Distributed Memory

✅ Ideal for:

  • Problems that do not fit on a single node

  • Applications that scale well with more nodes

  • Simulations that require many resources

  • When the total memory needed > 128 GB

❌ Not recommended for:

  • Small problems that fit on one node

  • Applications with a lot of communication (overhead)

  • When network latency is critical

Scalability

Amdahl’s Law

The maximum speedup is limited by the serial portion of the code:

Speedup = 1 / (S + (1-S)/P)

Where:

  • S = serial fraction

  • P = number of processes

Example: If 10% of the code is serial, the maximum theoretical speedup is 10x, even with infinite processes.

Testing Scalability

test_mpi_scaling.sh
#!/bin/bash
# Strong scaling test (fixed problem)
for nodes in 1 2 4 8; do
    cat > job_${nodes}n.sh << EOF
#!/bin/bash
#SBATCH -J mpi_${nodes}n
#SBATCH -N ${nodes}
#SBATCH --ntasks-per-node=28
#SBATCH -t 01:00:00
#SBATCH --mem-per-cpu=2G

module load openmpi/4.1.5
srun ./mpi_program
EOF
    sbatch job_${nodes}n.sh
done

Common Problems

  1. Deadlock: Processes waiting for each other indefinitely

    // WRONG: can cause deadlock
MPI_Send(...);  // Blocking send
MPI_Recv(...);  // Receive

// CORRECT: use combined send/receive
MPI_Sendrecv(...);

  2. Race condition: Concurrent access without synchronization

    // Use atomic or collective operations
MPI_Reduce(&local, &global, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);

  3. Granularity too fine: Communication overhead dominates

    • Group small messages into larger messages

    • Use collective communication when possible

Optimization Tips

  1. Collective vs point-to-point communication

    • Use MPI_Reduce, MPI_Bcast, MPI_Scatter when appropriate

    • They are optimized for the hardware

  2. Overlap communication/computation

    MPI_Isend(...);  // Non-blocking
// Computation while the message is being sent
MPI_Wait(...);   // Wait for completion

  3. Use communicators to organize groups

    MPI_Comm_split(MPI_COMM_WORLD, color, key, &new_comm);

  4. Avoid very small messages

    • Latency overhead dominates

    • Group data into larger buffers

See also