Timing Out Blocking Calls in Threads¶
await asyncio.wait_for(asyncio.to_thread(slow_call), 0.2) looks like it bounds a blocking call at 200 ms. It bounds your waiting at 200 ms. The thread keeps running the call to completion, holding a worker in the executor for as long as the call takes — measured below, the coroutine raised TimeoutError after 0.20 s and the thread still printed its result a full second later. Under load this is how a slow dependency stalls a service that appears to have timeouts everywhere: each request gives up on its call, the workers stay busy, and the next request's trivial thread call waits behind them. Python has no way to interrupt a thread from outside, so the fix is always to make the call itself stoppable, or to run it somewhere killable.
Prerequisites¶
- Python 3.11+, standard library only. Measurements are from a 24-core Linux machine.
- Offloading basics from CPU-Bound Task Offloading and Hybrid Concurrency Models.
- Cancellation semantics from Cancellation Patterns — cancellation in asyncio is cooperative, and threads do not cooperate.
1. Confirm what the timeout does not do¶
The demonstration takes six lines and is worth running once in your own codebase:
def blocking(seconds, flag):
time.sleep(seconds)
flag.append("finished anyway")
flag = []
try:
await asyncio.wait_for(asyncio.to_thread(blocking, 1.0, flag), 0.2)
except TimeoutError:
print("timed out") # after 0.20 s
await asyncio.sleep(1.1)
print(flag) # ['finished anyway']
Nothing here is a bug. wait_for cancels the coroutine awaiting the thread's future; concurrent.futures.Future.cancel() returns False for work that has already started, and Python provides no primitive to interrupt a running thread. The TimeoutError is truthful about your wait and silent about the work.
Two consequences follow. The worker is occupied until the call finishes, and any side effect of that call still happens — the row is still written, the file is still uploaded, well after you reported a timeout to the caller.
Verify: add a print at the end of the blocking function and watch it fire after your timeout handler.
2. Measure the cost to the pool¶
to_thread uses the loop's default executor, which holds min(32, cpu_count + 4) threads — 28 on this machine. That sounds like plenty until abandoned calls accumulate. On a two-worker pool with both workers running one-second calls, a trivial 10 ms call queued behind them took 0.81 s. With all four workers of a four-thread pool occupied by hung calls, a trivial call was still queued after 0.5 s and only ran once the hung calls were released.
ex = concurrent.futures.ThreadPoolExecutor(max_workers=1)
running = ex.submit(blocking, 0.5)
queued = ex.submit(blocking, 0.5)
queued.cancel() # True: it had not started
running.cancel() # False: too late
That asymmetry is the whole story in one snippet. Cancellation works only before the work begins.
The practical mitigation while you fix the real problem: give blocking calls their own bounded executor rather than the shared default, so a stuck dependency cannot exhaust the pool that everything else uses. That is the bulkhead pattern applied to threads.
Verify: count busy workers during an incident; if the number equals max_workers, thread starvation is the symptom you are seeing.
3. Push the timeout into the call¶
The only approach that genuinely stops the work at the operating-system level is a timeout the library itself enforces, because it becomes a bounded syscall rather than an unbounded one:
socket.settimeout(5) # socket operations
requests.get(url, timeout=(3, 10)) # connect, read
psycopg.connect(..., connect_timeout=5) # driver-level
subprocess.run(cmd, timeout=30) # process-level
cursor.execute("SET statement_timeout = '5s'") # server-side
Every mature blocking library has one, and it is the first thing to look for. subprocess.run(timeout=) is a special case worth knowing: it kills the child, so the timeout is real, and it is often the simplest way to bound an otherwise unbounded external tool.
Verify: with a library timeout set, the thread returns at the deadline and the worker is released.
4. Pass a stop flag when the code is yours¶
If the blocking function is yours — or can be wrapped — make it check a flag between units of work. threading.Event is the right type, and Event.wait(interval) doubles as both the sleep and the check:
def work(stop: threading.Event) -> int:
ticks = 0
while not stop.wait(0.05): # sleeps 50 ms, or returns early
ticks += 1
return ticks
async def call_with_timeout(fn, timeout: float):
stop = threading.Event()
future = asyncio.get_running_loop().run_in_executor(None, fn, stop)
try:
return await asyncio.wait_for(asyncio.shield(future), timeout)
except TimeoutError:
stop.set() # ask it to stop
ticks = await future # and wait until it has
raise TimeoutError(f"worker stopped after {ticks} ticks") from None
That returned worker stopped after 5 ticks in 0.30 s, with the thread count back to its baseline — the worker was genuinely released. Two details make it work. asyncio.shield keeps the future alive after wait_for gives up, so the cleanup branch can still await it. And awaiting the future after setting the flag is what turns "I stopped waiting" into "the work has stopped": without it you are back to an occupied worker, just with a flag set.
The granularity of the check is the worst-case overshoot. Checking between rows of a result set, chunks of a file or iterations of a loop is usually easy; checking inside a single 30-second C call is impossible, which is what the next section is for.
Verify: after the timeout, the executor's active thread count returns to baseline within one check interval.
5. Use a process when the call cannot cooperate¶
Opaque native code — an image codec, a scientific library, a vendor SDK — cannot be asked to stop. A process can be killed:
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as pool:
future = loop.run_in_executor(pool, opaque_call, arg)
try:
return await asyncio.wait_for(asyncio.shield(asyncio.wrap_future(future)), 0.3)
except TimeoutError:
for process in pool._processes.values():
os.kill(process.pid, signal.SIGKILL) # the work really stops here
raise
The killed worker's future then raises BrokenProcessPool, and the pool must be replaced — recovering from BrokenProcessPool covers doing that cleanly, and is required reading before shipping this. The costs are real: arguments and results are pickled, startup is tens of milliseconds, and killing the process discards any other work it was doing, which is why max_workers=1 pools per killable job are a common shape.
For CPU-bound Python, the free-threaded builds and InterpreterPoolExecutor change the arithmetic on offloading, but not this: only a separate process can be terminated from outside.
Verify: the worker process disappears from ps at the timeout, and the pool is rebuilt before the next call.
Verification¶
Blocking calls are properly bounded when:
- Every blocking library call passes its own timeout, and that timeout is shorter than the caller's budget.
- Worker occupancy returns to baseline after a timeout, not after the abandoned call finishes.
- Blocking work has a dedicated executor whose size you chose, not the shared default.
- Stop flags are awaited: the code waits for the worker to acknowledge before reporting the timeout.
- Uncooperative work runs in a process that is killed, with the pool rebuilt afterwards.
Pitfalls & edge cases¶
wait_forwithoutshieldin the cleanup path. Oncewait_forcancels the wrapper, the future is still running; withoutshieldyou lose the handle needed to await the worker's acknowledgement.ThreadPoolExecutor.shutdown(wait=False). It does not stop running threads, and the interpreter still joins them at exit —cancel_futures=Trueonly drops queued work.signal.alarmfor thread timeouts. Signals are delivered to the main thread only, so this cannot interrupt a worker; it is also incompatible with a running event loop's own signal handling.- Raising asynchronously into a thread.
ctypes.pythonapi.PyThreadState_SetAsyncExcexists, is not supported, and does not interrupt a blocking syscall anyway. - Timeouts shorter than the work's natural time. If every call times out, you are burning workers for nothing; fix the dependency or the budget.
- Counting on
to_threadfor many concurrent calls. The default executor caps at 32 threads; a thousand concurrent blocking calls queue, they do not parallelise.
Frequently Asked Questions¶
Does asyncio.wait_for cancel a thread started with to_thread?
No. It cancels the coroutine that is waiting, and raises TimeoutError to you. The operating-system thread runs the blocking call to completion and keeps its executor worker the whole time — measured, the call returned after 0.20 s while the thread finished its full 1.0 s of work.
How do I actually stop a blocking call in Python?
Three ways, in order of preference: pass a timeout to the library so the call itself returns at the deadline; pass a threading.Event the function checks between chunks; or run the call in a process pool and kill the worker. Nothing can interrupt an arbitrary running thread from outside.
Why did my service slow down when a dependency got slow, even though I have timeouts?
Because the timeouts freed your coroutines but not the executor threads. The abandoned calls keep their workers, and new thread calls queue behind them. Give blocking work a dedicated bounded executor and push a real timeout into the library call.
How many threads does asyncio.to_thread use?
It uses the loop's default ThreadPoolExecutor, which defaults to min(32, os.cpu_count() + 4) workers — 28 on a 24-core machine. Beyond that, calls queue. Use loop.set_default_executor() or an explicit executor when you need a different size or isolation.
Is it safe to kill a process pool worker on timeout?
It works, and the pending future then raises BrokenProcessPool. You must replace the executor afterwards, and any other work that worker was running is lost. Use a small dedicated pool for killable jobs so a kill does not take unrelated work down with it.
Related¶
- Timeouts & Deadlines — up to the topic overview.
- CPU-Bound Task Offloading — choosing between threads, processes and interpreters.
- Recovering from BrokenProcessPool — what to do after killing a worker.
- Resilience, Cancellation & Error Handling — the section overview.