Concurrency & Parallelism Explained — Threads, Mutexes & Deadlock Prevention

import threading
import time

def task(name, duration):
    print(f"[{name}] Starting, will run {duration}s")
    time.sleep(duration)
    print(f"[{name}] Done")

# Sequential execution
print("=== Sequential (no concurrency) ===")
start = time.time()
task("A", 2)
task("B", 2)
print(f"Total: {time.time() - start:.1f}s\n")

# Concurrent execution (single-core illusion)
print("=== Concurrent (threading) ===")
start = time.time()
t1 = threading.Thread(target=task, args=("A", 2))
t2 = threading.Thread(target=task, args=("B", 2))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"Total: {time.time() - start:.1f}s")

import threading

counter = 0
lock = threading.Lock()

def increment_without_lock():
    global counter
    for _ in range(100000):
        counter += 1  # Not atomic!

def increment_with_lock():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

# Without lock — race condition
counter = 0
t1 = threading.Thread(target=increment_without_lock)
t2 = threading.Thread(target=increment_without_lock)
t1.start(); t2.start()
t1.join(); t2.join()
print(f"Without lock: {counter} (expected 200000)")

# With lock — correct
counter = 0
t1 = threading.Thread(target=increment_with_lock)
t2 = threading.Thread(target=increment_with_lock)
t1.start(); t2.start()
t1.join(); t2.join()
print(f"With lock: {counter} (expected 200000)")

import threading
import time

pool = threading.Semaphore(3)  # Allow 3 concurrent accesses

def access_database(thread_id):
    with pool:
        print(f"[Thread {thread_id}] Accessing database...")
        time.sleep(1)
        print(f"[Thread {thread_id}] Done")

threads = []
for i in range(10):
    t = threading.Thread(target=access_database, args=(i,))
    threads.append(t)

print("Starting 10 threads with pool limit of 3")
for t in threads:
    t.start()
for t in threads:
    t.join()

def is_safe_state(available, allocated, max_need):
    n_processes = len(allocated)
    n_resources = len(available)

    need = [[max_need[i][j] - allocated[i][j]
             for j in range(n_resources)]
            for i in range(n_processes)]

    finished = [False] * n_processes
    safe_sequence = []
    work = available.copy()

    while len(safe_sequence) < n_processes:
        found = False
        for i in range(n_processes):
            if not finished[i] and all(need[i][j] <= work[j]
                                       for j in range(n_resources)):
                for j in range(n_resources):
                    work[j] += allocated[i][j]
                finished[i] = True
                safe_sequence.append(i)
                found = True
        if not found:
            return False, []

    return True, safe_sequence

# Example: 5 processes, 3 resources
available = [3, 3, 2]
allocated = [
    [0, 1, 0],
    [2, 0, 0],
    [3, 0, 2],
    [2, 1, 1],
    [0, 0, 2],
]
max_need = [
    [7, 5, 3],
    [3, 2, 2],
    [9, 0, 2],
    [2, 2, 2],
    [4, 3, 3],
]

safe, sequence = is_safe_state(available, allocated, max_need)
print(f"Safe state: {safe}")
if safe:
    print(f"Safe sequence: P{', P'.join(map(str, sequence))}")

import threading
import time

lock_a = threading.Lock()
lock_b = threading.Lock()

def thread_1():
    with lock_a:
        time.sleep(0.1)
        with lock_b:
            print("Thread 1: got both locks")

def thread_2():
    with lock_a:  # Same order as thread_1
        time.sleep(0.1)
        with lock_b:
            print("Thread 2: got both locks")

t1 = threading.Thread(target=thread_1)
t2 = threading.Thread(target=thread_2)
t1.start(); t2.start()
t1.join(); t2.join()
print("No deadlock — lock ordering works!")

import threading
import time
import random

class Philosopher(threading.Thread):
    def __init__(self, index, left_fork, right_fork):
        super().__init__()
        self.index = index
        self.left_fork = left_fork
        self.right_fork = right_fork

    def run(self):
        for _ in range(3):
            self.think()
            self.eat()

    def think(self):
        print(f"Philosopher {self.index} thinking...")
        time.sleep(random.random())

    def eat(self):
        # Acquire forks in order of index to prevent deadlock
        first = self.left_fork if self.index % 2 == 0 else self.right_fork
        second = self.right_fork if self.index % 2 == 0 else self.left_fork

        with first:
            with second:
                print(f"Philosopher {self.index} eating! 🍝")
                time.sleep(random.random())

forks = [threading.Lock() for _ in range(5)]
philosophers = [
    Philosopher(i, forks[i], forks[(i + 1) % 5])
    for i in range(5)
]

for p in philosophers:
    p.start()
for p in philosophers:
    p.join()
print("All philosophers finished!")

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("Worker %d starting\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }

    wg.Wait()
    fmt.Println("All workers completed")
}

import asyncio

async def fetch_data(url, delay):
    print(f"Fetching {url}...")
    await asyncio.sleep(delay)  # Yield control
    data = f"Data from {url}"
    print(f"Got {len(data)} bytes from {url}")
    return data

async def main():
    tasks = [
        fetch_data("https://api.example.com/users", 2),
        fetch_data("https://api.example.com/posts", 1),
        fetch_data("https://api.example.com/comments", 3),
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(f"Result: {result}")

asyncio.run(main())

import threading
import atomic  # Requires: pip install atomicwrites

# Python's threading provides atomic increment via锁
# but true lock-free structures use CAS
class LockFreeCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()

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

    def get(self):
        return self._value