How to run multiple async tasks concurrently-python

The primary goal of asynchronous programming is to enable the execution of tasks in a non-blocking manner. This is achieved by moving from task to task until all tasks are fully executed. Shifting from one task is normally triggered if a delay is encountered on the currently executing task, the task with the delay gets paused and rescheduled for later continuation, then the execution moves to the next task in the schedule. This ensures that long-running tasks or those that may face delays will not stop other tasks from being executed.

There are two pproaches for executing multiple non-blocking tasks:

Execute Multiple tasks with asyncio.gather()

The asyncio.gather() function groups multiple tasks/coroutines for execution in a non-blocking manner.

The function has the following basic syntax:

asyncio.gather(*tasks, return_exceptions = False)

Where *tasks  parameter represents one or more tasks to be executed, tasks are usually coroutines but they can also be any other awaitable objects such as Task or Future objects. 

The gather() function returns a Future object which is itself awaitable.

import asyncio

#prints numbers
async def task1():
   for num in range(5):
      print(num)
      await asyncio.sleep(0.001) #simulate a small delay

#prints letters
async def task2():
   for let in "ABCDE":
      print(let)
      await asyncio.sleep(0.001) #simulate a small delay

async def main():
   group = asyncio.gather(task1(), task2()) #combine the tasks
   print(group)
   await group

asyncio.run(main())
 Visualize 

<_GatheringFuture pending>0A1B2C3D4E[Finished in 0.0633515459485352s]

In the above example, we defined two coroutine to represent the asynchronous tasks. One coroutine displays numbers while the other displays letters. We used the asyncio.sleep() function to simulate a small delay which when encountered causes the execution to move from one coroutine to the other. The result is that the two coroutines are executed alternately as proved by the outputs.

The gather() function returns a Future object as proved by the first line of the output. When the Future object is awaited, the grouped tasks are executed asynchronously.  When a delay, is encountered in one task, the task is paused and marked for continuation while the next coroutine in the schedule gets executed.

Our previous tasks did not return any values, however if a coroutine has a return values, we can access the values as a result of the await expression. The return values are returned in a list in the order that the tasks are given.

Let us consider an example where the awaited tasks have return values.

get return values

import asyncio

async def square(x):
   return x * x

async def main():
   group = asyncio.gather(square(20), square(100), square(200)) #combine the tasks

   results = await group
   print(results)

asyncio.run(main())
 Visualize 

[400, 10000, 40000][Finished in 0.05724762589670718s]

In the above example, we have a coroutine square, which returns the square of the number given as input. We grouped multiple calls to square as tasks  in the gather() function. As shown in the outputs, the return value of the awaited  operation is a list of the values returned by the tasks in the order that the task are entered.

By default, if an exception is raised in a running coroutine, the execution of all tasks is halted and the exception is propagated to the callers, but if the return_exceptions parameter is set to True, the execption is inserted as a value in place of that particular coroutine's return value and the execution continues with the remaining coroutines.

import asyncio

async def square(x):
   return x * x

async def main():
   group = asyncio.gather(square(20), square('ten'), square(100)) #combine the tasks
   
   results = await group
   print(results)

asyncio.run(main())
 Visualize 

Traceback (most recent call last):  File "<string>", line 12, in <module>  File "/app/.heroku/python/lib/python3.11/asyncio/runners.py", line 190, in run    return runner.run(main)           ^^^^^^^^^^^^^^^^  File "/app/.heroku/python/lib/python3.11/asyncio/runners.py", line 118, in run    return self._loop.run_until_complete(task)           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  File "/app/.heroku/python/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete    return future.result()           ^^^^^^^^^^^^^^^  File "<string>", line 4, in main  File "<string>", line 4, in squareTypeError: can't multiply sequence by non-int of type 'str'[Finished in 0.05763437692075968s]

import asyncio

async def square(x):
   return x * x

async def main():
   group = asyncio.gather(square(20), square('ten'), square(100), return_exceptions = True) #combine the tasks
   
   results = await group
   print(results)

asyncio.run(main())
 Visualize 

[400, TypeError("can't multiply sequence by non-int of type 'str'"), 10000][Finished in 0.06099662696942687s]

In the above example, the call to square('ten') raises a TypeError exception. When the return_exception is not set to True, the exception is raised normally and the program immediately stops. However, when return_exceptrions set to True as in second snippet, the raised exception is captured and inserted in place of the involved coroutine's return value in the returned list.

Execute Multiple tasks with asyncio.create_task()

A Task object wraps a coroutine and schedules it for execution as soon as possible in the event loop.

The scheduled task gets executed in a non-blocking manner, this means that the execution can shift to another task if a delay is encountered or the task takes too long.

The asyncio.create_task() function creates a Task object for the coroutine given as argument. 

import asyncio

async def display_nums():
   for num in range(5):
      print(num)
      await asyncio.sleep(0.001) #simulate a delay

async def display_letters():
   for let in 'ABCDE':
      print(let)
      await asyncio.sleep(0.001) #simulate a delay

async def main():
   task1 = asyncio.create_task(display_nums())
   task2 = asyncio.create_task(display_letters())
   await task1
   await task2
   
asyncio.run(main())
 Visualize 

0A1B2C3D4E[Finished in 0.06387769989669323s]

As you can see from the above example, the two tasks are being executed simultaneously as proved by the alternating numbers and letters in the output. The small delay caused by asyncio.sleep() calls is enough to make the execution to alternately switch between the two tasks.

Note that while it is not necessary to await the tasks in order for their execution to commence, it is important to await them at some point as this ensures that a task is fully executed. Await statements tells the execution to ensure that the awaited task/coroutine has been executed fully before closing the event loop.

Without awaiting a task, the execution may be started but never finished, consider the following example where we do not  await the tasks.

import asyncio

async def display_nums():
   for num in range(5):
      print(num)
      await asyncio.sleep(0.001) #simulate a delay

async def display_letters():
   for let in 'ABCDE':
      print(let)
      await asyncio.sleep(0.001) #simulate a delay

async def main():
   task1 = asyncio.create_task(display_nums())
   task2 = asyncio.create_task(display_letters())
   
asyncio.run(main())
 Visualize 

0A[Finished in 0.056999952998012304s]

As shown above, the execution of either task is started but never finished. When a delay is encountered in a task, the execution simply moves on and does not return to that task.

It is also important to await the tasks if we are expecting outputs from the wrapped coroutines. The return value of the await expression is the return value of the coroutine.

import asyncio

async def square(x):
    return x * x

async def main():
   task1 = asyncio.create_task(square(10))
   task2 = asyncio.create_task(square(20))
   result1 = await task1
   result2 = await task2

   print(result1)
   print(result2)
   
asyncio.run(main())
 Visualize 

100400[Finished in 0.058332601096481085s]

Conclusion

• We can use either asyncio.gather() or asyncio.create_task() functions to execute multiple asynchronous tasks in a non-blocking manner.
• The non-blocking mode of execution, ensures that a long running task or one facing delays does not block the execution of other tasks.
• The asyncio.gather() function groups multiple tasks in to a Future object. When this object is awaited, the grouped tasks gets executed in a non-blocking manner.
• The asyncio.create_task() function wraps a coroutine in a Task object. A task object schedules the wrapped coroutine for execution as soon as possible in the event loop.
• It is important to await the tasks to ensure that the will be fully executed.