Skip to content

Instantly share code, notes, and snippets.

@jorenham
Last active August 29, 2022 23:38
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jorenham/6806d494a482e3543c8a1f4b6d774c1d to your computer and use it in GitHub Desktop.
Save jorenham/6806d494a482e3543c8a1f4b6d774c1d to your computer and use it in GitHub Desktop.
Directly measures the asyncio event loop iteration speed, and compares it with a sync for-loop
import asyncio
import time
from time import perf_counter_ns
# noinspection PyPep8Naming
class ns_per_it:
__slots__ = 'n', 'res'
def __init__(self, n: int):
assert n > 0
self.n = n
self.res = None
def __await__(self):
n = self.n
t_0 = perf_counter_ns()
for _ in range(n):
yield None
t_n = perf_counter_ns()
self.res = round((t_n - t_0) / n)
return self.res
async def wait(self):
return await self
def _measure(sync: bool, n: int):
if sync:
obj = ns_per_it(n)
for _ in obj.__await__():
pass
res = obj.res
assert res is not None
else:
res = asyncio.run(ns_per_it(n).wait())
it_per_s = round(1e9 / res)
print(f'{res:>10_} ns/it')
print(f'{it_per_s:>10_} it/s')
def _cooldown(dt: int):
print()
print(f'cooling down for {dt}s...')
time.sleep(dt)
print()
def main():
N = 10_000_000
DT = 42
print('sync:')
_measure(True, N)
_cooldown(DT)
print('async:')
_measure(False, N)
try:
import uvloop
except ImportError:
uvloop = None
else:
_cooldown(DT)
print('async + uvloop:')
uvloop.install()
_measure(False, N)
if __name__ == '__main__':
main()
@jorenham
Copy link
Author

example output on Python 3.10 with an Intel i7-7700HQ:

sync:
        44 ns/it
22_727_273 it/s

cooling down for 42s...

async:
     3_514 ns/it
   284_576 it/s

cooling down for 42s...

async + uvloop:
     1_225 ns/it
   816_327 it/s

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment