forking_deadlock.py

import threading
import time
from concurrent.futures import ProcessPoolExecutor

lock = threading.Lock()

def process_items(name):
    lock_id = id(lock)
    print(f"{name}: acquiring lock:{lock_id}")
    with lock:
        print(f"{name}: has lock:{lock_id}")
        time.sleep(10)
    print(f"{name}: released lock:{lock_id}")

if __name__ == "__main__":
    t = threading.Thread(target=process_items, args=("Thread",))
    t.start()

    time.sleep(2)

    with ProcessPoolExecutor() as e:
        e.submit(process_items, "Process")

only on linux .. python:3.12-alpine