Python asyncio 데드락 문제 해결

Introduction

Asyncio deadlocks are among the most difficult Python bugs to diagnose. Unlike threading deadlocks, which at least produce a visible hang or timeout, asyncio deadlocks cause your entire event loop to freeze silently. There is no exception, no traceback, no error message — your program simply stops responding.

I encountered this while building a real-time data pipeline for cryptocurrency trading in Lisbon. My application needed to fetch data from multiple exchanges simultaneously, but under heavy load it would occasionally freeze completely with no error output. After days of debugging, I discovered that a coroutine was calling await on a future that was waiting for itself to complete.

Environment

Python 3.12.3
asyncio 3.4.3 (built-in)
Linux 6.5.0 (Ubuntu 22.04)

Problem

Here is a minimal reproduction of an asyncio deadlock:

import asyncio

async def fetch_data(queue: asyncio.Queue):
    data = await queue.get()  # Waits for queue.put()
    print(f"Got data: {data}")

async def process_data(queue: asyncio.Queue):
    result = await some_coroutine(queue)  # Deadlock!
    queue.put_nowait(result)

async def some_coroutine(queue: asyncio.Queue):
    data = await queue.get()  # Waiting for something that will never come
    return data * 2

async def main():
    queue = asyncio.Queue()
    
    # Schedule both coroutines
    task1 = asyncio.create_task(fetch_data(queue))
    task2 = asyncio.create_task(process_data(queue))
    
    await asyncio.gather(task1, task2)

asyncio.run(main())  # Hangs forever

Running this produces nothing — the program just hangs:

$ python deadlock_example.py
# (no output, program hangs indefinitely)
# Ctrl+C to exit

A more subtle version:

import asyncio

async def producer(queue: asyncio.Queue):
    while True:
        await asyncio.sleep(1)
        queue.put_nowait("data")

async def consumer(queue: asyncio.Queue):
    while True:
        # If this runs before producer puts anything,
        # and producer is also waiting on queue, deadlock
        data = await queue.get()
        print(f"Processing: {data}")

async def main():
    queue = asyncio.Queue(maxsize=1)
    
    # If queue is full, put() blocks the event loop
    # But since we're in async context, we need to use await put()
    # This can cause deadlock if consumer is not running
    await queue.put("initial")
    
    tasks = [
        asyncio.create_task(producer(queue)),
        asyncio.create_task(consumer(queue)),
    ]
    
    await asyncio.gather(*tasks)

asyncio.run(main())

Analysis

Asyncio deadlocks occur when two or more coroutines are each waiting for the other to complete, creating a circular dependency that the event loop cannot resolve.

Root Cause 1: Using blocking calls in async code. The most common cause is calling a blocking function directly instead of using its async equivalent.

# DEADLOCK: This blocks the entire event loop
import requests
response = requests.get("https://api.example.com/data")  # Blocking!

# CORRECT: Use aiohttp or httpx with async
import aiohttp
async with aiohttp.ClientSession() as session:
    async with session.get("https://api.example.com/data") as response:
        data = await response.json()

Root Cause 2: Circular await dependencies. When coroutine A waits for coroutine B, which waits for coroutine A.

async def a():
    result = await b()  # Waits for b
    return result

async def b():
    result = await a()  # Waits for a — deadlock!
    return result

Root Cause 3: Queue operations without proper async handling. Using queue.put() instead of await queue.put() or vice versa.

# WRONG: This can block if queue is full
queue.put_nowait(data)  # Raises QueueFull if full, but doesn't wait

# CORRECT: This properly awaits when queue is full
await queue.put(data)

Root Cause 4: Lock contention. Acquiring the same lock from different coroutines without releasing it.

lock = asyncio.Lock()

async def task1():
    async with lock:
        # If task2 holds the lock and waits for task1
        # This is a deadlock
        await some_operation()

async def task2():
    async with lock:
        await task1()  # Waits for task1, which waits for this lock

Solution

Strategy 1: Use asyncio.wait_for() with timeouts to detect deadlocks.

import asyncio

async def safe_coroutine():
    try:
        result = await asyncio.wait_for(
            some_blocking_coroutine(),
            timeout=5.0  # 5 second timeout
        )
        return result
    except asyncio.TimeoutError:
        print("Coroutine timed out — possible deadlock detected")
        return None

Strategy 2: Use asyncio.shield() to prevent cancellation chains.

async def protected_operation():
    result = await asyncio.shield(some_coroutine())
    return result

Strategy 3: Always use non-blocking queue operations.

async def producer(queue: asyncio.Queue):
    while True:
        await asyncio.sleep(1)
        await queue.put("data")  # Use await, not put_nowait

async def consumer(queue: asyncio.Queue):
    while True:
        data = await queue.get()  # Properly awaits
        print(f"Processing: {data}")
        queue.task_done()  # Always call task_done after get

Strategy 4: Add deadlock detection with asyncio.current_task() debugging.

import asyncio
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

async def monitored_coroutine(name: str):
    logger.debug(f"Starting coroutine: {name}")
    try:
        result = await some_coroutine()
        logger.debug(f"Coroutine {name} completed")
        return result
    except Exception as e:
        logger.error(f"Coroutine {name} failed: {e}")
        raise

# Enable debug mode to see blocking calls
asyncio.run(main(), debug=True)

Strategy 5: Use structured concurrency with asyncio.TaskGroup() (Python 3.11+).

import asyncio

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(producer(queue))
        task2 = tg.create_task(consumer(queue))
    # All tasks are guaranteed to complete or cancel together

Lessons Learned

• Never call blocking functions directly in async code. Use asyncio.to_thread() or native async libraries.
• Always use timeouts with asyncio.wait_for() to catch potential deadlocks before they hang your application.
• Enable asyncio debug mode during development to detect blocking calls and slow coroutines.
• Use queue.task_done() and queue.join() to properly synchronize producer-consumer patterns.
• Prefer asyncio.TaskGroup() over manual task management in Python 3.11+ for safer concurrent operations.

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.