Synchronization Primitives

Introduction: Why Synchronization?

When multiple threads access shared data concurrently, we need synchronization primitives to ensure correctness and prevent race conditions. These tools are the foundation of thread-safe programming.

Visual: The Problem Without Synchronization

---

Critical Sections and Race Conditions

What is a Critical Section?

A critical section is a code segment that accesses shared resources and must be executed atomically (as a single, indivisible operation) to prevent race conditions.

Visual: Critical Section

Example: Race Condition

Python

race_condition.py

import threading

# Shared counter
counter = 0

def increment():
    global counter
    # Critical section - NOT protected!
    for _ in range(100000):
        counter += 1  # Not atomic: read-modify-write

# Create multiple threads
threads = []
for _ in range(5):
    thread = threading.Thread(target=increment)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"Expected: 500000, Got: {counter}")
# Output: Expected: 500000, Got: 342156 (WRONG!)

Java

RaceCondition.java

public class RaceCondition {
    private static int counter = 0;

    public static void increment() {
        // Critical section - NOT protected!
        for (int i = 0; i < 100000; i++) {
            counter++;  // Not atomic: read-modify-write
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread[] threads = new Thread[5];

        for (int i = 0; i < 5; i++) {
            threads[i] = new Thread(() -> increment());
            threads[i].start();
        }

        for (Thread thread : threads) {
            thread.join();
        }

        System.out.println("Expected: 500000, Got: " + counter);
        // Output: Expected: 500000, Got: 342156 (WRONG!)
    }
}

Race Condition

The counter++ operation is not atomic. It consists of three steps:

1. Read current value
2. Increment it
3. Write it back

When multiple threads do this simultaneously, some increments are lost!

---

Mutual Exclusion: Locks (Mutex)

Locks (also called mutexes) provide mutual exclusion—ensuring only one thread can execute a critical section at a time.

Visual: How Locks Work

Basic Lock Usage

Python

lock_example.py

import threading

counter = 0
lock = threading.Lock()  # Create a lock

def increment():
    global counter
    for _ in range(100000):
        lock.acquire()  # Acquire lock
        try:
            counter += 1  # Critical section - protected!
        finally:
            lock.release()  # Always release lock

# Or use context manager (recommended)
def increment_safe():
    global counter
    for _ in range(100000):
        with lock:  # Automatically acquire/release
            counter += 1  # Critical section

threads = []
for _ in range(5):
    thread = threading.Thread(target=increment_safe)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"Expected: 500000, Got: {counter}")
# Output: Expected: 500000, Got: 500000 (CORRECT!)

Java

LockExample.java

import java.util.concurrent.locks.ReentrantLock;

public class LockExample {
    private static int counter = 0;
    private static ReentrantLock lock = new ReentrantLock();

    public static void increment() {
        for (int i = 0; i < 100000; i++) {
            lock.lock();  // Acquire lock
            try {
                counter++;  // Critical section - protected!
            } finally {
                lock.unlock();  // Always release lock
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread[] threads = new Thread[5];

        for (int i = 0; i < 5; i++) {
            threads[i] = new Thread(() -> increment());
            threads[i].start();
        }

        for (Thread thread : threads) {
            thread.join();
        }

        System.out.println("Expected: 500000, Got: " + counter);
        // Output: Expected: 500000, Got: 500000 (CORRECT!)
    }
}

Reentrant Locks

A reentrant lock allows the same thread to acquire the lock multiple times without deadlocking. This is useful for recursive methods or when calling other methods that need the same lock.

Visual: Reentrant Lock

Example: Reentrant Lock

Python

reentrant_lock.py

import threading

lock = threading.RLock()  # Reentrant Lock

def outer_function():
    with lock:  # First acquisition
        print("Outer: Lock acquired")
        inner_function()  # Calls inner function
        print("Outer: Lock released")

def inner_function():
    with lock:  # Second acquisition (same thread!)
        print("Inner: Lock acquired (reentrant)")
        # Do work
        print("Inner: Lock released")

# This works without deadlock!
outer_function()

Java

ReentrantLockExample.java

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {
    private static ReentrantLock lock = new ReentrantLock();

    public static void outerMethod() {
        lock.lock();  // First acquisition
        try {
            System.out.println("Outer: Lock acquired");
            innerMethod();  // Calls inner method
        } finally {
            lock.unlock();
            System.out.println("Outer: Lock released");
        }
    }

    public static void innerMethod() {
        lock.lock();  // Second acquisition (same thread!)
        try {
            System.out.println("Inner: Lock acquired (reentrant)");
            // Do work
        } finally {
            lock.unlock();
            System.out.println("Inner: Lock released");
        }
    }

    public static void main(String[] args) {
        outerMethod();  // Works without deadlock!
    }
}

synchronized vs ReentrantLock (Java)

Java provides two ways to achieve mutual exclusion:

Feature	synchronized	ReentrantLock
Syntax	Keyword (implicit)	Class (explicit)
Fairness	No	Yes (optional)
Try Lock	No	Yes (tryLock())
Interruptible	No	Yes (lockInterruptibly())
Condition Support	Limited	Full (newCondition())
Performance	Slightly faster	Slightly slower

Example: synchronized vs ReentrantLock

synchronized

SynchronizedExample.java

public class SynchronizedExample {
    private int counter = 0;

    // Implicit locking with synchronized
    public synchronized void increment() {
        counter++;
    }

    // Synchronized block
    public void incrementBlock() {
        synchronized (this) {
            counter++;
        }
    }
}

ReentrantLock

ReentrantLockExample.java

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {
    private int counter = 0;
    private ReentrantLock lock = new ReentrantLock(true);  // Fair lock

    // Explicit locking
    public void increment() {
        lock.lock();
        try {
            counter++;
        } finally {
            lock.unlock();
        }
    }

    // Try lock (non-blocking)
    public boolean tryIncrement() {
        if (lock.tryLock()) {
            try {
                counter++;
                return true;
            } finally {
                lock.unlock();
            }
        }
        return false;  // Lock not available
    }
}

---

Resource Limiting: Semaphores

A semaphore controls access to a resource with a counter. Unlike a lock (which allows only one thread), a semaphore can allow N threads to access a resource simultaneously.

Visual: Semaphore Concept

Binary Semaphore vs Counting Semaphore

• Binary Semaphore: Count = 1 (similar to a lock, but can be released by different thread)
• Counting Semaphore: Count = N (allows N concurrent accesses)

Example: Rate Limiter with Semaphore

Python

rate_limiter.py

import threading
import time

class RateLimiter:
    def __init__(self, max_requests, time_window):
        self.semaphore = threading.Semaphore(max_requests)
        self.time_window = time_window
        self.last_reset = time.time()
        self.lock = threading.Lock()

    def acquire(self):
        """Acquire a permit, blocking if necessary"""
        self.semaphore.acquire()
        with self.lock:
            current_time = time.time()
            if current_time - self.last_reset >= self.time_window:
                # Reset permits
                for _ in range(self.semaphore._value):
                    self.semaphore.release()
                self.last_reset = current_time

    def release(self):
        """Release a permit"""
        self.semaphore.release()

# Usage: Allow max 5 concurrent requests
limiter = RateLimiter(max_requests=5, time_window=1.0)

def make_request(request_id):
    limiter.acquire()
    try:
        print(f"Request {request_id} processing...")
        time.sleep(0.5)  # Simulate work
    finally:
        limiter.release()
        print(f"Request {request_id} completed")

# Create 10 threads
threads = []
for i in range(10):
    thread = threading.Thread(target=make_request, args=(i,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

Java

RateLimiter.java

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class RateLimiter {
    private final Semaphore semaphore;
    private final int maxRequests;
    private final long timeWindowMillis;
    private long lastReset;
    private final Object lock = new Object();

    public RateLimiter(int maxRequests, long timeWindowMillis) {
        this.maxRequests = maxRequests;
        this.timeWindowMillis = timeWindowMillis;
        this.semaphore = new Semaphore(maxRequests);
        this.lastReset = System.currentTimeMillis();
    }

    public void acquire() throws InterruptedException {
        semaphore.acquire();
        synchronized (lock) {
            long currentTime = System.currentTimeMillis();
            if (currentTime - lastReset >= timeWindowMillis) {
                // Reset permits
                int available = maxRequests - semaphore.availablePermits();
                semaphore.release(available);
                lastReset = currentTime;
            }
        }
    }

    public void release() {
        semaphore.release();
    }

    public static void main(String[] args) throws InterruptedException {
        RateLimiter limiter = new RateLimiter(5, 1000);  // 5 requests per second

        for (int i = 0; i < 10; i++) {
            final int requestId = i;
            new Thread(() -> {
                try {
                    limiter.acquire();
                    System.out.println("Request " + requestId + " processing...");
                    Thread.sleep(500);
                    limiter.release();
                    System.out.println("Request " + requestId + " completed");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }
    }
}

---

Coordination: Condition Variables

Condition variables allow threads to wait for a specific condition to become true, enabling efficient thread coordination.

Visual: Condition Variable Pattern

Example: Producer-Consumer with Condition Variables

Python

condition_example.py

import threading
import time

class BoundedBuffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = []
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)

    def put(self, item):
        with self.lock:
            # Wait while buffer is full
            while len(self.buffer) >= self.capacity:
                self.not_full.wait()  # Releases lock, waits

            self.buffer.append(item)
            print(f"Produced: {item}")
            self.not_empty.notify()  # Wake up consumer

    def get(self):
        with self.lock:
            # Wait while buffer is empty
            while len(self.buffer) == 0:
                self.not_empty.wait()  # Releases lock, waits

            item = self.buffer.pop(0)
            print(f"Consumed: {item}")
            self.not_full.notify()  # Wake up producer
            return item

# Usage
buffer = BoundedBuffer(capacity=5)

def producer():
    for i in range(10):
        buffer.put(i)
        time.sleep(0.1)

def consumer():
    for _ in range(10):
        buffer.get()
        time.sleep(0.2)

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()

Java

ConditionExample.java

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionExample {
    static class BoundedBuffer {
        private final Queue<Integer> buffer = new LinkedList<>();
        private final int capacity;
        private final Lock lock = new ReentrantLock();
        private final Condition notFull = lock.newCondition();
        private final Condition notEmpty = lock.newCondition();

        public BoundedBuffer(int capacity) {
            this.capacity = capacity;
        }

        public void put(int item) throws InterruptedException {
            lock.lock();
            try {
                // Wait while buffer is full
                while (buffer.size() >= capacity) {
                    notFull.await();  // Releases lock, waits
                }
                buffer.offer(item);
                System.out.println("Produced: " + item);
                notEmpty.signal();  // Wake up consumer
            } finally {
                lock.unlock();
            }
        }

        public int get() throws InterruptedException {
            lock.lock();
            try {
                // Wait while buffer is empty
                while (buffer.isEmpty()) {
                    notEmpty.await();  // Releases lock, waits
                }
                int item = buffer.poll();
                System.out.println("Consumed: " + item);
                notFull.signal();  // Wake up producer
                return item;
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) {
        BoundedBuffer buffer = new BoundedBuffer(5);

        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    buffer.put(i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();

        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    buffer.get();
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

notify() vs notifyAll()

• notify(): Wakes up one waiting thread (unpredictable which one)
• notifyAll(): Wakes up all waiting threads (they compete for the lock)

When to Use Which

• Use notify() when only one thread can proceed (e.g., single consumer)
• Use notifyAll() when multiple threads might be able to proceed (e.g., multiple consumers, complex conditions)

---

Coordination: Barriers and Latches

Barriers

A barrier makes threads wait until all threads reach a synchronization point.

Visual: Barrier

Example: Barrier

Python

barrier_example.py

import threading

barrier = threading.Barrier(3)  # 3 threads must wait

def worker(worker_id):
    print(f"Worker {worker_id}: Phase 1")
    barrier.wait()  # Wait for all threads
    print(f"Worker {worker_id}: Phase 2 (all arrived!)")

threads = []
for i in range(3):
    thread = threading.Thread(target=worker, args=(i,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

Java

BarrierExample.java

import java.util.concurrent.CyclicBarrier;

public class BarrierExample {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3);  // 3 threads must wait

        for (int i = 0; i < 3; i++) {
            final int workerId = i;
            new Thread(() -> {
                try {
                    System.out.println("Worker " + workerId + ": Phase 1");
                    barrier.await();  // Wait for all threads
                    System.out.println("Worker " + workerId + ": Phase 2 (all arrived!)");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

CountDownLatch (Java)

A CountDownLatch is a one-time synchronization point. Threads wait until a count reaches zero.

Java

CountDownLatchExample.java

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);  // Count starts at 3

        // Worker threads
        for (int i = 0; i < 3; i++) {
            final int workerId = i;
            new Thread(() -> {
                try {
                    System.out.println("Worker " + workerId + " working...");
                    Thread.sleep(1000);
                    latch.countDown();  // Decrement count
                    System.out.println("Worker " + workerId + " finished");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }

        System.out.println("Main thread waiting...");
        latch.await();  // Wait until count reaches 0
        System.out.println("All workers finished! Main thread proceeds.");
    }
}

---

Read Optimization: Read-Write Locks

Read-Write Locks optimize for read-heavy workloads by allowing multiple readers or a single writer.

Visual: Read-Write Lock

Example: Thread-Safe Cache with Read-Write Lock

Java

ReadWriteLockExample.java

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockExample {
    private final Map<String, String> cache = new HashMap<>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    public String get(String key) {
        lock.readLock().lock();  // Multiple readers allowed
        try {
            return cache.get(key);
        } finally {
            lock.readLock().unlock();
        }
    }

    public void put(String key, String value) {
        lock.writeLock().lock();  // Exclusive access
        try {
            cache.put(key, value);
        } finally {
            lock.writeLock().unlock();
        }
    }

    public static void main(String[] args) {
        ReadWriteLockExample cache = new ReadWriteLockExample();

        // Multiple readers can read simultaneously
        for (int i = 0; i < 10; i++) {
            final int readerId = i;
            new Thread(() -> {
                String value = cache.get("key");
                System.out.println("Reader " + readerId + " read: " + value);
            }).start();
        }

        // Writer has exclusive access
        new Thread(() -> {
            cache.put("key", "value");
            System.out.println("Writer updated cache");
        }).start();
    }
}

Python Limitation

Python’s threading.RLock doesn’t support read-write locks natively. For read-write locks in Python, consider:

• Using multiprocessing with separate processes
• Third-party libraries
• Implementing your own with condition variables

---

Thread-Local Storage

Thread-Local Storage provides each thread with its own independent copy of a variable, avoiding the need to pass parameters through method calls.

Visual: Thread-Local Storage

Example: Request Context with ThreadLocal

Python

thread_local_example.py

import threading

# Thread-local storage
request_context = threading.local()

def set_user(user_id):
    request_context.user_id = user_id
    request_context.request_id = threading.current_thread().name

def get_user():
    return getattr(request_context, 'user_id', None)

def process_request(user_id):
    set_user(user_id)
    print(f"Thread {threading.current_thread().name}: "
          f"Processing request for user {get_user()}")

# Each thread has its own context
threads = []
for i in range(3):
    thread = threading.Thread(
        target=process_request,
        args=(f"user-{i}",),
        name=f"Thread-{i}"
    )
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

Java

ThreadLocalExample.java

public class ThreadLocalExample {
    private static ThreadLocal<String> userContext = new ThreadLocal<>();
    private static ThreadLocal<Integer> requestId = new ThreadLocal<>();

    public static void setUser(String userId) {
        userContext.set(userId);
        requestId.set((int) Thread.currentThread().getId());
    }

    public static String getUser() {
        return userContext.get();
    }

    public static void processRequest(String userId) {
        setUser(userId);
        System.out.println("Thread " + Thread.currentThread().getName() +
            ": Processing request for user " + getUser());
    }

    public static void main(String[] args) throws InterruptedException {
        Thread[] threads = new Thread[3];

        for (int i = 0; i < 3; i++) {
            final int userId = i;
            threads[i] = new Thread(() -> {
                processRequest("user-" + userId);
            }, "Thread-" + i);
            threads[i].start();
        }

        for (Thread thread : threads) {
            thread.join();
        }
    }
}

Memory Leaks

ThreadLocal variables can cause memory leaks if threads are pooled (like in thread pools). Always clean up ThreadLocal values when done, or use remove() method.

---

Memory Visibility: Volatile (Java)

The volatile keyword in Java ensures visibility of changes across threads and prevents certain compiler optimizations.

Visual: Volatile Visibility

Example: Volatile Flag

Java

VolatileExample.java

public class VolatileExample {
    private volatile boolean flag = false;  // Volatile ensures visibility

    public void setFlag() {
        flag = true;  // Write is immediately visible to all threads
    }

    public boolean getFlag() {
        return flag;  // Read sees the latest value
    }

    public static void main(String[] args) throws InterruptedException {
        VolatileExample example = new VolatileExample();

        // Reader thread
        Thread reader = new Thread(() -> {
            while (!example.getFlag()) {
                // Busy wait - will see flag change immediately
            }
            System.out.println("Flag is now true!");
        });

        reader.start();
        Thread.sleep(1000);

        // Writer thread
        example.setFlag();  // This change is immediately visible

        reader.join();
    }
}

Volatile vs synchronized

• volatile: Ensures visibility, but doesn’t provide atomicity for compound operations
• synchronized: Provides both visibility AND atomicity
• Use volatile for simple flags or single variable updates
• Use synchronized for compound operations or when you need mutual exclusion

---

Comparison Table

Primitive	Use Case	Java	Python	When to Use
Lock/Mutex	Mutual exclusion	ReentrantLock	threading.Lock	Protect critical sections
Reentrant Lock	Recursive locking	ReentrantLock	threading.RLock	Same thread needs lock multiple times
Read-Write Lock	Read-heavy workloads	ReentrantReadWriteLock	Limited support	Many readers, few writers
Semaphore	Limit concurrent access	Semaphore	threading.Semaphore	Rate limiting, resource pools
Condition	Thread coordination	Condition	threading.Condition	Wait for conditions (Producer-Consumer)
Barrier	Synchronization point	CyclicBarrier	threading.Barrier	All threads wait, then proceed together
Latch	One-time sync	CountDownLatch	N/A	Wait for N events to complete
Thread-Local	Per-thread variables	ThreadLocal	threading.local()	Avoid parameter passing
Volatile	Memory visibility	volatile	N/A	Simple flags, single variable updates

---

Practice Problems

Easy: Thread-Safe Counter

Design a thread-safe counter that supports increment(), decrement(), and get() operations.

Solution

Python

import threading

class ThreadSafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()

    def increment(self):
        with self._lock:
            self._value += 1

    def decrement(self):
        with self._lock:
            self._value -= 1

    def get(self):
        with self._lock:
            return self._value

Java

import java.util.concurrent.locks.ReentrantLock;

public class ThreadSafeCounter {
    private int value = 0;
    private final ReentrantLock lock = new ReentrantLock();

    public void increment() {
        lock.lock();
        try {
            value++;
        } finally {
            lock.unlock();
        }
    }

    public void decrement() {
        lock.lock();
        try {
            value--;
        } finally {
            lock.unlock();
        }
    }

    public int get() {
        lock.lock();
        try {
            return value;
        } finally {
            lock.unlock();
        }
    }
}

Medium: Rate Limiter with Semaphore

Implement a rate limiter that allows N requests per second using semaphores.

Solution

See the Rate Limiter example in the Semaphores section above.

---

Interview Questions

Q1: “What’s the difference between synchronized and ReentrantLock?”

Answer:

• synchronized: Implicit locking keyword, simpler syntax, slightly faster, but less flexible
• ReentrantLock: Explicit lock class, more features (fairness, tryLock, interruptible), more control, slightly slower
• When to use: Use synchronized for simple cases, ReentrantLock when you need advanced features

Q2: “When would you use a semaphore vs a lock?”

Answer:

• Lock (Mutex): Allows only ONE thread at a time (mutual exclusion)
• Semaphore: Allows N threads at a time (resource limiting)
• Use semaphore: When you need to limit concurrent access to a resource (e.g., database connections, API rate limiting)
• Use lock: When you need exclusive access to shared data

Q3: “Explain the difference between notify() and notifyAll().”

Answer:

• notify(): Wakes up ONE waiting thread (unpredictable which one)
• notifyAll(): Wakes up ALL waiting threads (they compete for the lock)
• Use notify(): When only one thread can proceed (e.g., single consumer)
• Use notifyAll(): When multiple threads might proceed (e.g., multiple consumers, complex conditions)

Q4: “What is ThreadLocal and when would you use it?”

Answer:

• ThreadLocal: Provides each thread with its own independent copy of a variable
• Use cases: Request context in web servers, avoiding parameter passing, per-thread configuration
• Benefits: No synchronization needed, thread-safe by design
• Caution: Can cause memory leaks in thread pools if not cleaned up

---

Key Takeaways

Remember

• Locks: Provide mutual exclusion (one thread at a time)
• Semaphores: Control resource access (N threads at a time)
• Condition Variables: Enable thread coordination and waiting
• Read-Write Locks: Optimize for read-heavy workloads
• Thread-Local: Per-thread variables (no synchronization needed)
• Volatile: Ensures memory visibility (Java)
• Choose wisely: Match the primitive to your use case!

---

Next Steps

Continue learning concurrency patterns:

• Producer-Consumer Pattern - Apply condition variables in a real pattern
• Thread Pools & Executors - Efficient resource management

Mastering synchronization primitives is essential for building thread-safe systems! 🔒