Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save ShashkovS/05791485cf78ad4f245a675edc7764c1 to your computer and use it in GitHub Desktop.
Save ShashkovS/05791485cf78ad4f245a675edc7764c1 to your computer and use it in GitHub Desktop.
import asyncio
from time import process_time as time
async def anoop() -> None:
pass
def noop() -> None:
pass
async def create_task_await(count: int = 100) -> float:
start = time()
[await anoop() for _ in range(count)]
return time() - start
async def create_task_async(count: int = 100) -> float:
start = time()
await asyncio.wait([asyncio.create_task(anoop()) for _ in range(count)])
return time() - start
async def create_task_gather(count: int = 100) -> float:
start = time()
await asyncio.gather(*(anoop() for _ in range(count)))
return time() - start
async def create_task_group(count: int = 100) -> float:
start = time()
async with asyncio.TaskGroup() as tg:
[tg.create_task(anoop()) for _ in range(count)]
return time() - start
def create_task_sync(count: int = 100) -> float:
start = time()
[noop() for _ in range(count)]
return time() - start
def bench_sync() -> None:
for count in range (100_000, 1_000_000 + 1, 100_000):
elapsed = create_task_sync(count)
try:
rate = 1 / (elapsed / count)
except ZeroDivisionError:
rate = float("inf")
print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")
async def bench_async() -> None:
for count in range (100_000, 1_000_000 + 1, 100_000):
elapsed = await create_task_async(count)
try:
rate = 1 / (elapsed / count)
except ZeroDivisionError:
rate = float("inf")
print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")
async def bench_await() -> None:
for count in range (100_000, 1_000_000 + 1, 100_000):
elapsed = await create_task_await(count)
try:
rate = 1 / (elapsed / count)
except ZeroDivisionError:
rate = float("inf")
print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")
async def bench_gather() -> None:
for count in range (100_000, 1_000_000 + 1, 100_000):
elapsed = await create_task_gather(count)
try:
rate = 1 / (elapsed / count)
except ZeroDivisionError:
rate = float("inf")
print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")
async def bench_group() -> None:
for count in range (100_000, 1_000_000 + 1, 100_000):
elapsed = await create_task_group(count)
try:
rate = 1 / (elapsed / count)
except ZeroDivisionError:
rate = float("inf")
print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")
def main() -> None:
print("Synchronous")
bench_sync()
print()
print("await")
asyncio.get_event_loop().run_until_complete(bench_await())
print()
print("asyncio.create_task()")
asyncio.get_event_loop().run_until_complete(bench_async())
print()
print("asyncio.gather()")
asyncio.get_event_loop().run_until_complete(bench_gather())
print()
print("asyncio.TaskGroup.create_task()")
asyncio.get_event_loop().run_until_complete(bench_group())
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment