Deadlocks and How to Avoid Them
What is a deadlock?
A deadlock happens when threads wait forever because each holds a resource the other needs.
Example deadlock
deadlock.py
import threading
import time
lock_a = threading.Lock()
lock_b = threading.Lock()
def t1():
with lock_a:
time.sleep(0.1)
with lock_b:
print("t1")
def t2():
with lock_b:
time.sleep(0.1)
with lock_a:
print("t2")
threading.Thread(target=t1).start()
threading.Thread(target=t2).start()deadlock.py
import threading
import time
lock_a = threading.Lock()
lock_b = threading.Lock()
def t1():
with lock_a:
time.sleep(0.1)
with lock_b:
print("t1")
def t2():
with lock_b:
time.sleep(0.1)
with lock_a:
print("t2")
threading.Thread(target=t1).start()
threading.Thread(target=t2).start()This can deadlock because each thread holds one lock and waits for the other.
How to prevent deadlocks
- Lock ordering: always acquire locks in the same order.
- Timeouts: use
lock.acquire(timeout=...)lock.acquire(timeout=...). - Keep critical sections short.
- Prefer higher-level concurrency tools (Queue, Executor) when possible.
Example: lock ordering fix
deadlock_fix.py
import threading
lock_a = threading.Lock()
lock_b = threading.Lock()
def safe():
# Always acquire A then B
with lock_a:
with lock_b:
print("safe")
threads = [threading.Thread(target=safe) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()deadlock_fix.py
import threading
lock_a = threading.Lock()
lock_b = threading.Lock()
def safe():
# Always acquire A then B
with lock_a:
with lock_b:
print("safe")
threads = [threading.Thread(target=safe) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()๐งช Try It Yourself
Exercise 1 โ Basic Lock Usage
Exercise 2 โ Lock as Context Manager
Exercise 3 โ Detect Locked State
If this helped you, consider buying me a coffee โ
Buy me a coffeeWas this page helpful?
Let us know how we did
