Python asyncio deadlocks streaming llm applications typically appear as silent hangs: your event loop stays alive but no progress occurs, and logs show workers stuck mid-async for. These bugs rarely show in unit tests because they depend on cancellation, backpressure, and unread response bodies under concurrency.
1. Reproduce the hang with a minimal client
Start with a stripped-down streaming call to any OpenAI-compatible endpoint. The naive version spawns a task per prompt and cancels it after a short delay.
import asyncio, httpx
async def stream_llm(prompt: str):
async with httpx.AsyncClient() as client:
async with client.stream(
"POST", "https://api.example.com/v1/chat/completions",
json={"messages": [{"role": "user", "content": prompt}], "stream": True}
) as resp:
async for chunk in resp.aiter_text():
process(chunk)
async def main():
task = asyncio.create_task(stream_llm("hello"))
await asyncio.sleep(0.1)
task.cancel()
try:
await task
except asyncio.CancelledError:
passRun this and you will often see ResourceWarning: unclosed transport or the process hangs if process() blocks. Cancellation tears down the coroutine but leaves the HTTP connection half-open because the stream iterator was never exhausted. The same pattern with aiohttp produces a stuck ConnectionAcquire waiter.
2. Inspect live tasks instead of guessing
Before changing code, confirm where the stall is. In asyncio, a deadlock is just tasks waiting on each other or on I/O that never arrives. Dump the stacks of all tasks:
import asyncio
def dump_tasks():
for t in asyncio.all_tasks():
print(f"Task {t.get_name()}:")
t.print_stack()Call dump_tasks() from a signal handler or a periodic debug coroutine. You will typically see one task stuck in httpx._async.stream awaiting read(), while another waits on asyncio.Semaphore.acquire or Queue.get. That tells you the stream consumer is not draining.
Enable debug mode for extra signal:
loop = asyncio.new_event_loop()
loop.set_debug(True)
asyncio.set_event_loop(loop)With debug on, asyncio logs coroutines that take too long and emits ResourceWarning on leaked transports. For production, py-spy dump --pid <pid> shows native and Python stacks without code changes.
3. Understand why unread streams cause deadlocks
HTTP clients reuse connections from a pool. When you async for a response body and break early, the library expects you to either finish iterating or explicitly release the connection. If you cancel the task, the async generator’s aclose may not run, leaving the socket in a read await. The pool then blocks on acquire() for the next request because the slot is occupied by a dead connection.
With python asyncio deadlocks streaming llm workloads, this compounds: you fire many concurrent streams, each holds a connection, and soon the whole pool is exhausted. The event loop is idle but every new request waits on a semaphore that will never free. httpx defaults to 100 connections; aiohttp to 100 per host. Starve that and you get TimeoutError on acquire, not a clear deadlock message.
4. Always close the stream on cancellation
The fix is explicit resource management. Use try/finally and ensure response.aclose() is awaited even when the caller cancels.
async def stream_llm_safe(prompt: str):
async with httpx.AsyncClient() as client:
async with client.stream(
"POST", "https://api.example.com/v1/chat/completions",
json={"stream": True, "messages": [{"role": "user", "content": prompt}]}
) as resp:
try:
async for chunk in resp.aiter_text():
process(chunk)
except asyncio.CancelledError:
await resp.aclose()
raiseIf you use asyncio.wait_for or asyncio.timeout, the same rule applies:
async def stream_with_timeout(prompt: str, timeout: float):
async with httpx.AsyncClient() as client:
async with client.stream("POST", "...", json={"stream": True}) as resp:
try:
async with asyncio.timeout(timeout):
async for chunk in resp.aiter_text():
process(chunk)
except TimeoutError:
await resp.aclose()
raisePitfall: asyncio.wait_for cancels the inner coroutine but does not close the httpx response. Catch and close, or the connection leaks.
The asyncio.shield tradeoff
If you must drain the remainder of the stream before closing (e.g., to log a partial completion), asyncio.shield can help:
async def drain(resp):
async for chunk in resp.aiter_text():
log(chunk)
try:
await asyncio.shield(drain(resp))
except asyncio.CancelledError:
await resp.aclose()
raiseShielding delays cancellation and can violate your timeout budget. Use it only when clean shutdown matters more than immediate cancel.
5. Bound concurrency and apply backpressure
Deadlocks also arise when producers outpace consumers. An unbounded queue grows until memory pressure stalls the loop. Use a bounded queue and a semaphore for outbound streams.
sem = asyncio.Semaphore(10)
async def worker(queue: asyncio.Queue):
while True:
prompt = await queue.get()
async with sem:
try:
await stream_llm_safe(prompt)
finally:
queue.task_done()
async def producer(queue: asyncio.Queue, prompts):
for p in prompts:
await queue.put(p) # blocks when full, providing backpressureThe queue.put await forces the producer to slow down instead of spawning unlimited tasks. Tradeoff: end-to-end latency increases under load, but the system stays live. For Python 3.11+, prefer asyncio.TaskGroup for structured cancellation—but remember sibling tasks still need the aclose pattern inside their streams.
6. Test under induced latency
A unit test using a local server with asyncio.sleep before each chunk exposes cancellation bugs.
import pytest, httpx, asyncio
from httpx import ASGITransport, AsyncClient
@pytest.mark.asyncio
async def test_stream_cancel():
async def slow_stream(request):
await asyncio.sleep(0.05)
return httpx.Response(200, text="data: chunk\n\n")
transport = ASGITransport(app=slow_stream)
async with AsyncClient(transport=transport) as client:
task = asyncio.create_task(stream_llm_safe("x"))
await asyncio.sleep(0.01)
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
assert not asyncio.all_tasks() - {asyncio.current_task()}Run with -W error::ResourceWarning to catch leaked connections. Add a second test that opens 200 concurrent streams against a slow server to verify pool exhaustion no longer hangs.
7. Server-side fallback is not client-side immunity
If you route through a gateway such as n4n.ai, its OpenAI-compatible endpoint performs automatic fallback across 240+ models when a provider is rate-limited or degraded. That capability is server-side; your client code still owns the stream lifecycle. A fallback event may cause a brief stall or a new response object—if your coroutine is cancelled mid-stream, you must still aclose the original response, or the connection stays pinned in your event loop. The gateway will not reclaim local resources for you.
8. Actionable debugging checklist
- Reproduce with a minimal async client that cancels mid-stream.
- Dump task stacks with
task.print_stack()orpy-spyto locate the block. - Confirm whether the block is in
stream.reador in a lock/acquire. - Add
try/finallywithresponse.aclose()in every streaming path. - Replace bare
create_task+cancelwithasyncio.timeoutand explicit close. - Limit concurrency with
Semaphoreand use a boundedQueuefor backpressure. - Write a latency-injected test that asserts zero pending tasks after cancel.
Following this order turns a mysterious hang into a five-minute fix. Python asyncio deadlocks streaming llm services are almost always unclosed transports or unbounded queues—not the event loop itself.