You can develop a helper coroutine to record and report the overall execution time of coroutines and tasks in asyncio programs.
The helper coroutine can be implemented using a try-finally block so that it is still able to report the overall benchmark execution time even if the target task fails with an exception or is canceled.
In this tutorial, you will discover how to develop a helper coroutine to benchmark asyncio tasks and coroutines.
Let’s get started.
Help to Benchmark Asyncio Coroutines and Tasks
We can benchmark Python code using the time module.
The time.perf_counter() function will return a value from a high-performance counter.
Return the value (in fractional seconds) of a performance counter, i.e. a clock with the highest available resolution to measure a short duration.
— time — Time access and conversions
The difference between the two calls to the time.perf_counter() function can provide a high-precision estimate of the execution time of a block of code.
Unlike the time.time() function, the time.perf_counter() function is not subject to updates, such as daylight saving and synchronizing the system clock with a time server. This makes the time.perf_counter() function is a reliable approach to benchmarking Python code.
We can call the time.perf_counter() function at the beginning of the code we wish to benchmark, and again at the end of the code we wish to benchmark.
For example:
|
1 2 3 4 5 6 7 |
... # record start time time_start = time.perf_counter() # call benchmark code task() # record end time time_end = time.perf_counter() |
The difference between the start and end time is the total duration of the program in seconds.
For example:
|
1 2 3 4 5 |
... # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') |
You can learn more about benchmarking Python code with the time.perf_counter() function in the tutorial:
This approach to benchmarking can be used to benchmark asyncio programs that await coroutines and tasks.
How can we hide all of this code so that we can benchmark with a simple interface?
Can we develop a custom coroutine that will benchmark our code automatically?
Run loops using all CPUs, download your FREE book to learn how.
How to Develop a Benchmark Coroutine for Asyncio
We can develop a helper coroutine to automatically benchmark our Python code.
Our coroutine can take an awaitable to run. This could be a created coroutine object or an instance of an asyncio.Task.
For example:
|
1 2 3 |
# benchmark coroutine async def benchmark(awaitable): # ... |
We want our benchmark() to be a coroutine so that we can await it and the target awaitable will execute.
Recall that when we call a coroutine we are in fact creating an instance of a coroutine object that is an awaitable type that can be awaited via the await expression.
You can learn more about coroutines in the tutorial:
We don’t need to provide any arguments to the target, we can assume that the awaitable has been created and is provided ready to be executed.
Our coroutine can then record the start time, await target awaitable, record the end time, and report the overall duration.
Importantly, we need to handle the case that the target is canceled for some reason, or perhaps terminated due to an exception.
Recall that a task can be canceled at any time, even by the event loop when it is shut down.
We can handle this case by awaiting the target awaitable within a try-finally block and recording and reporting the duration in the finally block.
For example:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
... # record start time time_start = perf_counter() try: # await the target await awaitable finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') |
Tying this together with a helper coroutine for benchmarking arbitrary asyncio tasks and coroutines is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# benchmark coroutine async def benchmark(awaitable): # record start time time_start = perf_counter() try: # await the target await awaitable finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') |
Using this coroutine requires that we import the time.perf_counter() function.
|
1 |
from time import perf_counter |
To benchmark the coroutine we can create a coroutine and pass it as an argument then await the benchmark() coroutine directly.
For example:
|
1 2 3 |
... # benchmark a coroutine await benchmark(work()) |
To benchmark an asyncio.Task, we can create the task directly or via a TaskGroup then pass it to the benchmark() coroutine and await it.
For example:
|
1 2 3 |
... # benchmark a task await benchmark(task) |
Now that we know how to develop a helper benchmark coroutine, let’s look at some examples.
Example of Benchmarking a Coroutine
We can explore an example of benchmarking the execution time of a coroutine using our helper.
In this case, we will define a coroutine that takes a long time performing a CPU-bound task. We will then execute this task via our benchmark() coroutine developed above and report the overall execution time.
Firstly, we can define a coroutine that runs a long time. Our task will prepare a list of 100 million squared integers.
The work() coroutine below implements this.
|
1 2 3 4 |
# task to benchmark async def work(): # create a large list data = [i*i for i in range(100000000)] |
The main() coroutine will then execute this task via our benchmark() coroutine developed above.
|
1 2 3 4 5 6 7 8 |
# main coroutine async def main(): # report a message print('Main starting') # benchmark the execution of our task await benchmark(work()) # report a message print('Main done') |
This will run our task and report the overall execution time for the task.
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
# SuperFastPython.com # example of benchmarking an asyncio coroutine with our helper coroutine from time import perf_counter import asyncio # benchmark coroutine async def benchmark(awaitable): # record start time time_start = perf_counter() try: # await the target await awaitable finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') # task to benchmark async def work(): # create a large list data = [i*i for i in range(100000000)] # main coroutine async def main(): # report a message print('Main starting') # benchmark the execution of our task await benchmark(work()) # report a message print('Main done') # start the event loop asyncio.run(main()) |
Running the example first starts the event loop and runs the main() coroutine.
The main() coroutine runs and reports a start message.
It then suspends and awaits the benchmark() coroutine and passes in an instance of our work() coroutine.
The benchmark() coroutine runs and records the start time. It then awaits the passed-in coroutine object.
The work() coroutine runs and creates a list of 100 million squared integers. This blocks the event loop.
The work() coroutine terminates and the benchmark() coroutine resumes. It records the end time, calculates the duration, and reports the duration in seconds.
In this case, we can see that the work() coroutine took about 6.1 seconds to complete.
The benchmark() coroutine terminates and the main() coroutine resumes and reports a final message before the program exits.
This highlights how we can benchmark the execution time of a coroutine using our helper coroutine.
|
1 2 3 |
Main starting Took 6.178 seconds Main done |
Next, let’s look at how we might benchmark an asyncio task.
Free Python Benchmarking Course
Get FREE access to my 7-day email course on Python Benchmarking.
Discover benchmarking with the time.perf_counter() function, how to develop a benchmarking helper function and context manager and how to use the timeit API and command line.
Example of Benchmarking an Asyncio Task
We can explore how to benchmark the execution time of an asyncio task using our benchmark() helper coroutine.
In this case, we can update the above example to first create and schedule the work() coroutine as an asyncio.Task, then pass the asyncio.Task instance to the benchmark() coroutine.
|
1 2 3 4 5 |
... # create the task task = asyncio.create_task(work()) # benchmark the execution of our task await benchmark(task) |
The benchmark() helper coroutine can take any awaitable, such as a coroutine object or an asyncio.Task, so the target task will be awaited and its execution time reported as expected.
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
# SuperFastPython.com # example of benchmarking an asyncio task with our helper coroutine from time import perf_counter import asyncio # benchmark coroutine async def benchmark(awaitable): # record start time time_start = perf_counter() try: # await the target await awaitable finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') # task to benchmark async def work(): # create a large list data = [i*i for i in range(100000000)] # main coroutine async def main(): # report a message print('Main starting') # create the task task = asyncio.create_task(work()) # benchmark the execution of our task await benchmark(task) # report a message print('Main done') # start the event loop asyncio.run(main()) |
Running the example first starts the event loop and runs the main() coroutine.
The main() coroutine runs and reports a start message.
It then creates and schedules a new asyncio.Task instance and passes it a work() coroutine instance.
The main() coroutine then suspends and awaits the benchmark() coroutine and passes in an instance of our task instance.
The benchmark() coroutine runs and records the start time. It then awaits the passed-in asyncio.Task object.
The work() coroutine runs and creates a list of 100 million squared integers. This blocks the event loop.
The work() coroutine terminates and the benchmark() coroutine resumes. It records the end time, calculates the duration, and reports the duration in seconds.
In this case, we can see that the work() coroutine took about 6.3 seconds to complete.
The benchmark() coroutine terminates and the main() coroutine resumes and reports a final message before the program exits.
This highlights how we can benchmark the execution time of an asyncio.Task using our helper coroutine.
|
1 2 3 |
Main starting Took 6.384 seconds Main done |
Next let’s look at benchmarking a coroutine as a background task.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Benchmarking in a Background Task
We can explore an example of benchmarking an asyncio coroutine as a background task.
In this case, we can update the above example to benchmark our work() coroutine directly but to do so as a background task.
This means that the benchmark() coroutine will be scheduled and executed as an asyncio.Task while the main() coroutine proceeds with other tasks, in this case, a long sleep.
|
1 2 3 4 5 |
... # create a task to perform the benchmarking task = asyncio.create_task(benchmark(work())) # wait around doing other things await asyncio.sleep(8) |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
# SuperFastPython.com # example of benchmarking a target awaitable in a background task from time import perf_counter import asyncio # benchmark coroutine async def benchmark(awaitable): # record start time time_start = perf_counter() try: # await the target await awaitable finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') # task to benchmark async def work(): # create a large list data = [i*i for i in range(100000000)] # main coroutine async def main(): # report a message print('Main starting') # create a task to perform the benchmarking task = asyncio.create_task(benchmark(work())) # wait around doing other things await asyncio.sleep(8) # report a message print('Main done') # start the event loop asyncio.run(main()) |
The main() coroutine runs and reports a start message.
It then creates and schedules a new asyncio.Task instance and passes it a benchmark() coroutine that in turn is passed an instance of our work() coroutine.
The main() coroutine then suspends and sleeps for 8 seconds, enough time for our 6-second CPU-bound task to complete
The benchmark() background task runs and records the start time. It then awaits the passed-in coroutine object.
The work() coroutine runs and creates a list of 100 million squared integers. This blocks the event loop.
The work() coroutine terminates and the benchmark() task resumes. It records the end time, calculates the duration, and reports the duration in seconds.
In this case, we can see that the work() coroutine took about 6.0 seconds to complete. The benchmark() coroutine terminates.
Later, the main() coroutine finishes its sleep and resumes. It reports a final message before the program, and then exits.
This highlights how we can benchmark the execution time of a coroutine using our helper coroutine in the background.
|
1 2 3 |
Main starting Took 6.018 seconds Main done |
Next, let’s look at what happens if the target awaitable that is being benchmarked is canceled.
Example of Benchmarking in a Target that is Canceled
We can explore an example of the target coroutine being canceled while the execution time is being benchmarked.
In this case, we will update the above example so that the target task can be canceled, e.g., is awaiting a call to asyncio.sleep().
|
1 2 3 4 |
# task to benchmark async def work(): # create a large list await asyncio.sleep(5) |
We will then benchmark our work() coroutine in the background, but update the main() sleep so that it does not give enough time for the work() task to complete, e.g. 3 seconds instead of 5 or more seconds.
|
1 2 3 4 5 |
... # create a task to perform the benchmarking task = asyncio.create_task(benchmark(work())) # wait around doing other things await asyncio.sleep(3) |
This will mean that the main() coroutine will exit and shut down the event loop while the benchmark() coroutine is awaiting the work() coroutine, canceling both.
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
# SuperFastPython.com # example of benchmarking a target that is cancelled from time import perf_counter import asyncio # benchmark coroutine async def benchmark(awaitable): # record start time time_start = perf_counter() try: # await the target await awaitable finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'Took {time_duration:.3f} seconds') # task to benchmark async def work(): # create a large list await asyncio.sleep(5) # main coroutine async def main(): # report a message print('Main starting') # create a task to perform the benchmarking task = asyncio.create_task(benchmark(work())) # wait around doing other things await asyncio.sleep(3) # report a message print('Main done') # start the event loop asyncio.run(main()) |
The main() coroutine runs and reports a start message.
It then creates and schedules a new asyncio.Task instance and passes it a benchmark() coroutine that in turn is passed an instance of our work() coroutine.
The main() coroutine then suspends and sleeps for 3 seconds, not enough time for our 5-second work() coroutine to complete
The benchmark() background task runs and records the start time. It then awaits the passed-in coroutine object.
The work() coroutine runs and sleeps for 5 seconds.
The main() coroutine resumes and exits. This shuts down the asyncio event loop and cancels all running tasks.
The work() coroutine is canceled, raising a CancelledError exception and terminating.
The CancelledError exception is raised in the benchmark() which exits, first executing the finally block. This records the end time, calculates the duration, and reports the duration in seconds.
We can see that the work() task execution time is trimmed to about 3.0 seconds, instead of the expected 5 seconds, as we expect.
This highlights how we can still report the benchmark execution time of coroutines and tasks that are canceled while being benchmarked.
|
1 2 3 |
Main starting Main done Took 3.003 seconds |
Further Reading
This section provides additional resources that you may find helpful.
Books
- Python Benchmarking, Jason Brownlee (my book!)
Also, the following Python books have chapters on benchmarking that may be helpful:
- Python Cookbook, 2013. (sections 9.1, 9.10, 9.22, 13.13, and 14.13)
- High Performance Python, 2020. (chapter 2)
Guides
- 4 Ways to Benchmark Python Code
- 5 Ways to Measure Execution Time in Python
- Python Benchmark Comparison Metrics
Benchmarking APIs
- time — Time access and conversions
- timeit — Measure execution time of small code snippets
- The Python Profilers
References
Takeaways
You now know how to develop a helper coroutine to benchmark asyncio tasks and coroutines.
Did I make a mistake? See a typo?
I’m a simple humble human. Correct me, please!
Do you have any additional tips?
I’d love to hear about them!
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Mário Rui André on Unsplash
Do you have any questions?