How to run same async function multiple times concurrent?

user_15229101 0
Viewed 2439

Setup:

So I have a clients array consisting of client objects. Also two async functions, one them is processing the incoming messages from a WebSocket and the other one is executing some API calls with the given client object.

When a certain message arrives from the WebSocket connection, function_a should detect it and start a for loop in the clients array. For each client object in the clients array I want to trigger function_b. But the function_b should run concurrent for the each client.

Example Code:

clients = {}

async def function_a(ws_message): # this is being called each time when a new message arrives
    if ws_message == 'signal_message':
        for client in clients:
             await function_b(client)

async def function_b(client):
    # get client parameters
    # make an API request with those parameters

UPDATE || Code with asyncio.gather()

So I have updated my actual code with asyncio.gather() but there is still ~500 ms delay between each function task. I am not sure but I guess this is because my function_b has some non-async blocking code thanks to requests library.

clients = {}
tasks = []

async def function_a(ws_message): 
    if ws_message == 'signal_message':
        for client in clients:
            task = asyncio.ensure_future(function_b(client))
            tasks.append(task)
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

async def function_b(client):
    # get client parameters
    # make an API request with those parameters

As I said this doesn't work either.

Issue:

Right now the function_b gets triggered without any issue, however function_b runs about 500ms and the await calls from the function_a waits for the one before it to finish. So for example the function_b process for the second client object in the clients array takes ~1000ms to finish after the WebSocket message has received. And this delay gets bigger with the each call obviously.

Solution?

Have can fix this so the each function_b process can run concurrently? Thanks in advance.

python-asyncio concurrency python
1 Answers

To launch code in parallel from within a co-routine, you have to create a "task" with it (and pause, without necessarily awaiting for the task).

The asyncio.gather call does the wrapping, calls, execution, and wraps the return values all in one go:

clients = {}

async def function_a(ws_message): # this is being called each time when a new message arrives
    if ws_message == 'signal_message':
        await asyncio.gather(*(function_b(client) for client in clients))

Full example on the interactive prompt:


In [8]: import asyncio

In [9]: async def coro1():
   ...:     await asyncio.sleep(1)
   ...:     print("step 1")
   ...:     await asyncio.sleep(1)
   ...:     print("step 2")
   ...: 
   ...: 
   ...: 

In [10]: async def coro2():
    ...:     await asyncio.gather(*(coro1() for i in range(5)))
    ...: 

In [11]: asyncio.run(coro2())
step 1
step 1
step 1
step 1
step 1
step 2
step 2
step 2
step 2
step 2

As can be seen, all co-routines run in parallel, outputing the result of the first print, and after 1s, the output of the second print call is shown.

If no parallelism is observe it may be due to the co-routine you not yielding the execution to the main loop by itself awaiting other co-routines. For CPU bound code, one can only achieve parallelism by running the functions in different threads or processes.

user_1082050
Related