228. Handling Deadlocks in Python

🔹 1. Basic Deadlock Example

This snippet demonstrates a classic deadlock scenario where two threads wait on each other forever.

import threading
import time

# Shared resources
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    lock1.acquire()
    print("Thread 1 acquired lock1")
    time.sleep(1)
    lock2.acquire()
    print("Thread 1 acquired lock2")
    lock2.release()
    lock1.release()

def thread2():
    lock2.acquire()
    print("Thread 2 acquired lock2")
    time.sleep(1)
    lock1.acquire()
    print("Thread 2 acquired lock1")
    lock1.release()
    lock2.release()

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# Start threads
t1.start()
t2.start()

t1.join()
t2.join()

💡 Problem: This leads to a deadlock because thread1 holds lock1 and waits for lock2, while thread2 holds lock2 and waits for lock1.

---

🔹 2. Avoiding Deadlock with Lock Ordering

To avoid deadlocks, always acquire locks in the same order.

import threading
import time

# Shared resources
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    with lock1:
        print("Thread 1 acquired lock1")
        time.sleep(1)
        with lock2:
            print("Thread 1 acquired lock2")

def thread2():
    with lock1:
        print("Thread 2 acquired lock1")
        time.sleep(1)
        with lock2:
            print("Thread 2 acquired lock2")

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# Start threads
t1.start()
t2.start()

t1.join()
t2.join()

✅ Solution: By acquiring the locks in a fixed order, we avoid the circular dependency and prevent deadlock.

---

🔹 3. Using timeout to Avoid Deadlocks

You can specify a timeout for acquiring a lock to avoid getting stuck forever.

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    if lock1.acquire(timeout=2):
        print("Thread 1 acquired lock1")
        time.sleep(1)
        if lock2.acquire(timeout=2):
            print("Thread 1 acquired lock2")
            lock2.release()
        else:
            print("Thread 1 failed to acquire lock2")
        lock1.release()

def thread2():
    if lock2.acquire(timeout=2):
        print("Thread 2 acquired lock2")
        time.sleep(1)
        if lock1.acquire(timeout=2):
            print("Thread 2 acquired lock1")
            lock1.release()
        else:
            print("Thread 2 failed to acquire lock1")
        lock2.release()

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# Start threads
t1.start()
t2.start()

t1.join()
t2.join()

✅ Solution: Using timeout prevents the threads from blocking forever and allows them to handle failure scenarios gracefully.

---

🔹 4. Using a Deadlock Detection Mechanism

Deadlock detection involves periodically checking for deadlocks and recovering from them.

import threading
import time

# Shared resources
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    lock1.acquire()
    print("Thread 1 acquired lock1")
    time.sleep(1)
    lock2.acquire()
    print("Thread 1 acquired lock2")
    lock2.release()
    lock1.release()

def thread2():
    lock2.acquire()
    print("Thread 2 acquired lock2")
    time.sleep(1)
    lock1.acquire()
    print("Thread 2 acquired lock1")
    lock1.release()
    lock2.release()

# Function to detect deadlocks
def detect_deadlock():
    while True:
        # Simple check for deadlock (not comprehensive)
        if threading.active_count() == 1:
            print("Possible deadlock detected!")
        time.sleep(2)

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)
detection_thread = threading.Thread(target=detect_deadlock, daemon=True)

# Start threads
t1.start()
t2.start()
detection_thread.start()

t1.join()
t2.join()

💡 Note: This approach is basic and may not be suitable for all scenarios. In practice, using a more robust approach such as analyzing the state of locks and thread dependencies is recommended.

---

🔹 5. Using threading.Condition to Avoid Deadlocks

threading.Condition can be used to synchronize threads more effectively, reducing the likelihood of deadlocks.

import threading

condition = threading.Condition()

def thread1():
    with condition:
        print("Thread 1 waiting for condition")
        condition.wait()  # Wait until notified
        print("Thread 1 proceeding after notification")

def thread2():
    with condition:
        print("Thread 2 notifying others")
        condition.notify()  # Notify all waiting threads

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# Start threads
t1.start()
t2.start()

t1.join()
t2.join()

✅ Solution: The Condition object helps synchronize the execution of threads without the risk of deadlock by providing a more structured signaling mechanism.

---

🔹 6. Using ThreadPoolExecutor for Simpler Thread Management

Using ThreadPoolExecutor reduces the complexity of thread management and helps in avoiding deadlocks.

import concurrent.futures
import threading
import time

def task1():
    print("Task 1 started")
    time.sleep(1)
    print("Task 1 completed")

def task2():
    print("Task 2 started")
    time.sleep(2)
    print("Task 2 completed")

# Using ThreadPoolExecutor for thread management
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    future1 = executor.submit(task1)
    future2 = executor.submit(task2)
    concurrent.futures.wait([future1, future2])

print("All tasks completed")

✅ Solution: The ThreadPoolExecutor manages threads efficiently, reducing the chances of encountering deadlocks by handling thread life cycles automatically.

---

🔹 7. Handling Lock Acquisition in Sequence

To avoid deadlocks, make sure that locks are always acquired in the same order across the application.

import threading
import time

# Shared resources
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1():
    with lock1:
        print("Thread 1 acquired lock1")
        time.sleep(1)
        with lock2:
            print("Thread 1 acquired lock2")

def thread2():
    with lock1:
        print("Thread 2 acquired lock1")
        time.sleep(1)
        with lock2:
            print("Thread 2 acquired lock2")

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# Start threads
t1.start()
t2.start()

t1.join()
t2.join()

✅ Solution: This ensures that both threads acquire lock1 and lock2 in the same sequence, which prevents circular waiting.

---

🔹 8. Using RLock to Avoid Deadlocks in Recursive Locks

RLock (reentrant lock) allows the same thread to acquire the same lock multiple times, preventing deadlocks.

import threading
import time

# Shared resource
rlock = threading.RLock()

def thread1():
    with rlock:
        print("Thread 1 acquired lock")
        time.sleep(1)
        with rlock:  # Reentrant, same lock
            print("Thread 1 acquired lock again")

def thread2():
    with rlock:
        print("Thread 2 acquired lock")
        time.sleep(1)

# Create threads
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# Start threads
t1.start()
t2.start()

t1.join()
t2.join()

✅ Solution: RLock allows threads to acquire the same lock multiple times, reducing the risk of deadlocks in scenarios where a thread needs to acquire the lock multiple times.

---

🔹 9. Using multiprocessing for Better Concurrency

Sometimes, using multiprocessing (which avoids Python's Global Interpreter Lock) can help avoid issues with threading deadlocks.

from multiprocessing import Process
import time

def task1():
    print("Task 1 started")
    time.sleep(1)
    print("Task 1 completed")

def task2():
    print("Task 2 started")
    time.sleep(2)
    print("Task 2 completed")

# Create processes
p1 = Process(target=task1)
p2 = Process(target=task2)

# Start processes
p1.start()
p2.start()

p1.join()
p2.join()

✅ Solution: Using multiprocessing eliminates GIL limitations and can avoid threading-related deadlocks.

---

🔹 10. Deadlock Avoidance Strategy: Timeout and Retry

If deadlock is suspected, you can implement a retry mechanism with a timeout.

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def try_acquire_locks():
    for _ in range(3):  # Retry mechanism
        if lock1.acquire(timeout=1):
            print("Acquired lock1")
            if lock2.acquire(timeout=1):
                print("Acquired lock2")
                lock2.release()
                lock1.release()
                return
            else:
                lock1.release()
                time.sleep(1)  # Wait before retrying
    print("Failed to acquire locks")

# Create thread
t = threading.Thread(target=try_acquire_locks)
t.start()
t.join()

✅ Solution: By retrying lock acquisition, we reduce the chances of deadlocks, giving tasks a chance to recover.

---