Module-1

Syllabus: Introduction to parallel programming, Parallel

hardware and parallel software - Classifications of parallel
computers, SIMD systems, MIMD systems, Interconnection
networks, Cache coherence, Shared-memory vs. distributed-
memory, Coordinating the processes/threads, Shared-
memory, Distributed-memory.
Flynn’s Taxonomy
Based on Number of Instruction Streams & Data Streams.
Categories:
SISD (Single Instruction, Single Data)
Classical von Neumann architecture.
Executes one instruction at a time on one data item.
SIMD (Single Instruction, Multiple Data)
One instruction applied to many data items in parallel.
Used in graphics, image processing, vector operations.
MIMD (Multiple Instruction, Multiple Data)
Multiple instructions on multiple data streams.
Most modern multi-core CPUs & distributed systems.
MISD (Multiple Instruction, Single Data)
Multiple instructions act on the same data.
Mostly theoretical/fault-tolerant systems.
SIMD (Single Instruction, Multiple Data)
•Definition: SIMD systems apply the same instruction simultaneously to
multiple data items.
•Structure:
• One Control Unit → broadcasts the instruction.
• Multiple Datapaths → execute that instruction on different pieces of
data in parallel.
Example 1: Vector Addition

for (i = 0; i < n; i++)

x[i] += y[i];
If the system has m datapaths and m<n, we can simply execute the
additions
in blocks of m elements at a time.

For example, if m = 4 and n = 15, we can first

add elements 0 to 3, then elements 4 to 7, then elements 8 to 11, and finally
elements
12 to 14.

if y[i] is positive:
for ( i = 0; i < n ; i ++)
i f ( y [ i ] > 0 . 0 ) x [ i] += y [ i ] ;
Vector processors:
Operate on arrays (vectors) of data rather than individual scalars.
Key difference from conventional CPUs:
Vector processors → arrays
CPUs → scalars
Key Characteristics of recent systems
1. Vector Registers:
• Store a vector of operands, Fixed vector length (4–256 elements, 64-bit
each)
• Operate simultaneously on contents.
2. Vector Instructions:
• Operate on entire vectors instead of scalars
for (i = 0; i < n; i++)
x[i] += y[i];
• Vectorized as a single load, add, store for each vector block
3. Vectorized & Pipelined Functional Units
• Apply the same operation across all vector elements
• Efficient parallel execution
4. Interleaved Memory:
• Multiple banks allow parallel access with Reduces delays in loading/storing
successive elements
5. Strided Memory Access/Scatter/Gather
• Access elements at fixed intervals (e.g., 1st, 5th, 9th…)
• Scatter → Write elements at irregular intervals
• Gather → Read elements at irregular intervals
Advantages of Vector Processors:
• Very fast and efficient for many applications.
• Vectorizing compilers: Automatically detect vectorizable code.
• Provide feedback on non-vectorizable loops.
• High memory bandwidth – every loaded item is used (unlike cache-based
systems).
Limitations of Vector Processors:
• Poor handling of irregular data structures(linked lists, trees, graphs)
• Limited scalability (vector length cannot grow indefinitely) hardware
complexity, power consumption, and after a certain point make it
impractical.
• Long-vector processors are custom & expensive
• Commodity systems support only short vectors
Graphics Processing Units (GPUs):
• GPUs are specialized processors for real-time graphics rendering.
• Graphics APIs (e.g., OpenGL, DirectX, Vulkan) represent objects using
points, lines, and triangles.
• Use a graphics pipeline to convert objects into pixels for display.
• Some pipeline stages are programmable using shader functions.
• Shaders: Small C-like functions, often just a few lines of code.
• Applied to many elements (vertices, fragments, etc.) → implicitly parallel.
• SIMD in GPUs, Enables massive parallelism for graphics workloads.
• GPUs are not pure SIMD systems
• current generation GPUs can run more than one instruction stream on a
single core. typical GPUs can have dozens of cores, and these cores are
also capable of executing independent instruction streams.
• So GPUs are neither purely SIMD nor purely MIMD.
•GPUs can use shared memory, distributed memory, or both.
•They’re increasingly used in general-purpose computing (not just
graphics).
•Languages like CUDA make it possible to program GPUs efficiently.
•GPUs Beyond Graphics:
Useful in:
•Scientific computing (simulations, physics, chemistry)
•Artificial Intelligence & Machine Learning
•Big Data & Analytics
• MIMD systems typically consist of a collection of fully independent
processing units or cores, each of which has its own control unit and its
own datapath.
• MIMD systems are usually asynchronous,(the processors can operate at
their own pace, there is no global clock, and there may be no relation
between the system times on two different processors)
Two principal types of MIMD systems:
1. Shared memory systems
• All processors share a global memory
• Multiple processors connected to a common memory via an
interconnection network
• Communication is Implicit (via shared variables) through shared data
structures
• Easy to program, but limited scalability
2. Distributed-memory systems
• Each processor paired with private memory
• Communication is Explicit via message passing or special functions
• Highly scalable (used in supercomputers, Cluster computers,
cloud data centers)
Shared-memory systems:
• The most widely available shared-memory systems use one or more
multicore processors.
• A multicore processor has multiple CPUs or cores on a single chip.
• Each core usually has: Private L1 cache (fastest, only for that core)
• Higher-level caches (L2, L3) which may be shared or private.
• Connection to Main Memory
There are two types of shared-memory systems:
1. UMA – Uniform Memory Access
• All processors are connected to main memory through the same
interconnect.
• Memory access time is the same for all cores, regardless of which
memory location is accessed.
• Example: Small-scale symmetric multiprocessing (SMP) systems.
2. NUMA – Non-Uniform Memory Access
• Each processor (or chip) is connected directly to a local block of memory.
• A core can access its local memory faster than memory attached to another
processor.
• To access another processor’s memory, it must go through special
hardware.
Distributed-Memory Systems
• The most widely available distributed-memory systems are called clusters
• Each processor has private memory
• Communication is explicit (messages)
• Examples: Clusters, Grids
Grids
• Connects geographically distributed computers
• Infrastructure for large-scale distributed computing
• Nodes often heterogeneous
Interconnection Networks:
• Importance of Interconnection Networks
• Interconnect plays a crucial role in both distributed- and shared-memory
systems.
• The interconnect is the communication backbone that links processors
memory (in shared-memory) or processors processors (in
distributed-memory).
• Even if CPUs and memory are very fast, a slow interconnect becomes the
bottleneck.
• Thus, interconnect design strongly affects latency, bandwidth, and
scalability of parallel systems.
• A weak interconnect can seriously degrade performance of parallel
programs.
[Link]-memory interconnects
➢ Bus-based Interconnects
➢ Switched Interconnects (Crossbar)

2. Distributed-memory interconnects
2.1 Direct interconnects
➢ Ring
➢ Two dimensional toroidal mesh.
➢ Hypercube

2.2 Indirect interconnects

➢ Distributed-memory Crossbar
➢ Omega Network.
Shared-Memory Interconnects
1. Bus-based Interconnects

•Early systems used bus connections between processors and memory

Communication wires are shared among devices

• Advantages: Low cost, flexible, easy to connect and access to main

memory.
• Disadvantages: Contention(two or more components compete for the
same resource at the same time) increases with more processors →
performance drops(processors would frequently have to wait)
• Not scalable for large systems
2. Switched Interconnects (Crossbar)

• Switches Replaces bus in larger systems

• Uses bidirectional links + switches for processor–memory connections
• Allows simultaneous communication among devices
• Switch Configurations:
Switch can connect vertical horizontal links in two ways (Fig).
• Switches are used to control the routing of data among the connected
devices.
• The most commonly used one is crossbar switch.
• The circle represent the switches.
Example:P1 → M4, P2 → M3, P3 → M1, P4 → M2
P1 writes to M4, P2 reads from M3, P3 reads from M1, and P4 writes to M2.
• No two processors are requesting the same memory at the same moment.
• All four communications happen in parallel, due to the switches.
• A conflict only arises when two processors try to access the same
memory module at the same time.
• Crossbar: High performance(Faster), scalable, but expensive
Feature Bus Crossbar
Low (cheap to High (switches + links
Cost
implement) are expensive)
Easy to connect Requires more
Flexibility
multiple devices hardware complexity
Shared wires → one at Multiple simultaneous
Communication
a time connections
Poor (performance Good (supports many
Scalability
drops with size) processors)
Slower due to Much faster, less
Performance
contention contention
Small shared-memory Large, high-
Use Case
systems performance systems
Distributed-Memory Interconnects:

Divided into Direct and Indirect interconnects

• Direct interconnect:
• Each switch directly connects to a processor–memory pair
• Switches are also connected to each other
• Also called as Fully Connected Network.

Most commonly used direct interconnects are:

➢ Ring
➢ Two dimensional toroidal mesh.
➢ Hypercube

Ring
• Simple structure, processors arranged in a loop
• Allows multiple simultaneous communications
Limitation:
• Some processors must wait for others
• In a ring interconnect, all processors are connected in a closed loop.
• If two processors want to communicate with different processors, but
their communication paths overlap, one of them has to wait.
Example:

•Suppose P1 wants to send data to P3 and at the same time P2 also wants to
send data to P3.
•Both messages must travel along the same ring path → contention occurs.
•This means the ring cannot always handle many simultaneous
communications efficiently.
•If there are ‘n’ processors, the number of links is 2n.
Two-dimensional Toroidal Mesh
• The complex switches are mainly used.
• Supports higher number of simultaneous communication paths.
• Each switch handles 5 links instead of 3 (higher cost).
• Provides better scalability and higher connectivity than ring.
• Connectivity is using bisection width(Number of links that must be cut
to divide the network into two equal halves)
• Example (Ring with 8 nodes):
• Divide into 2 groups (4 nodes each).
• Only 2 links connect the halves.
• So, bisection width = 2.
• This means at most 2 simultaneous communications can cross
between the two halves at the same time.

• Toroidal Mesh = costlier, more complex, but supports greater

scalability, parallelism, and higher connectivity.
• what’s the bisection width? The bisection width is supposed to give a
“worst-case” estimate, so the bisection width is two—not four.
• Bisection width = minimum links removed to split network into 2 equal
halves.
Square 2D Toroidal Mesh:
• Total nodes: 𝒑 = 𝒒𝟐 ,where 𝑞(𝑒𝑣𝑒𝑛)is the number of nodes along each
row/column.
• Each node is connected in a grid-like structure with wraparound links
(edges on one side connect to the opposite side, like a torus).
Dividing into Two Equal Halves:
• The middle horizontal links (between two halves of the grid).
• The wraparound horizontal links (because the torus wraps around and
connects both halves again).
• Each row requires 2 cuts.
• Total cuts = 2𝑞.
• Since 𝑝 = 𝑞 2 :
The figure represents a toroidal mesh network (here drawn as a 4×4
grid).
•Each square represents a processor node.
•The lines between nodes represent bidirectional communication links.
Bandwidth: Rate at which a link transmits data.
• Measured in megabits/sec (Mbps) or megabytes/sec (MBps).
Bisection Bandwidth:
• Measure of network quality.
Computation of Bisection Bandwidth:
• Divide network into two equal halves.
• Bisection Bandwidth = total bandwidth of links crossing the cut.
• Example: if the links in a ring have a bandwidth of one billion bits per
second, then the bisection bandwidth of the ring will be two billion bits
per second or 2000 megabits per second.
Hypercube Interconnection Network:
• A highly connected direct interconnect used in parallel systems.
• Hypercubes are built inductively(step-by-step construction based on a
pattern or rule)
• 1D hypercube → 2 processors, fully connected.
• 2D hypercube → formed from two 1D hypercubes.
• 3D hypercube → formed from two 2D hypercubes.
• In general: d-dimensional hypercube has p = 2d nodes, and a switch in a
d-dimensional hypercube is directly connected to a processor and d
switches.
• The bisection width of a hypercube is p/2, so it has more connectivity
than a ring or toroidal mesh, but the switches must be more powerful,
since they must support 1 + d = 1 + log2(p) wires, while the mesh
switches only require five wires.
• hypercube are more expensive to construct than a toroidal mesh.
Indirect interconnects:
• The switches may not be directly connected to a processor
• Uses the unidirectional links and a collection of processors, each of which
has an outgoing and an incoming link, and a switching network.
Types:
[Link]
2. omega network

Distributed-memory Crossbar:
• Distributed-memory crossbar in Fig. 2.14 has unidirectional links.
• Two processors don’t attempt to communicate with the same
processor, all the processors can simultaneously communicate with
another processor.
omega network:
• A multistage interconnection network (MIN) built using 2×2 switches
(two inputs, two outputs).
It connects processors to memory modules through log₂(p) stages.
• Very powerful, but very expensive.
• Not all communication patterns can occur simultaneously.
Latency:
• Definition: The time it takes for a message (or the first bit of data) to
travel from source to destination.
• Usually measured in nanoseconds (ns) or microseconds (µs).
Bandwidth:
• Amount of data that can be transmitted per unit time
• Typically measured in bytes/second (B/s), MB/s, or GB/s.
• Latency dominates when transmitting small amounts of data, because
the setup time is more significant than the actual data transfer.
• Bandwidth dominates when transmitting large blocks of data, because
once the pipeline is full, the sustained rate matters more.
Example:
• Latency = 1 µs
• Bandwidth = 10 GB/s
• Sending 1 KB: The latency (1 µs) dominates; transmission time is tiny.
• Sending 1 GB: Latency is negligible, and bandwidth (10 GB/s → ~0.1 s)
Shared-memory vs. Distributed-memory (MIMD Systems)

Shared-memory systems:
• Easy programming model (shared data structures).
• Bus interconnects: simple but limited to few processors (contention).
• Crossbars: support more processors but very expensive.
• Not scalable for large systems.

Distributed-memory systems:
• Each processor has its own memory → communication via messages.
• Inexpensive interconnects (e.g., hypercube, toroidal mesh).
• Scales to thousands of processors.
• Well-suited for large data & computation-intensive problems.
Cache coherence:
• Shared-memory system with multiple cores, each with private cache.
• No issue if processors only read shared data.
• Problem arises when one core updates a shared variable:
• Other cores may still use the old cached value.
• Example:
• P1 updates A = 8.
• P2 updates A = 9 in its cache.
• Leads to inconsistent results
• Root cause: Caches do not automatically synchronize updates across
cores.
• This inconsistency is called the Cache Coherence Problem.
• CPU caches are managed by system hardware: programmers don’t have
• direct control over them
Snooping cache coherence is one main solution:
• Based on the idea of broadcasting updates across a shared interconnect
(often a bus).
• Each core snoops (monitors) the interconnect for updates made by others.
• If a core detects that another has updated a cache line, it can invalidate or
update its own cached copy.
• Works with both write-through and write-back caches:
• Write-through: Updates go directly to memory, so snooping cores can just
“watch” the bus.
• Write-back: Since updates aren’t sent immediately to memory, extra
communication is required to inform others.
• All caches maintain coherence by monitoring (snooping) shared
communication channels.
Problem with Snooping:
• Works well for small systems.
• Not scalable because it requires broadcasts for every update.
• In large systems, broadcasting across the entire interconnect is too costly.
Directory-based cache coherence:
• Use the Centralized directory to track which cores cache each memory
block.
Operation:On read: Directory updated to record which core has a copy.
• On write: Directory consulted → only cores with that block are notified to
invalidate/update.
Advantages:
• Eliminates costly broadcasts.
• Scales well to large multiprocessor systems.
Disadvantages:
• Requires extra storage and complexity for directory structures.
False sharing
• CPU caches work with cache lines (usually 64 bytes), not with individual
variables.
• If an array element lies inside a cache line, loading one element pulls in
neighboring ones too.
Example:
• Array y[m] is shared between threads/cores.
• Suppose m = 8, each element is a double (8 bytes).
• So y occupies 8 × 8 = 64 bytes → exactly 1 cache line.
• Work divided among cores: Core 0 → y[0..3] ,Core 1 → y[4..7]
• Even though they access different elements, both y[0..3] and y[4..7] are
inside the same cache line.
• When Core 0 updates y[0], the cache line is marked "dirty" in Core 0’s
cache → invalidated in Core 1.
When Core 1 updates y[4], the line is fetched again into Core 1’s cache →
invalidated in Core 0.
The Consequence:

When one core updates its variable, the entire cache line is invalidated
for the other cores.

Even though they are accessing different variables, the other cores must
then fetch a fresh copy of the entire cache line from memory, causing a
significant performance
Coordinating the processes/threads
• For simple problems, just divide the work
• Some problems are(Trivial Parallelism ) "embarrassingly parallel,"
meaning they can be solved by simply dividing the work among
processes or threads. A simple array addition is given as an example:
• Example: Adding two arrays
for (int i = 0; i < n; i++)
x[i] += y[i];
➢ To parallelize:
• Divide the array into chunks.
• Each thread handles its own chunk:
Thread 0 → 0 … n/p – 1, Thread 1 → n/p … 2n/p – 1,so on
➢ Key benefits:
• Independent work: No thread needs data from another.
• No synchronization needed: Each thread works on separate elements.
• Scales perfectly (near linear speedup).
Conditions for Good Parallelism
Load balancing:
• Each thread should get about the same amount of work.
• Prevents idle threads waiting for others to finish.
Minimal communication:
• Threads should work with local data as much as possible.
• Extra communication hurts performance.
[Link] most (non-Trivial)problems, additional coordination is needed.
• The vast majority of problems are more complex and require additional
coordination between processes and threads. For these problems,
programmers also need to:
• Arrange for synchronization: ensuring that processes or threads wait for
each other at certain points in the program before proceeding.
• Arrange for communication: This involves setting up the transfer of data
between processes or threads.
For Synchronization: Distributed memory (MPI), Shared memory (OpenMP,
Shared-memory:
• Communication among the threads is usually done through shared
variables.
• communication is implicit rather than explicit.
Dynamic and static threads
• shared-memory programs use dynamic threads.
How it works:
• Master thread is always running.
• New work request arrives → master spawns (forks) a worker thread.
• Worker thread executes the task, then joins back with master and
terminates.
Advantages:
• Efficient use of resources → threads exist only while working.
Disadvantages:
• If work requests are frequent, thread creation/destruction can slow things
down.
Alternative to the dynamic paradigm is the static thread paradigm.
How it works:
• Master thread forks all worker threads at the start (after initialization).
• Threads remain active until all tasks are completed.
• At the end, threads join the master, and cleanup is done.
Advantages:
• Lower overhead (fork/join done only once).
• Can achieve better performance if enough resources are available.
• Closer to distributed-memory programming models (MPI style).
Disadvantages:
• Resource usage may be inefficient if some threads are idle.
• Memory allocated for idle threads (stack, PC, etc.) stays occupied.
Nondeterminism and Race Conditions

Nondeterminism:
• This occurs in asynchronous MIMD systems, where a program's output
can vary from one run to another even with the same input.
• Caused by different thread scheduling, OS behavior, and hardware timing.
• For example, a simple printf statement on 2 thread, can have its output
order change from run to run:
printf("Thread %d > my_x = %d\n", my_rank, my_x);
Output 1: Thread 0 > my_x = 7
Thread 1 > my_x = 19
Output 2: Thread 1 > my_x = 19
Thread 0 > my_x = 7
Both correct, order nondeterministic.
Race Conditions
When It’s a Problem
• Two threads want to update a shared variable x:
my_val = Compute_val(my_rank);
x += my_val;
• The x += my_val operation is not atomic and can lead to an incorrect final
value for x if both threads try to update it at the same time.
Solutions to Race Conditions
• To prevent race conditions, a critical section is used.
• This is a block of code that can only be executed by one thread at a time,
ensuring mutual exclusion.
• Implemented via locks, semaphores, mutexes, or atomic operations.
Mutex (Mutual Exclusion Lock):
• Without protection, x += my_val can cause a race condition.
• Fix: ensure only one thread modifies x at a time → enforce mutual
exclusion.
Lock(&add_my_val_lock);
x += my_val;
Unlock(&add_my_val_lock);
• Mutex (mutual exclusion lock) ensures only one thread enters the critical
section.
• No predetermined order → either thread may go first.
Drawback: Critical section is serialized (threads wait for each other).
fewer and shorter critical sections = better performance.
Busy-Waiting (Spinlock)
if (my_rank == 1)
while (!ok_for_1); // Busy-wait loop
x += my_val; // Critical section
if (my_rank == 0)
ok_for_1 = true;
• Thread 1 loops until condition is true.
• Simple but wastes CPU cycles (keeps checking without useful work).
• Acceptable if wait times are very short, otherwise inefficient.
Semaphores
• Similar to mutexes but more flexible.
• Can allow more than one thread into a critical region (counting
semaphores).
• Useful for producer-consumer, bounded buffer, etc.
• Mutex = special case of semaphore with count = 1.
Distributed-Memory Systems:

• Each core can access only its private memory

• Communication is explicit
• Requires coordination among processes
• The most common way to exchange data is through message-passing APIs,
such as MPI (Message Passing Interface).
• Distributed programs usually run as multiple processes, not threads.
[Link] is Message-Passing?
• A message-passing API provides at least two functions: Send and Receive.
• Processes communicate by explicitly sending and receiving messages.
• Each process has a unique rank (0, 1, …, p–1).
Example:
char message[100];
my_rank = Get_rank();
if (my_rank == 1) {
sprintf(message, "Greetings from process 1");
Send(message, MSG_CHAR, 100, 0);
} else if (my_rank == 0) {
Receive(message, MSG_CHAR, 100, 1);
printf("Process 0 > Received: %s\n", message);
}
Process 1 → creates and sends message.
Process 0 → receives message and prints it.
• The program segment is SPMD(Single Program Multiple Data)
• All processes run the same executable, but their actions differ by rank.
• Message passing may be either Blocking or nonblocking-also known as
Synchronous and asynchronous.
Behaviour of Send & Receive:
• Blocking Send: The sending process is blocked until the message is
received by the receiving process.
• Buffered Send: copies message to internal buffer and returns immediately.
• Nonblocking Send: The sending process sends the message and resumes
operation.
• Blocking receive: The receiver blocks until a message is available.
• Non Blocking receive: The receiver retrieves either a valid message is
available.
Additional Functions in APIs
Collective Communication
• Broadcast: one process sends data to all others.
• Reduction: combines values from all processes (e.g., sum, min, max).
2. One-sided communication(Remote Memory Access – RMA)
• Only one process initiates the communication using Get and put
operation
• Put → copy data from local memory into a remote process’s memory.
• Get → copy data from a remote process’s memory into local memory.
• communication can be simpler and more efficient.
How It Works:
• Process 0 calls Put(value, target_rank=1, target_address)
→ It places a value into Process 1’s memory.
• Process 1 does nothing actively at that moment.
→ Its memory is updated “remotely” by Process 0.
• Later, Process 1 reads the value directly from its local memory.
• Advantages: Lower overhead, simpler, faster in some scenarios.
• Challenges: Requires careful synchronization, flagging, and is harder to
debug.
• Usage: Supported in MPI models, common in HPC.
[Link] global address space languages(PGAS)
PGAS: Shared-Memory Programming for Distributed Systems

PGAS languages provide a shared-memory programming model for

distributed-memory hardware, aiming to simplify programming while
maintaining high performance.
Core Concept: The language creates a single logical address space that
spans the private memories of all processes.
How it Works: To prevent poor performance from accessing remote memory,
PGAS languages give the programmer tools to control how shared data is
distributed. Programmers can ensure that a process primarily accesses data
that is local to its own memory, which is much faster.
Example: In a vector addition, a programmer can define the arrays as shared
but then ensure that each process only works on the portion of the array
that is located in its own local memory:

shared int n = ...;

shared double x[n], y[n];
private int i, my_first_element, my_last_element;

my_first_element = ...;
my_last_element = ...;

// Initialize x and y
...

for (i = my_first_element; i <= my_last_element; i++)

x[i] += y[i];