Awaiting a blocking callable

tl; dr

Working with frameworks that leverage the asyncio paradigm is now familiar to lots of Python programmers with popular asynchronous frameworks like FastAPI, Faust, Trio (name yours...).

We need sometimes to use good libs that do not (yet ?) leverage the - not so - new asyncio way, and provide only blocking APIs. Think about SQLAlchemy, or others.

Using such services from a coroutine under event loop control will lock the event loop while executing thus yield a performance issue.

Hopefully the stdlib provides the concurrent.futures package that let us work around this annoyance and make a coroutine from a blocking function or method.

The code

# Running a blocking callable inside a coroutine # ============================================== import asyncio import concurrent.futures import functools from typing import Callable, Tuple, Dict, Any # Use the better suited pool (see doc of ``concurrent.future```) #pool = concurrent.futures.ThreadPoolExecutor() #pool = concurrent.futures.ProcessPoolExecutor() pool = None # Default asyncio pool async def call_blocking(func: Callable, *args: Tuple[Any], **kwargs: Dict[str, Any]) -> Any: """ Enable to "await" a blocking I/O callable from an asyncio coroutine Args: func: The regular blocking callable (function, method) args: Positional arguments transmitted to ``func`` kwargs: Keyword arguments transmitted to ``func`` Returns: Anything that's returned by ``func`` """ loop = asyncio.get_running_loop() new_func = functools.partial(func, *args, **kwargs) result = await loop.run_in_executor(pool, new_func) return result

The demo

Just paste this code under the above one...

if __name__ == "__main__": import time from call_blocking import call_blocking def multiply(value, multiplier=2): # A blocking function time.sleep(0.1) result = value * multiplier time.sleep(0.1) return result class Operations: def instmultiply(self, value, multiplier=2): return self.clsmultiply(value, multiplier) @classmethod def clsmultiply(cls, value, multiplier=2): return cls.statmultiply(value, multiplier) @staticmethod def statmultiply(value, multiplier=2): return multiply(value, multiplier) async def test(): print("Blocking func") result = await call_blocking(multiply, 2, multiplier=3) print(result) ops = Operations() print("Blocking static method") result = await call_blocking(ops.statmultiply, 3) print(result) print("Blocking class method") result = await call_blocking(ops.clsmultiply, 3) print(result) print("Blocking instance method") result = await call_blocking(ops.instmultiply, 3) print(result) asyncio.run(test())

About the pool executor

You may choose the _executor being either None, a ThreadPoolexecutor or a ProcessPoolExecutor instance. Just read the doc of concurrent.futures to choose the best suited. And read the Drawbacks below.

Pool executor constructor take an optional max_workers argument you should perhaps tune to obtain the best possible performances.

My hints: choose a ThreadPoolExecutor for I/O expensive functions, and a ProcessPoolExecutor for heavy computation callables.

Drawbacks

Using threads

You must be careful when playing with shared objects that could be modified by your blocking code and coroutines under event loop control. Avoid this when possible, or use locks.

In addition lots of trird party packages expose resources that are not "thread safe", thus cannot be invoked with this recipe.

Using processes

Processes do not share any global. Blocking code and coroutines must communicate via a multiprocessing.Queue. That's not very comfortable.

Digital Snake and Family Gossips about (mostly) Python based technologies

Awaiting a blocking callable

tl; dr

The code

The demo

About the pool executor

Drawbacks

Using threads

Using processes

Comments !

tl; dr

The code

The demo

About the pool executor

Drawbacks

Using threads

Using processes

Comments !

links

social