Python added formal asynchronicity in the base language a while ago. It’s fun to play with asyncio
coroutines, the basic constructs that execute almost in parallel. But as you start to integrate more with a regular codebase, you may find that things can get tricky. Especially if you’re forced to interact with synchronous code.
The complication arises when invoking awaitable functions. Doing so requires an
async defined code block or coroutine. A non-issue except that if your caller has to be async, then you can’t call it either unless its caller is async. Which then forces its caller into an async block as well, and so on. This is “async creep”.
It can escalate fast, finding its way into all corners of your code. Not a big deal if the codebase is asynchronous, but can impede development when mixing it up.
Following are the two main mechanisms I use to work around this. Both assume we’re building an application that benefits from asynchronous execution.
Waiting for blocks of async code
Whether building an asynchronous application or enhancing a linear one, it’s important to determine the sections that will gain the most from async execution. This is usually not hard to answer, but no one else can do it for you. The general guideline is to start with things that wait on I/O, like file or socket access, HTTP requests, etc.
Once you know which pieces to optimize, start identifying the ones that can run on top of each other. The more you can group together, the better. A great example is code that needs information from several REST APIs that don’t depend on each other. You can use
aiohttp and make all the calls in parallel instead of waiting for each one to finish before getting the next one.
Now it’s a matter of loading up those blocks of code into the main event loop. There’s a few ways of doing that, I like putting them into async functions and using
asyncio.ensure_future() to put them in the loop and
loop.run_until_complete() to wait for completion:
import asyncio import aiohttp async def fetch(url): response = await aiohttp.request('GET', url) return await response.text() loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.gather( asyncio.ensure_future(fetch("http://www.google.com")), asyncio.ensure_future(fetch("http://www.github.com")), asyncio.ensure_future(fetch("http://www.reddit.com")) ))
This is a similar example to one I used in a previous article: Threaded Asynchronous Magic and How to Wield it.
asyncio.ensure_future() turns the functions into coroutines,
asyncio.gather() groups them together, while
loop.run_until_complete() blocks execution until all calls complete. The output of this is a list with the results from each call.
Following the points discussed so far will produce code that runs synchronous blocks. But some of those blocks will execute several asynchronous functions together.
Use a thread
Also discussed in my earlier article, it’s not hard to create a separate thread that operates as a worker. It runs its own event loop and you use thread-safe asyncio methods to give it work. The nice part is you can give it synchronous work with
call_soon() or async work with
from threading import Thread ... def start_background_loop(loop): asyncio.set_event_loop(loop) loop.run_forever() # Create a new loop new_loop = asyncio.new_event_loop() # Assign the loop to another thread t = Thread(target=start_background_loop, args=(new_loop,)) t.start() # Give it some async work future = asyncio.run_coroutine_threadsafe( fetch("http://www.google.com"), new_loop ) # Wait for the result print(future.result()) # Do it again but with a callback asyncio.run_coroutine_threadsafe( fetch("http://www.github.com"), new_loop ).add_done_callback(lambda future: print(future.result()))
We get a
Future back from
run_coroutine_threadsafe which we can wait on using the
result(timeout) method, or add a callback with
add_done_callback(function). The callback function will receive the future as an argument.
Supporting both async and sync calls in the same API methods
Let’s look at something more complicated. What if you have a library or module where most functions can run in parallel, but you only want to do so if the caller is async?
We can take advantage of the threaded model here because the scheduler methods are synchronous. Meaning that your user doesn’t need to declare itself as async and have to deal with the async creep in their code. Asynchronous blocks remain contained to your module.
It also allows for api interfaces that can choose whether to use asyncio or not. In fact, we can even go one step further and auto-detect when to be async using some inspect magic.
Without threads you have no control over your event loop. Users could do their own async fiddling that interferes with how your methods execute. The thread will at least guarantee asynchronous execution inside an event loop that you operate. One that you start and stop when needed. This leads to more predictable, repeatable results.
Let’s look at an example that builds on the earlier ones. Here we make a wrapper method that calls the appropriate sync or async function based on its caller.
import inspect import requests ... def is_async_caller(): """Figure out who's calling.""" # Get the calling frame caller = inspect.currentframe().f_back.f_back # Pull the function name from FrameInfo func_name = inspect.getframeinfo(caller) # Get the function object f = caller.f_locals.get( func_name, caller.f_globals.get(func_name) ) # If there's any indication that the function object is a # coroutine, return True. inspect.iscoroutinefunction() should # be all we need, the rest are here to illustrate. if any([inspect.iscoroutinefunction(f), inspect.isgeneratorfunction(f), inspect.iscoroutine(f), inspect.isawaitable(f), inspect.isasyncgenfunction(f) , inspect.isasyncgen(f)]): return True else: return False def fetch(url): """GET the URL, do it asynchronously if the caller is async""" # Figure out which function is calling us if is_async_caller(): print("Calling ASYNC method") # Run the async version of this method and # print the result with a callback asyncio.run_coroutine_threadsafe( _async_fetch(url), new_loop ).add_done_callback(lambda f: print(f.result())) else: print("Calling BLOCKING method") # Run the synchronous version and print the result print(_sync_fetch(url)) def _sync_fetch(url): """Blocking GET""" return requests.get(url).content async def _async_fetch(url): """Async GET""" resp = await aiohttp.request('GET', url) return await resp.text() def call_sync_fetch(): """Blocking fetch call""" fetch("http://www.github.com") async def call_async_fetch(): """Asynchronous fetch call (no different from sync call except this function is defined async)""" fetch("http://www.github.com") # Perform a blocking GET call_sync_fetch() # Perform an async GET loop = asyncio.get_event_loop() loop.run_until_complete(call_async_fetch())
is_async_caller() to get the function object that called us and determine whether its a coroutine or not. While this is fancy and illustrates the possibilities, it may not be very performant. We could easily replace the mechanism with an
async_execute argument in the
fetch wrapper and have the user decide.
call_async_fetch functions are there to show how a user could go about calling our wrapper. As you can see, there’s no need to await the fetch call because it’s done automatically by running in a separate thread.
This could prove useful to any python packages that are wanting to add support for asynchronous execution while still supporting legacy code. I’m sure there are pros and cons, feel free to start the discussion in the comments below.