.. _distributed_memory:

==================
Distributed Memory
==================

.. contents:: In this section:
   :local:
   :depth: 2

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
------------

.. code-block:: c
   :caption: 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
---------

.. code-block:: bash

   # 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.

.. code-block:: bash
   :caption: 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
-----------------------------------------------

.. code-block:: bash

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

Option 2: Specify processes per node
------------------------------------

.. code-block:: bash

   #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
-------------------------------------------

.. code-block:: bash

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

Communication Between Processes
===============================

Example: Distributed Sum
------------------------

.. code-block:: c
   :caption: 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

.. code-block:: bash
   :caption: 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:

.. code-block:: text

   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
-------------------

.. code-block:: bash
   :caption: 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

   .. code-block:: c

      // 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

   .. code-block:: c

      // 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**
   
   .. code-block:: c

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

3. **Use communicators to organize groups**
   
   .. code-block:: c

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

4. **Avoid very small messages**
   
   - Latency overhead dominates
   - Group data into larger buffers

.. seealso::

   - `MPI <https://en.wikipedia.org/wiki/Message_Passing_Interface>`_ - Message Passing Interface
   - :ref:`shared_memory` - For parallelism within the node
   - :ref:`running_simulations` - Basic submission concepts
   - :ref:`optimizing_performance` - How to find the ideal configuration
   - :ref:`storage_guide` - /store/ for multi-node jobs