async_to_sync microbenchmark

async_to_sync_bench.py

# /// script
# dependencies = [
#   "asgiref",
#   "pandas",
#   "numpy"
# ]
# ///

# this is a microbenchmark of async_to_sync, to get an idea 
# of how much that mechanism itself costs in terms of latency

# run with: uv run async_to_sync.py

from asgiref.sync import async_to_sync, sync_to_async

import time
import pandas as pd
import numpy as np

NUMBER_OPS = 100_000

# metrics[operation_idx] = (start_time, end_time)
metrics = []


async def operation(start_time):
    end_time = time.time_ns()
    metrics.append((start_time, end_time))


wrapped_operation = async_to_sync(operation)
wrapped_new_loop_operation = async_to_sync(operation, force_new_loop=True)


def bench1():
    # just running async_to_sync
    # (but not running the context manager init over and over again)
    for _ in range(NUMBER_OPS):
        wrapped_operation(time.time_ns())


def bench2():
    # running async_to_sync but forcing new loops
    for _ in range(NUMBER_OPS):
        wrapped_new_loop_operation(time.time_ns())


def bench3():
    # running async_to_sync within some async iterations
    async def main_loop():
        for _ in range(NUMBER_OPS):
            await sync_to_async(wrapped_operation)(time.time_ns())

    import asyncio

    loop = asyncio.new_event_loop()
    loop.run_until_complete(main_loop())


def bench4():
    # Running async_to_sync (including ctx manager buildup)
    for _ in range(NUMBER_OPS):
        async_to_sync(operation)(time.time_ns())


def bench5():
    # Running async_to_sync (including ctx manager buildup)
    # (with force_new_loop)
    for _ in range(NUMBER_OPS):
        async_to_sync(operation, force_new_loop=True)(time.time_ns())


def run_bench(bench):
    metrics.clear()
    bench()
    results = calc_metrics(bench.__name__)
    return results


PERCENTILES = [10, 25, 50, 75, 95, 99, 99.9]


def calc_metrics(name):
    duration_data_ms = [(m[1] - m[0]) / 1_000_000 for m in metrics]
    percentiles = np.percentile(duration_data_ms, [10, 25, 50, 75, 95, 99, 99.99])
    return {
        "name": name,
        **{
            f"P{percentile}": percentiles[idx]
            for idx, percentile in enumerate(PERCENTILES)
        },
    }


def main():
    print(f"Running with {NUMBER_OPS:,} iterations...")
    benches = [bench1, bench2, bench3, bench4, bench5]
    results = []
    for bench in benches:
        print(f"Starting {bench.__name__}...")
        results.append(run_bench(bench))

    results_df = pd.DataFrame.from_records(results)
    print("PERCENTILES")
    print("-----------")
    print(results_df)
    print("(Results in ms)")


if __name__ == "__main__":
    main()