You can develop a custom benchmark decorator that will automatically record and report the execution time of target coroutines in asyncio programs.
This requires defining the decorator and adding the decoration to the coroutine to be benchmarked.
In this tutorial, you will discover how to automatically benchmark target coroutines using a benchmark decorator in Python.
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 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 decorator that will benchmark our code automatically?
Run loops using all CPUs, download your FREE book to learn how.
How to Develop a Benchmark Coroutine Decorator
We can develop our custom coroutine decorator to automatically benchmark our target coroutines.
This involves a few steps, they are:
- How to develop function decorators.
- How to create a coroutine decorator
- How to develop a coroutine that benchmarks a target
Let’s dive in.
Function Decorators
A function decorator in Python allows a custom function to be called automatically that will in turn call our target function.
This can be used to insert code before and after calling our target function, such as recording the start time, end time, and calculating an overall execution duration.
A decorator can be defined as a custom function that returns a function that in turn calls our target function.
Typically, the new separate function is defined as an inner function, that is, within the called function.
For example:
|
1 2 3 4 5 6 |
# define the custom decorator def custom_decorator(func): # inner function that wraps the target function def inner_wrapper(): # call the target function func() |
The decorator can then be added to arbitrary functions in our program as follows:
|
1 2 3 4 |
# function that has the decorator @custom_decorator def task(): # ... |
The function we are adding the decorator to, e.g. task() may take arguments and may have a return value.
Therefore our inner wrapper function needs to handle this accordingly.
For example:
|
1 2 3 4 5 6 |
# define the custom decorator def custom_decorator(func): # inner function that wraps the target function def inner_wrapper(*args, **kwargs): # call the target function return func(*args, **kwargs) |
Additionally, we may want to leave the decorator on the target function and have the function look and feel unchanged, e.g. if we print the function or call help() on the function. With the decorator in place, this will not be the case.
Instead, we can have the decorator look like the target function passed in so that any interrogation of the target function looks normal.
This can be achieved by adding the functools.wraps decorator to our inner wrapper function.
This is a convenience function for invoking update_wrapper() as a function decorator when defining a wrapper function. […] Without the use of this decorator factory, the name of the example function would have been ‘wrapper’, and the docstring of the original example() would have been lost.
— functools — Higher-order functions and operations on callable objects
For example:
|
1 2 3 4 5 6 7 |
# define the custom decorator def custom_decorator(func): # inner function that wraps the target function @wraps(func) def inner_wrapper(*args, **kwargs): # call the target function return func(*args, **kwargs) |
And that’s about it for decorators for now.
You can learn more about how to develop a function decorator to benchmark arbitrary Python functions (not coroutines) in the tutorial:
Next, let’s consider how we can develop a decorator for asyncio coroutines.
Coroutine Decorators
We can develop a decorator for coroutine instead of functions.
A decorator for a coroutine must return a coroutine. This is because the caller of the original coroutine will await the object that is returned.
This can be achieved by changing the inner function to a coroutine, e.g. defined using the “async def” expression.
Instead of calling a target function, the inner coroutine must await the target coroutine and pass along any arguments.
For example:
|
1 2 3 4 5 6 7 |
# define the custom decorator def custom_decorator(coro): # inner coroutine that wraps the target coroutine @wraps(coro) def inner_wrapper(*args, **kwargs): # await the target return await coro(*args, **kwargs) |
To use the decorator on a target coroutine, we add it to the coroutine just like we did above for a target function.
For example:
|
1 2 3 |
@custom_decorator async def work(): # ... |
Next, let’s look at how we can define custom coroutine decorators to automatically benchmark our target coroutines.
Benchmark Coroutine Decorator
We can develop a coroutine decorator to benchmark a target coroutine automatically.
The decorator will be called benchmark() and take the name of the coroutine to be decorated as an argument.
|
1 2 3 |
# define the benchmark decorator def benchmark(coro): # ... |
Next, we can define the inner wrapper coroutine.
It must take arguments for the target coroutine, just in case. It then must record the start time before awaiting the target coroutine, and the end time after awaiting the target coroutine.
It then calculates the duration and reports it before returning any return value from the target coroutine itself.
For example:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
# inner coroutine that wraps the target coro @wraps(coro) async def wrapped(*args, **kwargs): # record start time time_start = perf_counter() # await the target return await coro(*args, **kwargs) # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'{coro.__name__} Took {time_duration:.3f} seconds') |
That is fine, but we can do better.
One issue is that a coroutine may be canceled at any time.
As such, we may want to record and report the execution time of the target, even if it is canceled.
This can be achieved by wrapping the awaited target coroutine in a try-finally structure and recording and reporting the execution time in the finally block.
For example:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
# inner coroutine that wraps the target coro @wraps(coro) async def wrapped(*args, **kwargs): # record start time time_start = perf_counter() try: # await the target return await coro(*args, **kwargs) finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'{coro.__name__} Took {time_duration:.3f} seconds') |
This will ensure that even if the target coroutine is canceled, it will still report the execution time.
Tying this together, the complete decorator for benchmarking target coroutines is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
# define the benchmark decorator def benchmark(coro): # inner coroutine that wraps the target coro @wraps(coro) async def wrapped(*args, **kwargs): # record start time time_start = perf_counter() try: # await the target return await coro(*args, **kwargs) finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'{coro.__name__} Took {time_duration:.3f} seconds') # return the wrapped coro return wrapped |
It requires two import statements, and these could be moved into the function itself if needed.
|
1 2 3 |
... from time import perf_counter from functools import wraps |
Finally, to use the decorator, we add “@benchmark” above the target coroutine.
For example:
|
1 2 3 |
@benchmark async def custom_coroutine(): # ... |
Now that we know how to develop a benchmark coroutine decorator, let’s look at some worked examples.
Example of Benchmarking Coroutine with a Decorator
We can explore how to use our benchmark decorator to benchmark the execution time of a custom coroutine.
In this example, we will define a custom coroutine that takes a moment to complete.
The coroutine creates a list of 100 million squared integers in a list comprehension. This will block the asyncio event loop for the duration.
For example:
|
1 2 3 4 |
# work to benchmark async def work(): # create a large list data = [i*i for i in range(100000000)] |
We can then add our @benchmark decoration to our task() coroutine.
|
1 2 3 4 5 |
# work to benchmark @benchmark async def work(): # create a large list data = [i*i for i in range(100000000)] |
Then, all we need to do is await our work() coroutine from the entry point of the program and it will be benchmarked automatically.
Tying this together, the complete example of using our coroutine decorator to estimate the duration of our work() target coroutine 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 39 40 41 42 43 |
# SuperFastPython.com # example of benchmarking an asyncio coroutine with decorator from time import perf_counter from functools import wraps import asyncio # define the benchmark decorator def benchmark(coro): # inner coroutine that wraps the target coro @wraps(coro) async def wrapped(*args, **kwargs): # record start time time_start = perf_counter() try: # await the target return await coro(*args, **kwargs) finally: # record end time time_end = perf_counter() # calculate the duration time_duration = time_end - time_start # report the duration print(f'{coro.__name__} Took {time_duration:.3f} seconds') # return the wrapped coro return wrapped # work to benchmark @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 work() # report a message print('Main done') # start the event loop asyncio.run(main()) |
Running the program starts the asyncio event loop and runs the main() coroutine.
The main() coroutine runs and reports an initial message before awaiting the work() coroutine.
The benchmark decorator runs and records the start time.
It then awaits work() coroutine with any arguments provided, in this case, no arguments.
The work() coroutine runs and creates a list of 100 million squared integers before terminating.
The inner benchmark coroutine records the end time and then calculates the duration. The duration is then reported to standard output.
The inner benchmark coroutine terminates and the main() coroutine resumes and reports a final message before the program exits.
In this case, we can see that the work() coroutine took about 6.305 seconds to complete.
This highlights how we can benchmark arbitrary Python coroutines using our benchmark decorator.
|
1 2 3 |
Main starting work Took 6.305 seconds Main done |
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.
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
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Takeaways
You now know how to automatically benchmark target coroutines using a benchmark decorator in Python.
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 Vidar Nordli-Mathisen on Unsplash
Do you have any questions?