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")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.
- Solution: Call
- Orphan process: The parent process terminates unexpectedly, leaving the child without a parent. The operating system re-parents it to the
initprocess (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