#!/usr/bin/env python3 import os import threading def a(): def printer_thread(): n = 0 while n < NUM_PRINTS: print(f"Hello from thread (loop {n})") n += 1 print(f"Thread dying happy") t = threading.Thread(target=printer_thread) t.start() # same effect happens if we start the thread using # _thread.start_new_thread, or threading.Thread, # or a ThreadPoolExecutor. def b(): pids = [] for n in range(NUM_FORKS): if pid := os.fork(): pids.append(pid) else: if LETS_CAUSE_A_DEADLOCK: print(f"Hello from process {n}") os._exit(0) for pid in pids: wpid, wsts = os.waitpid(pid, 0) # same effect happens if we start the process using # os.fork, or mp.Process, or a ProcessPoolExecutor. if __name__ == "__main__": NUM_FORKS = NUM_PRINTS = 500 # The deadlock occurs if and only if the child prints LETS_CAUSE_A_DEADLOCK = True # The user calls some function, unaware that it uses threads. a() # The user calls another function, unaware that it forks. b() print(f"Main dying happy")