Asyncio Benchmark Decorator

November 7, 2023 Python Benchmarking

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:

...
# 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:

...
# 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:

Benchmark Python with time.perf_counter()

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?

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:

# 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:

# 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:

# 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:

# 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:

Benchmark Decorator in Python

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:

# 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:

@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.

# 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:

# 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:

# 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.

# 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.

...
from time import perf_counter
from functools import wraps

Finally, to use the decorator, we add "@benchmark" above the target coroutine.

For example:

@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:

# 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.

# 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.

# 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.

Main starting
work Took 6.305 seconds
Main done

Takeaways

You now know how to automatically benchmark target coroutines using a benchmark decorator in Python.

If you enjoyed this tutorial, you will love my book: Python Benchmarking. It covers everything you need to master the topic with hands-on examples and clear explanations.