How to run same async function multiple times concurrent?

Question

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.

Does function b has await pauses in its code? If it is synchronous, asyncio can't run it in parallel. — jsbueno
– jsbueno, Commented May 20, 2021 at 18:15
@jsbueno No it doesn't have any asyncio.sleep() or time.sleep() in it. Only thing that is synchronous it has are the API calls which are handled by request module. Right now I'm converting to aiohttp instead of request to see if it makes any difference at all. — Alperen Ozturk
– Alperen Ozturk, Commented May 20, 2021 at 18:30

jsbueno · Accepted Answer · 2021-05-20 18:13:54Z

4

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.

edited May 20, 2021 at 18:13

answered May 19, 2021 at 18:20

jsbueno

113k11 gold badges159 silver badges239 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

Alperen Ozturk Over a year ago

Well I have tried your solution but it didn't help at all. It still executes functions one by one.

Alperen Ozturk Over a year ago

Updated my question.

Ayaz49 Over a year ago

It does not do anything. It takes the same time as normal function does. It does not call the function concurrently.

Collectives™ on Stack Overflow

How to run same async function multiple times concurrent?

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related