Your async service needs a library that has no async version, such as a
database driver or an image resizer. Calling it inside async def blocks the
event loop, as time.sleep does. Three one-second calls take three
seconds, and every other task waits in the meantime.
asyncio.to_thread runs the function in a worker thread instead:
import asyncio
import time
def lookup(user_id: int) -> str:
time.sleep(1) # stands in for a blocking library call
return f"user-{user_id}"
async def main() -> None:
start = time.perf_counter()
names = await asyncio.gather(
asyncio.to_thread(lookup, 1),
asyncio.to_thread(lookup, 2),
asyncio.to_thread(lookup, 3),
)
print(names, f"{time.perf_counter() - start:.1f}s")
asyncio.run(main()) # ['user-1', 'user-2', 'user-3'] 1.0s
Move the blocking call to a thread
await asyncio.to_thread(fn, *args, **kwargs) hands fn to a worker thread
and suspends your task until the thread returns. While the thread waits, the
event loop runs other tasks. If fn raises, the same exception comes out of
the await.
The threads come from the loop's default thread pool, which every to_thread
call shares. To give one kind of work its own pool, create a
ThreadPoolExecutor and pass it to loop.run_in_executor:
pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="resize")
loop = asyncio.get_running_loop()
thumb = await loop.run_in_executor(pool, functools.partial(resize, image, width=200))
run_in_executor passes on positional arguments only, so bind keyword
arguments with functools.partial. Threads help when the call waits for I/O.
For CPU-heavy pure-Python work, use a ProcessPoolExecutor, as the GIL
article showed.
Call into the loop from another thread
Most asyncio objects, such as tasks, events and queues, are not thread-safe. A worker thread that needs to reach the loop uses one of two methods built for it:
loop.call_soon_threadsafe(callback, *args)schedules a plain function to run on the loop.asyncio.run_coroutine_threadsafe(coro, loop)schedules a coroutine and returns aconcurrent.futures.Future. The thread calls.result()on it to wait for the answer.
Call .result() only from a different thread. On the loop's own thread,
.result() blocks the loop, so the coroutine it waits for never runs.
Keep the function's type
You can wrap to_thread in a decorator that turns any blocking function into
an async one. The wrapper should accept exactly the arguments the original
takes, and the type checker should know that.
Python 3.12 lets you declare type parameters in square brackets after the
function name. T stands for one type, which the type checker works out at
each call. **P is a ParamSpec, short for parameter specification. It
stands for a whole parameter list. For join below, P is
(n: int, *, sep: str = ":") and T is str.
import asyncio
import functools
from collections.abc import Callable, Coroutine
from typing import Any
def asyncify[**P, T](fn: Callable[P, T]) -> Callable[P, Coroutine[Any, Any, T]]:
@functools.wraps(fn)
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
return await asyncio.to_thread(fn, *args, **kwargs)
return wrapper
def join(n: int, *, sep: str = ":") -> str:
'''Join a number with itself.'''
return sep.join([str(n), str(n)])
async def main() -> None:
joined = asyncify(join)
print(await joined(7, sep="-")) # 7-7
print(joined.__name__) # join
asyncio.run(main())
Callable[P, T] means a function with parameters P that returns T. The
annotations *args: P.args, **kwargs: P.kwargs make the wrapper accept the
same arguments. Now mypy rejects await joined("seven"), because n must be
an int. With Callable[..., Any] it would accept anything. functools.wraps
copies __name__ and __doc__ from fn, so logs and tracebacks still say
join.
The return type is Coroutine[Any, Any, T] so that callers can pass the
result to TaskGroup.create_task, which accepts only coroutines.
Awaitable[T] would be too loose for that. The first two arguments describe
values a coroutine yields and receives, which async code rarely touches.
Cancelling the task leaves the thread running
If you cancel a task that is awaiting to_thread, the task stops waiting.
The thread keeps running until fn returns, because Python cannot stop a
thread from outside. Give blocking calls their own timeouts, such as
requests.get(url, timeout=5).
A pool you create needs shutting down. pool.shutdown(wait=True) waits for
the running jobs, so calling it on the loop's thread blocks the loop until
they finish. Run it in a thread too:
await asyncio.to_thread(pool.shutdown, wait=True)
For the same reason, avoid with ThreadPoolExecutor() as pool: inside
async def. Leaving the block calls pool.shutdown(wait=True) on the loop's
thread.
Check your understanding
You cancel a task that is awaiting asyncio.to_thread(download, url). What happens to the download?
Show answer
Hide answer
It keeps going. Cancelling ends the task's wait, but the worker thread runs
download until it returns, and nothing reads its result. To limit how long
the download can run, pass a timeout to the download call itself.
Going further
to_thread also copies context variables, such as a request ID, into the
worker thread. run_in_executor does not. A later article in this track
covers context variables.
Read more in the asyncio documentation on
tasks and threads and
the event loop,
and in the concurrent.futures documentation.