Concurrency: Multitasking, Threads and Processes

import threading

print(threading.active_count())          # 1 -> only the main thread exists so far
print(threading.current_thread().name)   # MainThread
print(threading.current_thread() is threading.main_thread())   # True

1
MainThread
True

import time

def greet(name):
    time.sleep(0.1)
    print(f"Hello, {name}!")

t = threading.Thread(target=greet, args=("Ada",))   # target + args
t.start()   # begins running greet() on a new OS thread
t.join()      # block here until t finishes
print("main thread continues")   # only reached after "Hello, Ada!" has printed

Hello, Ada!
main thread continues

class GreeterThread(threading.Thread):
    def __init__(self, name):
        super().__init__()   # required - sets up the Thread machinery
        self.person = name

    def run(self):   # override run() instead of passing target=
        time.sleep(0.1)
        print(f"Hello, {self.person}!")

t = GreeterThread("Grace")
t.start()   # calls run() on a new thread
t.join()
print("main thread continues")

Hello, Grace!
main thread continues

from concurrent.futures import ThreadPoolExecutor

def slow_greet(name):
    time.sleep(0.1)
    return f"Hello, {name}!"

with ThreadPoolExecutor(max_workers=2) as pool:   # a managed pool of (at most 2) worker threads
    results = pool.map(slow_greet, ["Ada", "Grace", "Alan"])
    print(list(results))   # ['Hello, Ada!', 'Hello, Grace!', 'Hello, Alan!']

['Hello, Ada!', 'Hello, Grace!', 'Hello, Alan!']

def worker():
    time.sleep(0.2)

t = threading.Thread(target=worker)
t.start()
try:
    t.start()   # a Thread can only be started ONCE
except RuntimeError as e:
    print("RuntimeError:", e)   # threads can only be started once
t.join()

RuntimeError: threads can only be started once

def worker_that_prints():
    time.sleep(0.2)
    print("worker done")

t1 = threading.Thread(target=worker_that_prints)
t1.start()
print("without join(): this often prints FIRST, before the worker finishes")

t2 = threading.Thread(target=worker_that_prints)
t2.start()
t2.join()   # blocks the main thread here until t2 actually finishes
print("with join(): this only prints AFTER the worker is done")

without join(): this often prints FIRST, before the worker finishes
worker done
worker done
with join(): this only prints AFTER the worker is done

def slow_worker():
    time.sleep(0.3)

t1 = threading.Thread(target=slow_worker)                       # default name, auto-generated
t2 = threading.Thread(target=slow_worker, name="Downloader")   # explicit name

print(t1.name)          # Thread-N (slow_worker) -> auto-generated, includes the target's name
print(t2.name)          # Downloader
print(t1.ident)           # None -> not started yet, no OS-level identifier assigned
print(t1.is_alive())    # False -> not started yet

t1.start()
t2.start()

print(t1.ident is not None)   # True -> now it has a real thread identifier
print(t1.is_alive())                # True -> currently running

print(threading.active_count())                            # 3 -> MainThread + t1 + t2
print([th.name for th in threading.enumerate()])   # every currently alive thread, by name

t1.join()
t2.join()

print(t1.is_alive())                    # False -> finished
print(threading.active_count())   # 1 -> back down to just MainThread

Thread-6 (slow_worker)
Downloader
None
False
True
True
3
['MainThread', 'Thread-6 (slow_worker)', 'Downloader']
False
1

def worker():
    time.sleep(0.1)

t = threading.Thread(target=worker)
print(t.daemon)   # False -> non-daemonic by default

False

t.start()
try:
    t.daemon = True   # too late - must be set BEFORE start()
except RuntimeError as e:
    print("RuntimeError:", e)   # cannot set daemon status of active thread
t.join()

RuntimeError: cannot set daemon status of active thread

import subprocess, tempfile, os

work_dir = tempfile.mkdtemp()

daemon_script = os.path.join(work_dir, "daemon_demo.py")
with open(daemon_script, "w") as f:
    f.write(
        "import threading, time\n"
        "def background_work():\n"
        "    time.sleep(1)\n"
        "    print('background finished')\n"
        "t = threading.Thread(target=background_work, daemon=True)\n"
        "t.start()\n"
        "print('main exiting')\n"
    )

start = time.time()
result = subprocess.run(["python3", daemon_script], capture_output=True, text=True)
elapsed = time.time() - start

print(result.stdout.strip())   # main exiting -> the ONLY output; "background finished" never runs
print(elapsed < 1)                    # True -> the process exited immediately, didn't wait a full second

main exiting
True

nondaemon_script = os.path.join(work_dir, "nondaemon_demo.py")
with open(nondaemon_script, "w") as f:
    f.write(
        "import threading, time\n"
        "def background_work():\n"
        "    time.sleep(1)\n"
        "    print('background finished')\n"
        "t = threading.Thread(target=background_work)\n"   # daemon defaults to False
        "t.start()\n"
        "print('main reaching its end')\n"
    )

start = time.time()
result = subprocess.run(["python3", nondaemon_script], capture_output=True, text=True)
elapsed = time.time() - start

print(result.stdout.strip())   # main reaching its end \n background finished -> both lines this time
print(elapsed >= 1)                    # True -> the process waited for the non-daemon thread

main reaching its end
background finished
True

import multiprocessing

counter = 0

def bump_in_thread():
    global counter
    counter += 1

def bump_in_process():
    global counter
    counter += 1
    print("counter INSIDE the child process:", counter)   # sees its own separate copy

if __name__ == "__main__":
    t = threading.Thread(target=bump_in_thread)
    t.start()
    t.join()
    print("counter after a THREAD modifies it:", counter)     # 1 -> shared memory, the change is visible

    p = multiprocessing.Process(target=bump_in_process)
    p.start()
    p.join()
    print("counter after a PROCESS modifies it:", counter)   # still 1 -> separate memory, no effect here

counter after a THREAD modifies it: 1
counter after a PROCESS modifies it: 1

def compute_square(n, result_queue):
    result_queue.put(n * n)   # the way to get a value back - shared memory doesn't work here

if __name__ == "__main__":
    q = multiprocessing.Queue()
    p = multiprocessing.Process(target=compute_square, args=(7, q))
    p.start()
    p.join()
    print(q.get())   # 49 -> retrieved across the process boundary

49

def cpu_bound(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

N = 8_000_000

start = time.time()
cpu_bound(N)
cpu_bound(N)
sequential_time = time.time() - start

start = time.time()
t1 = threading.Thread(target=cpu_bound, args=(N,))
t2 = threading.Thread(target=cpu_bound, args=(N,))
t1.start(); t2.start()
t1.join(); t2.join()
threaded_time = time.time() - start

print(f"sequential: {sequential_time:.2f}s, threaded: {threaded_time:.2f}s")   # roughly equal
print(threaded_time >= sequential_time * 0.9)   # True -> no real speedup - the GIL serializes CPU-bound work

sequential: 0.85s, threaded: 0.86s
True

if __name__ == "__main__":
    start = time.time()
    p1 = multiprocessing.Process(target=cpu_bound, args=(N,))
    p2 = multiprocessing.Process(target=cpu_bound, args=(N,))
    p1.start(); p2.start()
    p1.join(); p2.join()
    multiprocess_time = time.time() - start

    print(f"sequential: {sequential_time:.2f}s, multiprocess: {multiprocess_time:.2f}s")
    print(multiprocess_time < sequential_time * 0.75)   # True -> a real speedup, on separate cores

sequential: 0.85s, multiprocess: 0.47s
True

def io_bound():
    time.sleep(0.3)   # stands in for waiting on a network/disk response

start = time.time()
io_bound()
io_bound()
sequential_io_time = time.time() - start

start = time.time()
t1 = threading.Thread(target=io_bound)
t2 = threading.Thread(target=io_bound)
t1.start(); t2.start()
t1.join(); t2.join()
threaded_io_time = time.time() - start

print(f"sequential: {sequential_io_time:.2f}s, threaded: {threaded_io_time:.2f}s")
print(threaded_io_time < sequential_io_time * 0.75)   # True -> a real speedup - the GIL was released during sleep

sequential: 0.60s, threaded: 0.30s
True

import asyncio

async def fetch(name, delay):
    print(f"{name}: starting")
    await asyncio.sleep(delay)   # yields control back to the event loop instead of blocking
    print(f"{name}: done")
    return name

async def main():
    start = time.time()
    results = await asyncio.gather(fetch("A", 0.3), fetch("B", 0.3), fetch("C", 0.3))
    print(results)                                              # ['A', 'B', 'C']
    print(time.time() - start < 0.6)   # True -> all three ran CONCURRENTLY, not one after another

asyncio.run(main())
# A: starting   -> all three reach their first "await" before any of them resumes
# B: starting
# C: starting
# A: done            -> then all three finish sleeping at roughly the same time
# B: done
# C: done

A: starting
B: starting
C: starting
A: done
B: done
C: done
['A', 'B', 'C']
True

async def greet():
    print("hello")

async def demo_missing_await():
    result = greet()   # forgot "await" - this does NOT run the coroutine body yet
    print(type(result))   # <class 'coroutine'> -> just an unstarted coroutine object
    await result                # NOW it actually runs

asyncio.run(demo_missing_await())
# <class 'coroutine'>
# hello

<class 'coroutine'>
hello

async def bad_task(name):
    print(f"{name}: starting (blocking sleep)")
    time.sleep(0.2)   # BLOCKS the entire event loop - no other task can run meanwhile
    print(f"{name}: done")

async def demo_blocking():
    start = time.time()
    await asyncio.gather(bad_task("A"), bad_task("B"))
    print(f"took {time.time() - start:.2f}s")   # ~0.4s -> ran SEQUENTIALLY, not concurrently

asyncio.run(demo_blocking())
# A: starting (blocking sleep)
# A: done          -> B never even gets to start until A's blocking call finally returns
# B: starting (blocking sleep)
# B: done
# took 0.40s -> compare to the ~0.3s asyncio.sleep() version above, which ran truly concurrently

A: starting (blocking sleep)
A: done
B: starting (blocking sleep)
B: done
took 0.40s

from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

async def run_mixed():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        # hands CPU-bound work off to a separate process, without blocking the event loop
        result = await loop.run_in_executor(pool, cpu_heavy, 5_000_000)
    return result

if __name__ == "__main__":
    print(asyncio.run(run_mixed()))   # 41666654166667500000 -> asyncio + multiprocessing, mixed

41666654166667500000

shared_counter = 0

def increment_unsafe():
    global shared_counter
    for _ in range(5):
        temp = shared_counter    # read
        time.sleep(0.001)             # a deliberate window another thread can interleave into
        shared_counter = temp + 1   # write - based on a possibly-now-stale "temp"

threads = [threading.Thread(target=increment_unsafe) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(shared_counter)                # expected 20 (4 threads x 5 increments); usually LESS
print(shared_counter < 20)          # True -> some increments were silently lost

5
True

safe_counter = 0
lock = threading.Lock()

def increment_safe():
    global safe_counter
    for _ in range(5):
        with lock:   # only one thread at a time can be inside this block
            temp = safe_counter
            time.sleep(0.001)
            safe_counter = temp + 1

threads = [threading.Thread(target=increment_safe) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(safe_counter)               # 20, every time
print(safe_counter == 20)   # True -> the Lock forces the increments to happen one at a time

20
True

class Counter:
    def __init__(self):
        self.lock = threading.RLock()
        self.value = 0

    def increment(self):
        with self.lock:              # first acquire
            self.value += 1
            self.log()                    # calls another method that ALSO acquires the same lock

    def log(self):
        with self.lock:              # re-acquires the SAME lock, on the SAME thread - fine with RLock
            print(f"value is now {self.value}")

c = Counter()
t = threading.Thread(target=c.increment)
t.start()
t.join()
# value is now 1
# a plain Lock would DEADLOCK here: increment() holds it, then log() tries to acquire
# the SAME Lock on the SAME thread and blocks forever waiting for itself to release it

value is now 1

sem = threading.Semaphore(2)   # only 2 threads allowed inside at once
running = []

def limited_worker(n):
    with sem:
        running.append(n)
        print(f"worker {n} running, concurrent: {running}")
        time.sleep(0.05)
        running.remove(n)

threads = [threading.Thread(target=limited_worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# 5 threads total, but the printed "concurrent" list never shows more than 2 names at once -
# Semaphore(2) only ever lets 2 of them past ".acquire()" (the "with sem:") at the same time

worker 0 running, concurrent: [0]
worker 1 running, concurrent: [0, 1]
worker 2 running, concurrent: [2]
worker 3 running, concurrent: [2, 3]
worker 4 running, concurrent: [4]

event = threading.Event()

def waiter():
    print("waiter: waiting for the signal")
    event.wait()   # blocks here until another thread calls event.set()
    print("waiter: got the signal, proceeding")

t = threading.Thread(target=waiter)
t.start()
time.sleep(0.1)   # give the waiter time to reach event.wait() first
print("main: sending the signal")
event.set()   # unblocks the waiting thread
t.join()
# waiter: waiting for the signal
# main: sending the signal
# waiter: got the signal, proceeding

waiter: waiting for the signal
main: sending the signal
waiter: got the signal, proceeding