Concurrency & Parallelism Deep Dive: Multithreading, Async, and Event Loops Explained

"Concurrency is about dealing with lots of things at once.
 Parallelism is about doing lots of things at once.
 Concurrency is about structure.
 Parallelism is about execution."
                                          — Rob Pike

Concurrency:                    Parallelism:
One person juggling two tasks   Two people each doing one task

   Task A ━━━━━━╸               Task A ━━━━━━━━━━━━
                 ╻               Task B ━━━━━━━━━━━━
   Task B ━━━━━━╸
                 ╻               CPU Core 1 ━━━━━━━━
   Task A ━━━━━━╸               CPU Core 2 ━━━━━━━━
                 ╻
   CPU Core 1 ━━━━━━━━
   (time-slicing)               (physically simultaneous)

User threads:   T1    T2    T3    T4
                │      │      │      │
Kernel threads: KT1   KT2   KT3   KT4
                │      │      │      │
CPU cores:     Core1  Core2  Core1  Core2

User threads:   T1  T2  T3  T4  T5  T6
                └──┬──┘  └──┬──┘  └──┬──┘
                   │        │        │
Kernel threads:  KT1      KT1      KT1
                   │
CPU cores:       Core1

User threads:   G1  G2  G3  G4  G5  G6  G7  G8
                └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘
                   │       │       │       │
Kernel threads:  KT1     KT2     KT3     KT4
                   │       │       │       │
CPU cores:       Core1   Core2   Core3   Core4

┌─────────────────┬──────────────┬──────────────┬──────────────┐
│ Category        │ 1:1          │ N:1          │ M:N          │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│ Thread cost     │ ~1MB stack   │ ~KB stack    │ ~KB stack    │
│ Creation speed  │ Slow (~ms)   │ Fast (~us)   │ Fast (~us)   │
│ Context switch  │ Expensive    │ Cheap (user) │ Cheap (user) │
│ Multi-core      │ Yes          │ No           │ Yes          │
│ I/O blocking    │ Thread only  │ All blocked  │ Thread only  │
│ Max threads     │ ~thousands   │ ~millions    │ ~millions    │
│ Complexity      │ Low          │ Medium       │ High         │
│ Examples        │ Java, C++,   │ Early Ruby   │ Go, Erlang,  │
│                 │ Rust         │              │ Java 21+     │
└─────────────────┴──────────────┴──────────────┴──────────────┘

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    results := make(chan string, 10)

    urls := []string{
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    }

    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            results <- fmt.Sprintf("Fetched: %s", u)
        }(url)
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    for result := range results {
        fmt.Println(result)
    }
}

import java.util.concurrent.*;
import java.util.List;

public class VirtualThreadExample {
    public static void main(String[] args) throws Exception {
        // Million concurrent tasks with Virtual Threads
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = List.of(
                executor.submit(() -> fetchURL("https://api.example.com/users")),
                executor.submit(() -> fetchURL("https://api.example.com/orders")),
                executor.submit(() -> fetchURL("https://api.example.com/products"))
            );

            for (var future : futures) {
                System.out.println(future.get());
            }
        }

        // Structured Concurrency (Preview)
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            var user = scope.fork(() -> fetchUser(1));
            var order = scope.fork(() -> fetchOrder(1));

            scope.join();
            scope.throwIfFailed();

            System.out.println(user.get() + ", " + order.get());
        }
    }
}

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

        for url, result in zip(urls, results):
            print(f"Fetched {url}: {len(result)} bytes")

asyncio.run(main())

use tokio;
use reqwest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let urls = vec![
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    ];

    let mut handles = vec![];

    for url in urls {
        let handle = tokio::spawn(async move {
            let resp = reqwest::get(url).await?;
            let body = resp.text().await?;
            Ok::<String, reqwest::Error>(body)
        });
        handles.push(handle);
    }

    for handle in handles {
        match handle.await? {
            Ok(body) => println!("Fetched: {} bytes", body.len()),
            Err(e) => eprintln!("Error: {}", e),
        }
    }

    Ok(())
}

async function fetchAll() {
    const urls = [
        'https://api.example.com/users',
        'https://api.example.com/orders',
        'https://api.example.com/products',
    ];

    // Concurrent execution with Promise.all
    const results = await Promise.all(
        urls.map(url => fetch(url).then(r => r.json()))
    );

    // Promise.allSettled allows failures
    const settled = await Promise.allSettled(
        urls.map(url => fetch(url).then(r => r.json()))
    );

    settled.forEach((result, i) => {
        if (result.status === 'fulfilled') {
            console.log(`Success: ${urls[i]}`);
        } else {
            console.log(`Failed: ${urls[i]} - ${result.reason}`);
        }
    });
}

%% Erlang processes (lightweight actors)
-module(counter).
-export([start/0, increment/1, get/1]).

start() ->
    spawn(fun() -> loop(0) end).

loop(Count) ->
    receive
        {increment, From} ->
            From ! {ok, Count + 1},
            loop(Count + 1);
        {get, From} ->
            From ! {value, Count},
            loop(Count);
        stop ->
            ok
    end.

increment(Pid) ->
    Pid ! {increment, self()},
    receive {ok, NewCount} -> NewCount end.

get(Pid) ->
    Pid ! {get, self()},
    receive {value, Count} -> Count end.

┌─────────────┬─────────────┬───────────────┬──────────────┬──────────────┐
│ Language    │ Model       │ Unit          │ Scheduling   │ Key Feature  │
├─────────────┼─────────────┼───────────────┼──────────────┼──────────────┤
│ Go          │ CSP (M:N)   │ Goroutine     │ Runtime      │ Channels     │
│ Java 21+    │ M:N         │ Virtual Thread│ JVM          │ Backward compat│
│ Python      │ Event loop  │ Coroutine     │ asyncio      │ GIL limit    │
│ Rust        │ async/await │ Future/Task   │ tokio etc.   │ Zero-cost    │
│ JavaScript  │ Event loop  │ Promise       │ libuv/V8     │ Single thread│
│ Erlang      │ Actor (M:N) │ Process       │ BEAM VM      │ Fault tolerant│
│ Kotlin      │ Coroutines  │ Coroutine     │ Dispatcher   │ Structured   │
└─────────────┴─────────────┴───────────────┴──────────────┴──────────────┘

┌───────────────────────────┐
│         timers            │  ← setTimeout, setInterval
│   (execute timer cbs)     │
├───────────────────────────┤
│     pending callbacks     │  ← system error callbacks
│   (deferred I/O cbs)      │
├───────────────────────────┤
│       idle, prepare       │  ← internal use
├───────────────────────────┤
│          poll             │  ← I/O callbacks (most)
│   (wait for new I/O)      │
├───────────────────────────┤
│          check            │  ← setImmediate callbacks
├───────────────────────────┤
│     close callbacks       │  ← socket.on('close')
└───────────────────────────┘
   Between each phase: microtask queue is drained
   (Promise.then, queueMicrotask, process.nextTick)

console.log('1: sync');

setTimeout(() => console.log('2: macro (setTimeout)'), 0);

Promise.resolve().then(() => console.log('3: micro (Promise)'));

queueMicrotask(() => console.log('4: micro (queueMicrotask)'));

console.log('5: sync');

// Output order:
// 1: sync
// 5: sync
// 3: micro (Promise)
// 4: micro (queueMicrotask)
// 2: macro (setTimeout)

┌──────────────────────────────────────────┐
│              Node.js                     │
│  ┌─────────────────────────────────┐     │
│  │     JavaScript (V8 Engine)      │     │
│  │  ┌───────────┐  ┌────────────┐  │     │
│  │  │  Async    │  │ Event      │  │     │
│  │  │  APIs     │  │ Queue      │  │     │
│  │  └─────┬─────┘  └──────┬─────┘  │     │
│  └────────┼────────────────┼────────┘     │
│           │   Event Loop   │              │
│  ┌────────▼────────────────▼────────┐     │
│  │           libuv                   │     │
│  │  ┌──────────┐  ┌──────────────┐  │     │
│  │  │ Thread   │  │ epoll/kqueue │  │     │
│  │  │ Pool     │  │ /IOCP        │  │     │
│  │  │ (4 thds) │  │ (non-block)  │  │     │
│  │  └──────────┘  └──────────────┘  │     │
│  └───────────────────────────────────┘     │
│           │                │               │
│      File I/O          Network I/O         │
│      DNS lookup        TCP/UDP             │
│      Crypto            HTTP                │
└──────────────────────────────────────────┘

import asyncio

async def process_item(item: str) -> str:
    await asyncio.sleep(1)  # I/O simulation
    return f"Processed: {item}"

async def main():
    items = ["a", "b", "c", "d", "e"]

    # Concurrent execution
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    print(results)  # All complete in 1 second

    # Timeout
    try:
        result = await asyncio.wait_for(
            process_item("slow"),
            timeout=0.5
        )
    except asyncio.TimeoutError:
        print("Timeout!")

    # Semaphore to limit concurrency
    semaphore = asyncio.Semaphore(3)

    async def limited_process(item):
        async with semaphore:
            return await process_item(item)

    results = await asyncio.gather(*[limited_process(i) for i in items])

asyncio.run(main())

import kotlinx.coroutines.*

suspend fun fetchUser(id: Int): User {
    delay(1000)
    return User(id, "User $id")
}

fun main() = runBlocking {
    // Structured concurrency: parent cancel propagates to children
    coroutineScope {
        val user = async { fetchUser(1) }
        val orders = async { fetchOrders(1) }

        println("User: ${user.await()}")
        println("Orders: ${orders.await()}")
    } // Waits for all child coroutines

    // SupervisorScope: child failure doesn't affect siblings
    supervisorScope {
        val job1 = launch {
            delay(100)
            throw RuntimeException("Failed!")
        }
        val job2 = launch {
            delay(200)
            println("Job2 completed") // Still runs
        }
    }
}

Traditional concurrency:        Structured concurrency:
┌─────────────┐                ┌─────────────┐
│    Parent    │                │    Parent    │
│  ┌───┐ ┌───┐│                │  ┌───┐ ┌───┐│
│  │ A │ │ B ││                │  │ A │ │ B ││
│  └─┬─┘ └─┬─┘│                │  └─┬─┘ └─┬─┘│
└────┼─────┼───┘                └────┼─────┼───┘
     │     │ ← Who manages?          │     │ ← Parent manages
     ▼     ▼   Leak possible!        ▼     ▼   Auto cleanup!

Rules:
1. All concurrent work starts within a scope
2. Scope waits for all child tasks to complete
3. Parent cancellation cancels all children
4. Child errors propagate to parent

┌──────────┐     message     ┌──────────┐
│  Actor A │ ──────────────→ │  Actor B │
│          │                 │          │
│ State: x │     message     │ State: y │
│ Mailbox  │ ←────────────── │ Mailbox  │
└──────────┘                 └──────────┘
     │
     │ message
     ▼
┌──────────┐
│  Actor C │
│ State: z │
│ Mailbox  │
└──────────┘

Key: No shared state, no locks, only message passing

                  ┌──────────────┐
                  │  Supervisor  │
                  │ (one_for_one)│
                  └──────┬───────┘
                    ┌────┼────┐
                    ▼    ▼    ▼
              ┌─────┐ ┌─────┐ ┌─────┐
              │ W1  │ │ W2  │ │ W3  │
              └─────┘ └─────┘ └─────┘

When W2 crashes:
1. Supervisor detects it
2. Only W2 restarts (one_for_one strategy)
3. W1, W3 are unaffected
4. If restarted 5+ times in 5 minutes, Supervisor also terminates

Good fit:
- Managing millions of independent states (chat rooms, IoT devices)
- Distributed systems (inter-node communication)
- Fault-tolerance required (telecom, finance)
- Real-time systems (game servers, trading)

Poor fit:
- Simple CRUD applications
- Strong consistency required
- Transaction processing is key
- Order guarantees critical in pipelines

// Go - Mutex
type SafeMap struct {
    mu   sync.Mutex
    data map[string]int
}

func (m *SafeMap) Set(key string, val int) {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.data[key] = val
}

func (m *SafeMap) Get(key string) (int, bool) {
    m.mu.Lock()
    defer m.mu.Unlock()
    val, ok := m.data[key]
    return val, ok
}

// Java - ReentrantLock
import java.util.concurrent.locks.ReentrantLock;

public class SafeCounter {
    private final ReentrantLock lock = new ReentrantLock();
    private int count = 0;

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

type Cache struct {
    mu   sync.RWMutex
    data map[string]string
}

// Multiple goroutines can read simultaneously
func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}

// Write requires exclusive lock
func (c *Cache) Set(key, val string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = val
}

import asyncio

async def worker(name, semaphore):
    async with semaphore:
        print(f"{name} acquired semaphore")
        await asyncio.sleep(1)
        print(f"{name} released semaphore")

async def main():
    semaphore = asyncio.Semaphore(3)  # max 3 concurrent
    tasks = [worker(f"Worker-{i}", semaphore) for i in range(10)]
    await asyncio.gather(*tasks)

asyncio.run(main())

type BoundedQueue struct {
    mu       sync.Mutex
    notEmpty *sync.Cond
    notFull  *sync.Cond
    items    []interface{}
    maxSize  int
}

func NewBoundedQueue(maxSize int) *BoundedQueue {
    q := &BoundedQueue{maxSize: maxSize}
    q.notEmpty = sync.NewCond(&q.mu)
    q.notFull = sync.NewCond(&q.mu)
    return q
}

func (q *BoundedQueue) Put(item interface{}) {
    q.mu.Lock()
    defer q.mu.Unlock()

    for len(q.items) == q.maxSize {
        q.notFull.Wait() // wait until space available
    }
    q.items = append(q.items, item)
    q.notEmpty.Signal() // wake consumer
}

func (q *BoundedQueue) Take() interface{} {
    q.mu.Lock()
    defer q.mu.Unlock()

    for len(q.items) == 0 {
        q.notEmpty.Wait() // wait until item available
    }
    item := q.items[0]
    q.items = q.items[1:]
    q.notFull.Signal() // wake producer
    return item
}

import "sync/atomic"

type LockFreeCounter struct {
    value int64
}

func (c *LockFreeCounter) Increment() int64 {
    for {
        old := atomic.LoadInt64(&c.value)
        new := old + 1
        if atomic.CompareAndSwapInt64(&c.value, old, new) {
            return new
        }
        // CAS failed, retry (another goroutine changed it first)
    }
}

func (c *LockFreeCounter) Get() int64 {
    return atomic.LoadInt64(&c.value)
}

Thread A: reads value A
                          Thread B: changes A to B
                          Thread B: changes B back to A
Thread A: CAS(A -> C) succeeds! (but value changed in between)

Solutions:
- Add version numbers (A,v1 -> B,v2 -> A,v3)
- Tagged pointers (AtomicStampedReference in Java)
- Hazard Pointers

Lock-free is good for:
- Very high contention scenarios
- Real-time systems (avoiding priority inversion)
- Simple data structures (counters, stacks, queues)

Avoid lock-free when (most cases):
- Regular applications (Mutex is sufficient)
- Complex data structures (trees, graphs)
- Correctness verification is difficult
- Team is not experienced with lock-free

"If a Mutex works, use a Mutex."

// Bug: multiple goroutines modifying counter simultaneously
var counter int

func buggyIncrement() {
    for i := 0; i < 1000; i++ {
        go func() {
            counter++ // read-modify-write is not atomic
        }()
    }
}

// Fix: use atomic or mutex
var safeCounter int64

func safeIncrement() {
    for i := 0; i < 1000; i++ {
        go func() {
            atomic.AddInt64(&safeCounter, 1)
        }()
    }
}

Four necessary conditions (all must hold for deadlock):
1. Mutual Exclusion: resource used by only one at a time
2. Hold and Wait: holding resource while waiting for another
3. No Preemption: cannot forcibly take held resource
4. Circular Wait: A waits for B, B waits for C, C waits for A

// Deadlock example
var mu1, mu2 sync.Mutex

func goroutine1() {
    mu1.Lock()
    defer mu1.Unlock()
    time.Sleep(time.Millisecond)
    mu2.Lock()       // waiting for mu2 (held by goroutine2)
    defer mu2.Unlock()
}

func goroutine2() {
    mu2.Lock()
    defer mu2.Unlock()
    time.Sleep(time.Millisecond)
    mu1.Lock()       // waiting for mu1 (held by goroutine1)
    defer mu1.Unlock()
}

// Fix: always acquire locks in the same order
func fixed1() {
    mu1.Lock()
    defer mu1.Unlock()
    mu2.Lock()
    defer mu2.Unlock()
}

func fixed2() {
    mu1.Lock()       // mu1 first (same order)
    defer mu1.Unlock()
    mu2.Lock()
    defer mu2.Unlock()
}

A and B meet in a narrow corridor:
A: steps left  <-> B: steps right (same direction!)
A: steps right <-> B: steps left  (same direction again!)
... (infinite loop, nobody progresses)

vs Deadlock: doing nothing
vs Livelock: constantly moving but no progress

1997 Mars Pathfinder incident:

Low-priority task holds mutex
   |
Medium-priority task preempts Low (unrelated to mutex)
   |
High-priority task waits for mutex... forever!
(Low needs to run to release mutex, but Medium keeps preempting)

Solution: Priority Inheritance Protocol
Low temporarily inherits High's priority while holding the mutex

# Go: Race Detector
go test -race ./...
go run -race main.go

# Java: Thread Sanitizer
java -XX:+UseThreadSanitizer MyApp

# Rust: Compiler detects at compile time!
# Send/Sync trait violations = compile error

# Python: asyncio debug mode
PYTHONASYNCIODEBUG=1 python app.py

# General: Helgrind (Valgrind tool)
valgrind --tool=helgrind ./program

func producerConsumer() {
    ch := make(chan int, 100) // buffered channel

    // Producers
    var prodWg sync.WaitGroup
    for i := 0; i < 3; i++ {
        prodWg.Add(1)
        go func(id int) {
            defer prodWg.Done()
            for j := 0; j < 10; j++ {
                ch <- id*100 + j
            }
        }(i)
    }

    go func() {
        prodWg.Wait()
        close(ch)
    }()

    // Consumers
    var consWg sync.WaitGroup
    for i := 0; i < 2; i++ {
        consWg.Add(1)
        go func(id int) {
            defer consWg.Done()
            for item := range ch {
                fmt.Printf("Consumer %d: %d\n", id, item)
            }
        }(i)
    }

    consWg.Wait()
}

import asyncio
from asyncio import Queue

async def worker(name: str, queue: Queue):
    while True:
        item = await queue.get()
        try:
            await process(item)
        finally:
            queue.task_done()

async def main():
    queue = Queue(maxsize=100)

    # Start 5 workers
    workers = [
        asyncio.create_task(worker(f"worker-{i}", queue))
        for i in range(5)
    ]

    # Add tasks
    for item in range(50):
        await queue.put(item)

    await queue.join()

    for w in workers:
        w.cancel()

asyncio.run(main())

Traditional thread pool:        Work Stealing:
┌─────────────┐                ┌─────────────┐
│ Shared queue │                │ Worker 1    │
│ [T1|T2|T3]  │                │ local:[T1]  │
│             │                ├─────────────┤
│ All workers  │                │ Worker 2    │
│ contend!    │                │ local:[T2]  │  ← steals from W1 when empty
│             │                ├─────────────┤
└─────────────┘                │ Worker 3    │
                               │ local:[T3]  │
                               └─────────────┘

Benefits: Better cache locality, reduced contention
Used by: Java ForkJoinPool, Go runtime, Tokio (Rust)

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

P = fraction that can be parallelized
N = number of processors

Example: 95% parallelizable program on 64 cores
Speedup = 1 / (0.05 + 0.95/64) = 1 / 0.0648 = 15.4x

Key insight: No matter how many cores, the serial portion (5%)
is the bottleneck! Maximum theoretical speedup = 1/0.05 = 20x

Speedup = N - alpha * (N - 1)

alpha = fraction of serial work
N = number of processors

Key difference:
- Amdahl: "Fixed problem size, add more cores?"
- Gustafson: "Fixed cores, grow the problem?"

In reality, as data grows, the parallel portion also grows.
Gustafson is more optimistic and realistic.

Context Switch Cost Comparison:
┌───────────────────────────┬──────────────┐
│ Switch Type               │ Approx Cost  │
├───────────────────────────┼──────────────┤
│ Function call             │ ~1-5 ns      │
│ Goroutine switch          │ ~100-200 ns  │
│ Coroutine switch (Python) │ ~1-5 us      │
│ Thread switch (same proc) │ ~1-10 us     │
│ Process switch            │ ~10-50 us    │
│ VM switch                 │ ~100+ us     │
└───────────────────────────┴──────────────┘

Problem: Different variables on the same cache line

CPU Core 1          CPU Core 2
    |                   |
+---v-------------------v---+
| Cache Line (64 bytes)     |
| [counter_A] [counter_B]   |  <- Different variables but
|                           |     same cache line!
+---------------------------+

Core 1 modifies counter_A -> invalidates cache line -> Core 2 must reload
Core 2 modifies counter_B -> invalidates cache line -> Core 1 must reload

Fix: Padding to separate cache lines

// Preventing False Sharing in Go
type PaddedCounter struct {
    value int64
    _pad  [56]byte // 64-byte cache line padding
}

type Counters struct {
    c1 PaddedCounter
    c2 PaddedCounter
}