Reference library

Concurrency & performance

asyncio, threading, multiprocessing, and profiling-friendly performance patterns.

7 matches

How to Run an Async Main with asyncio.run in Python

Show the canonical entry point for an asyncio program: define an async main, then launch it with asyncio.run.

import asyncio


async def main():
    print("Hello from async main")
    await asyncio.sleep(0.1)
    print("Done")


if __name__ == "__main__":
    asyncio.run(main())

16 0 Open

Concurrency & performance easy

How to Signal asyncio Workers to Stop with an Event in Python

Use an asyncio.Event to coordinate graceful shutdown of multiple concurrent worker tasks in Python.

asyncio events concurrency

Python

import asyncio
import random

async def worker(name, stop_event):
    while not stop_event.is_set():
        await asyncio.sleep(random.uniform(0.1, 0.5))
        print(f"Worker {name} processing...")
    print(f"Worker {name} stopped.")

async def main():
    stop_event = asyncio.Event()
    workers = [asyncio.create…

11 0 Open

Concurrency & performance easy

How to Use threading.Lock to Synchronize a Counter in Python

Safely increment a shared counter across multiple threads using threading.Lock as a mutex to prevent race conditions.

threading lock mutex

Python

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Final counter valu…

14 0 Open

Concurrency & performance easy

How to Use uvloop Faster Event Loop

Install uvloop at startup to replace asyncio's default event loop with a faster libuv-based one, with a graceful fallback when it's unavailable.

uvloop asyncio event-loop

Python

import asyncio
try:
    import uvloop
    uvloop.install()
    USING_UVLOOP = True
except ImportError:
    USING_UVLOOP = False


async def fetch_data(index):
    await asyncio.sleep(0.01)
    return f"data-{index}"


async def main():
    tasks = [fetch_data(i) for i in range(10)]
    results = await asyncio.gather(*…

14 0 Open

Concurrency & performance easy

How to set a timeout with asyncio.wait_for in Python

Use asyncio.wait_for to bound an async function with a timeout, catching TimeoutError when it exceeds the limit.

asyncio timeout concurrency

Python

import asyncio

async def slow_task():
    await asyncio.sleep(3)
    return "finished"

async def main():
    try:
        result = await asyncio.wait_for(slow_task(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out")

if __name__ == "__main__":
    asyncio.run(main())

13 0 Open

Concurrency & performance easy

Run Background Tasks with asyncio.create_task in Python

Create background tasks in an asyncio event loop with asyncio.create_task and run them concurrently using asyncio.gather.

asyncio async concurrency

Python

import asyncio
import time

async def background_worker(name, duration):
    """Simulates a long-running background task."""
    print(f"{name} started at t={time.monotonic():.1f}")
    await asyncio.sleep(duration)
    print(f"{name} finished at t={time.monotonic():.1f}")

async def main():
    print(f"Main starting …

13 0 Open

Concurrency & performance easy

asyncio sleep cooperative scheduling demo in Python

This demo shows how asyncio.sleep yields control between concurrent tasks, letting multiple workers interleave their ticks.

asyncio concurrency scheduling

Python

import asyncio

async def worker(name, delay):
    for i in range(3):
        print(f"{name}: tick {i}")
        await asyncio.sleep(delay)
    return f"{name} done"

async def main():
    tasks = [
        asyncio.create_task(worker("A", 0.1)),
        asyncio.create_task(worker("B", 0.2)),
        asyncio.create_tas…

12 0 Open

Browse by section

Each section groups closely related Python snippets.