☠️ Pattern 6: Deadlocks

The Classic Two-Lock Deadlock

import threading, time

L1 = threading.Lock()
L2 = threading.Lock()

def thread_a():
    with L1:
        print("A acquired L1")
        time.sleep(0.1)          # give B time to grab L2 — forces the race
        with L2:                 # A now waits for L2 (held by B) forever
            print("A acquired L2")

def thread_b():
    with L2:
        print("B acquired L2")
        time.sleep(0.1)
        with L1:                 # B now waits for L1 (held by A) forever
            print("B acquired L1")

a = threading.Thread(target=thread_a)
b = threading.Thread(target=thread_b)
a.start(); b.start()
a.join(); b.join()               # never returns — the program hangs here

The Four Coffman Conditions

A deadlock can only happen when all four of these hold at once. Break any one and deadlock becomes impossible — that's the whole toolbox for prevention.

• Mutual exclusion: a lock is held by at most one thread. (Inherent to locks.)
• Hold and wait: a thread holds one lock while requesting another.
• No preemption: a lock can't be forcibly taken away; only its owner releases it.
• Circular wait: a cycle of threads, each waiting on a lock held by the next.

Fix #1 (Primary): Consistent Lock Ordering

Break the circular wait: if every thread always acquires locks in the same global order, no cycle can form. Give each lock a stable rank (its id(), a name, an index) and always take the lower-ranked one first.

import threading

L1 = threading.Lock()
L2 = threading.Lock()

def acquire_ordered(first, second):
    # sort the two locks by a stable key so EVERY thread agrees on the order
    lo, hi = sorted((first, second), key=id)
    return lo, hi

def transfer(from_lock, to_lock):
    lo, hi = acquire_ordered(from_lock, to_lock)
    with lo:
        with hi:                 # both threads take id(lo) first → no cycle
            ...                  # move the money

# both directions now acquire in the same physical order
threading.Thread(target=transfer, args=(L1, L2)).start()
threading.Thread(target=transfer, args=(L2, L1)).start()

Fix #2: acquire(timeout=...) to Fail Fast

When a global ordering is impractical, break hold-and-wait: don't wait forever for the second lock. Give up after a timeout, release what you hold, back off, and retry. This trades a guaranteed hang for occasional wasted work.

import threading, time, random

L1 = threading.Lock()
L2 = threading.Lock()

def transfer():
    while True:
        with L1:                          # hold the first lock
            got = L2.acquire(timeout=0.05)   # but DON'T block forever on the second
            if got:
                try:
                    ...                   # do the work with both locks
                    return
                finally:
                    L2.release()
        # failed to get L2: L1 is now released (with block exited).
        # back off a random amount to avoid re-colliding, then retry.
        time.sleep(random.uniform(0, 0.05))

Fix #3: Avoid Nested Locks Entirely

The safest deadlock fix is to not create the condition in the first place. If a thread never holds two locks at once, hold-and-wait can't happen.

• Coarsen: replace two fine-grained locks with a single lock guarding both resources — simpler, and often fast enough.
• Shrink the critical section: release lock A, compute, then acquire lock B, instead of nesting.
• Never call out under a lock: don't invoke unknown callbacks, or code that itself locks, while holding a lock — you can't see what it acquires.

RLock Cures Self-Deadlock, Not Cross-Thread Deadlock

A common confusion: a thread deadlocking on itself (re-acquiring a plain Lock it already holds — see Pattern 2) is fixed by an RLock. But that is a different problem. An RLock does nothing for the two-thread circular wait above.

import threading

rlock = threading.RLock()
with rlock:
    with rlock:      # ✅ fine: same thread re-enters its OWN lock
        ...

# But between TWO threads holding TWO RLocks in opposite order,
# you still deadlock exactly as with plain Locks. RLock only relaxes
# re-entrancy for the owning thread — not the circular-wait relationship.

The Best Fix: Don't Share Locks — Share a Queue

Most deadlocks come from threads reaching into shared mutable state guarded by multiple locks. A lock-free design sidesteps the whole category: hand data between threads through a queue.Queue instead of sharing memory behind locks. The queue owns its single internal lock, and no application code holds two locks at once.

import queue, threading

tasks = queue.Queue()

def worker():
    while True:
        item = tasks.get()      # thread-safe; no application-level locks at all
        if item is None:
            break
        ...                     # process item, no shared mutable state
        tasks.task_done()

threading.Thread(target=worker, daemon=True).start()
for i in range(100):
    tasks.put(i)
tasks.join()                    # wait for all items to be processed

Key Takeaways