Processes

A process is the smallest unit of resource allocation in an operating system, and also an instance of a running program. Each process has its own independent memory space; data between processes is isolated by default. Python provides multi-process support through the multiprocessing module.

Concurrency vs Parallelism

Concurrency: Appears to run simultaneously at a macro level, but the CPU is actually switching rapidly (can be achieved on a single core).
Parallelism: Truly simultaneous execution, requiring multiple CPU cores.

Why Use Multiple Processes

CPython’s GIL prevents threads from truly parallelizing CPU-bound tasks. Each subprocess in a multi-process setup has its own interpreter, bypassing the GIL and making full use of multiple CPU cores.

Scenario	Recommended Approach
CPU-bound (numerical computation, compression)	`multiprocessing` / `ProcessPoolExecutor`
I/O-bound (network requests, file I/O)	`threading` / `asyncio`
Mixed scenarios	`ProcessPoolExecutor` + `ThreadPoolExecutor`

Creating Processes

Function-Based Approach

from multiprocessing import Process
import time
import os

def worker(name: str) -> None:
    print(f"[Child process {os.getpid()}] {name} started")
    time.sleep(1)
    print(f"[Child process {os.getpid()}] {name} finished")

if __name__ == "__main__":
    p1 = Process(target=worker, args=("Task A",))
    p2 = Process(target=worker, args=("Task B",))

    p1.start()
    p2.start()

    p1.join()   # Wait for the child process to finish
    p2.join()

    print(f"[Main process {os.getpid()}] All child processes complete")

On Windows, Process() calls must be placed inside an if __name__ == "__main__": guard block; otherwise, processes will be created recursively without end.

Subclassing Process

from multiprocessing import Process
import time

class DownloadTask(Process):
    def __init__(self, url: str):
        super().__init__()
        self.url = url

    def run(self) -> None:
        print(f"Downloading: {self.url}")
        time.sleep(1)
        print(f"Completed: {self.url}")

if __name__ == "__main__":
    tasks = [
        DownloadTask("https://example.com/file1.zip"),
        DownloadTask("https://example.com/file2.zip"),
    ]
    for t in tasks:
        t.start()
    for t in tasks:
        t.join()

Process Data Isolation

A child process’s modifications to shared variables do not affect the parent process:

from multiprocessing import Process

money = 100

def task() -> None:
    global money
    money = 666
    print(f"Child process: money = {money}")   # 666

if __name__ == "__main__":
    p = Process(target=task)
    p.start()
    p.join()
    print(f"Main process: money = {money}")   # Still 100 (data isolation)

Daemon Processes

A daemon process terminates when the main process’s code finishes executing:

from multiprocessing import Process
import time

def heartbeat() -> None:
    while True:
        print("♥ Child process heartbeat")
        time.sleep(1)

if __name__ == "__main__":
    p = Process(target=heartbeat)
    p.daemon = True   # Must be set before start()
    p.start()

    time.sleep(3)
    print("Main process ending, daemon process will also exit")

Mutex Lock (Lock)

When multiple processes operate on shared data, a lock is needed to prevent race conditions:

from multiprocessing import Process, Lock
import json
import time

def buy_ticket(name: str, lock: Lock) -> None:
    with open("tickets.json", encoding="utf-8") as f:
        data = json.load(f)
    print(f"{name} sees remaining tickets: {data['count']}")

    with lock:
        with open("tickets.json", encoding="utf-8") as f:
            data = json.load(f)
        if data["count"] > 0:
            data["count"] -= 1
            with open("tickets.json", "w", encoding="utf-8") as f:
                json.dump(data, f)
            print(f"{name} successfully purchased a ticket!")
        else:
            print(f"{name}: tickets sold out")

if __name__ == "__main__":
    import json
    with open("tickets.json", "w") as f:
        json.dump({"count": 3}, f)

    lock = Lock()
    users = ["User A", "User B", "User C", "User D", "User E"]
    processes = [Process(target=buy_ticket, args=(u, lock)) for u in users]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

Inter-Process Communication (Queue)

multiprocessing.Queue is a process-safe message queue and the preferred method for inter-process communication (IPC):

from multiprocessing import Process, Queue
import time
import random

def producer(q: Queue, name: str) -> None:
    for i in range(3):
        item = f"{name}-{i}"
        q.put(item)
        print(f"Produced: {item}")
        time.sleep(random.uniform(0.1, 0.5))
    q.put(None)   # Send termination signal

def consumer(q: Queue) -> None:
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        time.sleep(0.2)

if __name__ == "__main__":
    q: Queue = Queue(maxsize=10)
    p = Process(target=producer, args=(q, "bun"))
    c = Process(target=consumer, args=(q,))
    p.start()
    c.start()
    p.join()
    c.join()

Producer-Consumer Model

For multi-producer multi-consumer scenarios, use JoinableQueue to handle termination signals gracefully:

from multiprocessing import Process, JoinableQueue
import time
import random
import os

def producer(name: str, q: JoinableQueue) -> None:
    for i in range(5):
        item = f"{name}-{i}"
        q.put(item)
        print(f"[{os.getpid()}] Produced: {item}")
        time.sleep(random.uniform(0.1, 0.3))
    q.join()   # Wait until all items in the queue have been processed

def consumer(q: JoinableQueue) -> None:
    while True:
        item = q.get()
        print(f"[{os.getpid()}] Consumed: {item}")
        time.sleep(0.2)
        q.task_done()   # Notify q.join() that this item has been processed

if __name__ == "__main__":
    q: JoinableQueue = JoinableQueue()

    producers = [
        Process(target=producer, args=("bun", q)),
        Process(target=producer, args=("bone", q)),
    ]
    consumers = [
        Process(target=consumer, args=(q,), daemon=True),
        Process(target=consumer, args=(q,), daemon=True),
    ]

    for p in producers + consumers:
        p.start()

    for p in producers:
        p.join()

    # After all producers finish, consumer daemon processes exit with the main process
    print("All producers done, program exiting")

Process Pool (Recommended: ProcessPoolExecutor)

concurrent.futures.ProcessPoolExecutor is the modern Python recommended interface for process pools:

from concurrent.futures import ProcessPoolExecutor, as_completed
import os

def cpu_task(n: int) -> int:
    """CPU-bound task: compute the sum of range(n)"""
    result = sum(range(n))
    print(f"[{os.getpid()}] n={n} → {result}")
    return result

if __name__ == "__main__":
    tasks = [10_000, 20_000, 30_000, 40_000, 50_000]

    # map: preserves order, concise
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_task, tasks))
    print("map results:", results)

    # submit + as_completed: process whichever finishes first
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(cpu_task, n): n for n in tasks}
        for future in as_completed(futures):
            n = futures[future]
            try:
                print(f"n={n} done: {future.result()}")
            except Exception as e:
                print(f"n={n} failed: {e}")

Legacy Process Pool (multiprocessing.Pool)

from multiprocessing import Pool
import os

def worker(x: int) -> int:
    return x * x

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        # map: blocks until all results are ready
        results = pool.map(worker, range(10))
        print(results)

        # apply_async: non-blocking, submit asynchronously
        async_results = [pool.apply_async(worker, (i,)) for i in range(10)]
        values = [r.get(timeout=5) for r in async_results]
        print(values)

Sharing Data Between Processes (Manager)

When data sharing is truly necessary, use Manager (higher overhead than Queue; prefer Queue):

from multiprocessing import Process, Manager, Lock

def worker(d: dict, lock: Lock) -> None:
    with lock:
        d["count"] -= 1

if __name__ == "__main__":
    lock = Lock()
    with Manager() as m:
        shared_dict = m.dict({"count": 100})
        processes = [Process(target=worker, args=(shared_dict, lock)) for _ in range(100)]
        for p in processes:
            p.start()
        for p in processes:
            p.join()
        print(shared_dict)   # {'count': 0}

Zombie Processes and Orphan Processes

Zombie process: A child process has ended, but the parent process has not yet called wait()/join() to reclaim its resources. The process ID remains occupied.
- Solution: Call p.join() promptly in the parent process, or run the child as a daemon.
Orphan process: The parent process terminates unexpectedly, leaving the child without a parent. The operating system re-parents it to the init process (PID=1), which reclaims it automatically.

from multiprocessing import Process
import time

def child() -> None:
    time.sleep(2)
    print("Child process finished")

if __name__ == "__main__":
    p = Process(target=child)
    p.start()
    # Without p.join(), the child briefly becomes a zombie after it ends
    # The OS reclaims resources when the main process exits
    p.join()   # Correct approach: wait for the child process to finish

Last updated on 2026-06-23

Encryption Modules Threads