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

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.

There are comments.

Other articles

A Flask "g" object for FastAPI

Published: Dim 02 août 2020
By Gilles Lenfant

In Blog.

tags: Python Tip Flask FastAPI

The Flask "g" object is a cool feature, FastAPI has no such feature, but wait...
read more
There are comments.
A simple customizable configuration

Published: Ven 01 mars 2019
Updated: Ven 31 juillet 2020
By Gilles Lenfant

In Blog.

tags: Python Tip Configuration

A simple customizable configuration file
read more
There are comments.
Find the last iteration

Published: Sam 16 septembre 2017
By Gilles Lenfant

In Blog.

tags: Python Tip

Find the last iteration on any iterable object
read more
There are comments.
"Flatten" nested loops

Published: Lun 31 juillet 2017
By Gilles Lenfant

In Blog.

tags: Python Tip

One loop in place of x nested loops
read more
There are comments.
From dotted name to object

Published: Lun 31 juillet 2017
By Gilles Lenfant

In Blog.

tags: Python Tip

Make an object drom a simple dotted name
read more
There are comments.