Distributed Processing

Introduction

Distributed processing connects multiple independent computers over a network to work on a shared problem. Unlike shared-memory multiprocessors where all cores access the same RAM, distributed systems have private memory at each node and communicate exclusively through message passing. This is how the world's largest computations run — from Google's search index across millions of servers to weather simulations on supercomputers with thousands of nodes. Understanding distributed processing teaches you how to think about computation at internet scale.

Distributed vs Shared Memory

Shared Memory: Distributed Memory: ┌──────┐ ┌──────┐ ┌──────────┐ ┌──────────┐ │Core 1│ │Core 2│ │ Node 1 │ │ Node 2 │ └──┬───┘ └──┬───┘ │ CPU+RAM │ │ CPU+RAM │ │ │ └────┬─────┘ └────┬─────┘ ═══╪════════╪══ Bus │ │ ┌──▼────────▼──┐ └──── Network ──┘ │Shared Memory │ (Ethernet/InfiniBand) └──────────────┘ Communication: Read/write Communication: Send/Receive same address explicit messages

Aspect	Shared Memory	Distributed
Programming	Easier (shared variables)	Harder (explicit messages)
Scalability	Limited (~100 cores)	Unlimited (millions of nodes)
Latency	nanoseconds	microseconds to milliseconds
Fault tolerance	Single point of failure	Individual nodes can fail
Cost scaling	Expensive (custom hardware)	Cheap (commodity hardware)

Message Passing Model

All communication in distributed systems happens through explicit messages:

Example

Node A: Node B: compute_partial_result() send(result, to=B) ──────► receive(data, from=A) process(data) wait_for(result_from_B) ◄────── send(processed, to=A) receive(final, from=B)

MPI (Message Passing Interface)

MPI is the standard library for distributed computing:

Example

// Every node runs this same program with different rank int rank, size; MPI_Init(&argc, &argv); MPI_Comm_rank(MPI_COMM_WORLD, &rank); // My node number MPI_Comm_size(MPI_COMM_WORLD, &size); // Total nodes if (rank == 0) { // Master: distribute work for (int i = 1; i < size; i++) MPI_Send(&work_chunk[i], ..., i, ...); // Collect results for (int i = 1; i < size; i++) MPI_Recv(&results[i], ..., i, ...); } else { // Worker: receive, process, return MPI_Recv(&my_work, ..., 0, ...); result = compute(my_work); MPI_Send(&result, ..., 0, ...); }

Network Topologies

How nodes are interconnected affects performance:

Bus	All nodes share one channel (cheap, slow)
Star	Central switch (single point of failure)
Ring	Each node connected to 2 neighbors
Mesh (2D)	Grid connections (good for nearest-neighbor communication)
Torus	Mesh with wraparound (eliminates edge effects)
Fat Tree	Hierarchical switches (most common in data centers)
Hypercube	2^n nodes, n connections each (low diameter)

Fat Tree (Most Common in Modern Clusters)

Example

Properties: Full bisection bandwidth, any node can communicate with any other at full speed, scalable to thousands of nodes.

Communication Patterns

Point-to-Point

One sender, one receiver. Basic building block.

Broadcast

One sender, all receivers. Used for distributing global data.

Scatter/Gather

Master distributes different pieces to workers (scatter) and collects results (gather).

All-to-All

Every node sends to every other node. Expensive but needed for some algorithms (matrix transpose).

Reduction

Combine values from all nodes into single result (sum, max, min). Can be done in O(log n) steps with tree reduction:

Step 1: Pairs add: (0+1), (2+3), (4+5), (6+7)	4 partial sums
Step 2: Pairs add: (01+23), (45+67)	2 partial sums
Step 3: Final add: (0123+4567)	global sum

Performance Considerations

Communication Overhead

Latency	Time to initiate transfer (1-10 μs for InfiniBand)
Bandwidth	Data rate once transfer starts (100-400 Gbps)
Small messages	Dominated by latency (send many small messages = slow)
Large messages	Dominated by bandwidth (efficient bulk transfer)

Computation-to-Communication Ratio

Example

Efficiency = Compute_time / (Compute_time + Communication_time) If computation = 100ms and communication = 10ms: Efficiency = 100/110 = 91% (good!) If computation = 10ms and communication = 10ms: Efficiency = 10/20 = 50% (poor — half the time communicating)

Rule of thumb: Communication should be < 10% of compute time for good efficiency.

Load Balancing

If one node gets more work than others:

Node 0	████████████████████ 100ms (heavy load)
Node 1	████████ 40ms ← idle waiting for Node 0
Node 2	██████████ 50ms ← idle waiting for Node 0
Node 3	███████ 35ms ← idle waiting for Node 0

Dynamic load balancing: Idle nodes request work from busy nodes.

Real-World Distributed Systems

System	Scale	Application
Google Search	1M+ servers	Web indexing, query processing
AWS	Millions of VMs	Cloud computing (everything)
Bitcoin	10K+ miners	Distributed consensus
CERN LHC	170+ sites	Particle physics data analysis
Folding@home	700K+ devices	Protein structure simulation
MapReduce/Spark	1000s of nodes	Big data processing

Key Takeaways

Distributed systems scale to millions of nodes but require explicit message passing for communication
Network topology and bandwidth determine communication performance — fat trees provide full bisection bandwidth
The computation-to-communication ratio determines efficiency — algorithms must minimize communication relative to computation
Load balancing is critical — total time equals the SLOWEST node's time regardless of how fast others finish
Tree-based reduction allows O(log n) collective operations instead of O(n)
Modern cloud computing, big data, and AI training all rely on distributed processing principles

Introduction

Distributed vs Shared Memory

Aspect	Shared Memory	Distributed
Programming	Easier (shared variables)	Harder (explicit messages)
Scalability	Limited (~100 cores)	Unlimited (millions of nodes)
Latency	nanoseconds	microseconds to milliseconds
Fault tolerance	Single point of failure	Individual nodes can fail
Cost scaling	Expensive (custom hardware)	Cheap (commodity hardware)

Message Passing Model

All communication in distributed systems happens through explicit messages:

Example

MPI (Message Passing Interface)

MPI is the standard library for distributed computing:

Example

Network Topologies

How nodes are interconnected affects performance:

Bus	All nodes share one channel (cheap, slow)
Star	Central switch (single point of failure)
Ring	Each node connected to 2 neighbors
Mesh (2D)	Grid connections (good for nearest-neighbor communication)
Torus	Mesh with wraparound (eliminates edge effects)
Fat Tree	Hierarchical switches (most common in data centers)
Hypercube	2^n nodes, n connections each (low diameter)

Fat Tree (Most Common in Modern Clusters)

Example

Properties: Full bisection bandwidth, any node can communicate with any other at full speed, scalable to thousands of nodes.

Communication Patterns

Point-to-Point

One sender, one receiver. Basic building block.

Broadcast

One sender, all receivers. Used for distributing global data.

Scatter/Gather

Master distributes different pieces to workers (scatter) and collects results (gather).

All-to-All

Every node sends to every other node. Expensive but needed for some algorithms (matrix transpose).

Reduction

Combine values from all nodes into single result (sum, max, min). Can be done in O(log n) steps with tree reduction:

Step 1: Pairs add: (0+1), (2+3), (4+5), (6+7)	4 partial sums
Step 2: Pairs add: (01+23), (45+67)	2 partial sums
Step 3: Final add: (0123+4567)	global sum

Performance Considerations

Communication Overhead

Latency	Time to initiate transfer (1-10 μs for InfiniBand)
Bandwidth	Data rate once transfer starts (100-400 Gbps)
Small messages	Dominated by latency (send many small messages = slow)
Large messages	Dominated by bandwidth (efficient bulk transfer)

Computation-to-Communication Ratio

Example

Rule of thumb: Communication should be < 10% of compute time for good efficiency.

Load Balancing

If one node gets more work than others:

Node 0	████████████████████ 100ms (heavy load)
Node 1	████████ 40ms ← idle waiting for Node 0
Node 2	██████████ 50ms ← idle waiting for Node 0
Node 3	███████ 35ms ← idle waiting for Node 0

Dynamic load balancing: Idle nodes request work from busy nodes.

Real-World Distributed Systems

System	Scale	Application
Google Search	1M+ servers	Web indexing, query processing
AWS	Millions of VMs	Cloud computing (everything)
Bitcoin	10K+ miners	Distributed consensus
CERN LHC	170+ sites	Particle physics data analysis
Folding@home	700K+ devices	Protein structure simulation
MapReduce/Spark	1000s of nodes	Big data processing

Key Takeaways

Distributed systems scale to millions of nodes but require explicit message passing for communication
Network topology and bandwidth determine communication performance — fat trees provide full bisection bandwidth
The computation-to-communication ratio determines efficiency — algorithms must minimize communication relative to computation
Load balancing is critical — total time equals the SLOWEST node's time regardless of how fast others finish
Tree-based reduction allows O(log n) collective operations instead of O(n)
Modern cloud computing, big data, and AI training all rely on distributed processing principles

Distributed Processing

Introduction

Distributed vs Shared Memory

Message Passing Model

MPI (Message Passing Interface)

Network Topologies

Fat Tree (Most Common in Modern Clusters)

Communication Patterns

Point-to-Point

Broadcast

Scatter/Gather

All-to-All

Reduction

Performance Considerations

Communication Overhead

Computation-to-Communication Ratio

Load Balancing

Real-World Distributed Systems

Key Takeaways

Continue learning this concept

Distributed Processing

Introduction

Distributed vs Shared Memory

Message Passing Model

MPI (Message Passing Interface)

Network Topologies

Fat Tree (Most Common in Modern Clusters)

Communication Patterns

Point-to-Point

Broadcast

Scatter/Gather

All-to-All

Reduction

Performance Considerations

Communication Overhead

Computation-to-Communication Ratio

Load Balancing

Real-World Distributed Systems

Key Takeaways

Continue learning this concept