uvloop is a fast drop-in asyncio event loop built on libuv. pip install uvloop. Install globally: import uvloop; uvloop.install() — call before any asyncio code. Or: asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()). Or per-run: uvloop.run(main()). Check active loop: type(asyncio.get_event_loop()) → <class 'uvloop.Loop'>. Gunicorn: --worker-class uvloop.UVLoopWorker (or aiohttp.GunicornUVLoopWebWorker). aiohttp: web.run_app(app, loop=uvloop.new_event_loop()). FastAPI+uvicorn: uvicorn uses uvloop by default when installed (--loop uvloop flag). asyncio streams: fully compatible — asyncio.open_connection, asyncio.start_server. Subprocess: asyncio.create_subprocess_exec. Transport/protocol: compatible with raw asyncio protocols. Threads: loop.run_in_executor. Selector: uvloop replaces the selector event loop with libuv’s I/O poller. Platform: Linux and macOS only (not Windows). Performance: 2–4× faster than default asyncio for I/O-heavy workloads. uvloop.new_event_loop() — create a loop instance manually. loop.run_until_complete(coro). loop.run_forever(). Claude Code generates uvloop-accelerated FastAPI servers, aiohttp services, asyncpg database workers, and benchmarking harnesses.
CLAUDE.md for uvloop
## uvloop Stack
- Version: uvloop >= 0.19 | pip install uvloop
- Install: uvloop.install() at module top (before asyncio imports complete)
- Policy: asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
- Run: uvloop.run(main()) — equivalent to asyncio.run() with uvloop
- Uvicorn: uvicorn myapp:app --loop uvloop (default when uvloop installed)
- Platform: Linux + macOS only (no Windows support)uvloop Async Performance Pipeline
# app/fast_loop.py — uvloop installation, benchmarking, aiohttp, asyncpg patterns
from __future__ import annotations
import asyncio
import statistics
import time
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator, Callable, Coroutine
import uvloop
# ─────────────────────────────────────────────────────────────────────────────
# 1. Installation helpers
# ─────────────────────────────────────────────────────────────────────────────
def install(force: bool = False) -> bool:
"""
Install uvloop as the global asyncio event loop policy.
Must be called before asyncio.run() or any loop creation.
Returns True if uvloop is now active, False if already set (or force skipped).
Usage (top of main.py or app/__init__.py):
from app.fast_loop import install
install()
"""
policy = asyncio.get_event_loop_policy()
if isinstance(policy, uvloop.EventLoopPolicy) and not force:
return False # Already installed
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
return True
def run(coro: Coroutine, *, debug: bool = False) -> Any:
"""
Drop-in replacement for asyncio.run() that uses uvloop.
Equivalent to asyncio.run(coro) but with the uvloop event loop.
Usage:
uvloop_run(main())
"""
return uvloop.run(coro, debug=debug)
def new_loop() -> uvloop.Loop:
"""Create a new uvloop event loop instance."""
return uvloop.new_event_loop()
def active_loop_name() -> str:
"""Return the class name of the running event loop."""
try:
loop = asyncio.get_running_loop()
return type(loop).__name__
except RuntimeError:
return "no running loop"
def is_uvloop_active() -> bool:
"""Check whether uvloop is the currently active policy."""
return isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Benchmarking helpers
# ─────────────────────────────────────────────────────────────────────────────
async def _bench_once(
coro_factory: Callable[[], Coroutine],
concurrency: int = 100,
) -> float:
"""Run `concurrency` coroutines concurrently and return elapsed seconds."""
start = time.perf_counter()
await asyncio.gather(*[coro_factory() for _ in range(concurrency)])
return time.perf_counter() - start
async def benchmark(
coro_factory: Callable[[], Coroutine],
concurrency: int = 100,
rounds: int = 5,
warmup: int = 1,
label: str = "task",
) -> dict[str, float]:
"""
Benchmark an async workload under uvloop.
coro_factory: zero-arg callable that returns a fresh coroutine each call.
concurrency: tasks per round.
rounds: measured rounds (after warmup).
warmup: un-measured warm-up rounds.
Returns dict with min/max/median/mean elapsed seconds and ops/sec.
Example:
async def fetch():
async with session.get("http://localhost/ping") as r:
await r.read()
stats = await benchmark(fetch, concurrency=500, rounds=5)
print(f"median latency: {stats['median_ms']:.1f}ms ops/sec: {stats['ops_per_sec']:.0f}")
"""
for _ in range(warmup):
await _bench_once(coro_factory, concurrency)
times: list[float] = []
for _ in range(rounds):
elapsed = await _bench_once(coro_factory, concurrency)
times.append(elapsed)
return {
"label": label,
"concurrency": concurrency,
"rounds": rounds,
"min_ms": min(times) * 1000,
"max_ms": max(times) * 1000,
"median_ms": statistics.median(times) * 1000,
"mean_ms": statistics.mean(times) * 1000,
"ops_per_sec": (concurrency * rounds) / sum(times),
"loop": active_loop_name(),
}
async def compare_loops(
coro_factory: Callable[[], Coroutine],
concurrency: int = 200,
rounds: int = 3,
label: str = "workload",
) -> dict[str, dict]:
"""
Compare uvloop vs default asyncio event loop for the same workload.
Returns {"uvloop": stats, "asyncio": stats}.
Note: switches policy back and forth — not safe in production,
use only for benchmarking scripts.
Example:
results = await compare_loops(lambda: asyncio.sleep(0), concurrency=1000)
speedup = results["asyncio"]["median_ms"] / results["uvloop"]["median_ms"]
print(f"uvloop is {speedup:.2f}x faster")
"""
results: dict[str, dict] = {}
# Benchmark with uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
results["uvloop"] = await benchmark(coro_factory, concurrency, rounds, label=label)
# Benchmark with default asyncio
asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
results["asyncio"] = await benchmark(coro_factory, concurrency, rounds, label=label)
# Restore uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
return results
# ─────────────────────────────────────────────────────────────────────────────
# 3. aiohttp server pattern
# ─────────────────────────────────────────────────────────────────────────────
AIOHTTP_EXAMPLE = '''
# Install uvloop before creating the app
import uvloop
uvloop.install()
from aiohttp import web
async def handle(request):
return web.Response(text="Hello, uvloop!")
async def health(request):
return web.json_response({"status": "ok", "loop": type(asyncio.get_running_loop()).__name__})
app = web.Application()
app.router.add_get("/", handle)
app.router.add_get("/health", health)
if __name__ == "__main__":
# uvloop.install() already called above — aiohttp uses it automatically
web.run_app(app, host="0.0.0.0", port=8080)
'''
# ─────────────────────────────────────────────────────────────────────────────
# 4. FastAPI + uvicorn pattern
# ─────────────────────────────────────────────────────────────────────────────
FASTAPI_EXAMPLE = '''
# FastAPI with uvicorn — uvloop is used automatically when installed
# Install uvloop: pip install uvloop
# main.py
import uvloop
uvloop.install() # must be before any asyncio.run / uvicorn start
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio
@asynccontextmanager
async def lifespan(app: FastAPI):
loop_name = type(asyncio.get_running_loop()).__name__
print(f"Starting with event loop: {loop_name}")
yield
app = FastAPI(lifespan=lifespan)
@app.get("/health")
async def health():
return {"loop": type(asyncio.get_running_loop()).__name__}
# Run with: uvicorn main:app --loop uvloop --host 0.0.0.0 --port 8000
# Or programmatically:
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000, loop="uvloop")
'''
# ─────────────────────────────────────────────────────────────────────────────
# 5. asyncpg / database pattern
# ─────────────────────────────────────────────────────────────────────────────
ASYNCPG_EXAMPLE = '''
import uvloop
uvloop.install() # asyncpg benefits significantly from uvloop
import asyncio
import asyncpg
DATABASE_URL = "postgresql://user:pass@localhost/mydb"
async def init_pool():
return await asyncpg.create_pool(DATABASE_URL, min_size=5, max_size=20)
async def fetch_users(pool, limit=100):
async with pool.acquire() as conn:
return await conn.fetch("SELECT id, name, email FROM users LIMIT $1", limit)
async def main():
pool = await init_pool()
users = await fetch_users(pool)
print(f"Fetched {len(users)} users with {type(asyncio.get_running_loop()).__name__}")
await pool.close()
asyncio.run(main()) # Uses uvloop because uvloop.install() was called
'''
# ─────────────────────────────────────────────────────────────────────────────
# 6. Gunicorn deployment
# ─────────────────────────────────────────────────────────────────────────────
GUNICORN_EXAMPLE = '''
# gunicorn.conf.py — uvloop with gunicorn aiohttp worker
worker_class = "aiohttp.GunicornUVLoopWebWorker"
workers = 4
bind = "0.0.0.0:8000"
keepalive = 5
timeout = 30
# For FastAPI/ASGI with uvicorn workers:
# gunicorn myapp:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
# Or for maximum performance with uvloop worker:
# pip install uvicorn[standard] # includes uvloop
# gunicorn myapp:app -w 4 -k uvicorn.workers.UvicornWorker
'''
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
async def _do_nothing():
"""Ultra-lightweight coroutine for event loop overhead benchmarking."""
await asyncio.sleep(0)
async def _cpu_task(n: int = 10_000):
"""Simulate mixed async + CPU work."""
total = sum(range(n))
await asyncio.sleep(0)
return total
async def demo():
print(f"Active event loop: {active_loop_name()}")
print(f"uvloop installed: {is_uvloop_active()}")
print("\n=== Benchmarking asyncio.sleep(0) — 500 concurrent, 3 rounds ===")
stats = await benchmark(lambda: asyncio.sleep(0), concurrency=500, rounds=3,
label="sleep(0)")
print(f" loop: {stats['loop']}")
print(f" median: {stats['median_ms']:.2f}ms")
print(f" ops/sec: {stats['ops_per_sec']:.0f}")
print("\n=== uvloop vs asyncio comparison (100 concurrent) ===")
results = await compare_loops(lambda: asyncio.sleep(0), concurrency=100, rounds=3,
label="sleep(0)")
for name, r in results.items():
print(f" {name:8s}: median={r['median_ms']:.2f}ms ops/sec={r['ops_per_sec']:.0f}")
uv_med = results["uvloop"]["median_ms"]
aio_med = results["asyncio"]["median_ms"]
if aio_med > 0:
print(f"\n uvloop speedup: {aio_med / uv_med:.2f}x")
if __name__ == "__main__":
install() # Set uvloop globally
uvloop.run(demo()) # Run with uvloopFor the asyncio (default) alternative — Python’s built-in asyncio event loop uses a pure-Python selector implementation; uvloop replaces it with a C extension backed by libuv (the same event loop powering Node.js), giving 2–4× higher throughput for I/O-bound workloads with zero code changes beyond uvloop.install(). For the trio alternative — trio is a structured-concurrency async framework with scope-based cancellation, nurseries, and a focus on correctness; uvloop is not an alternative framework but a performance drop-in for the standard asyncio event loop — use uvloop when you want asyncio compatibility and better I/O throughput, trio when you want structured concurrency guarantees and checkpoints. The Claude Skills 360 bundle includes uvloop skill sets covering install()/run() helpers, is_uvloop_active()/active_loop_name() introspection, benchmark() concurrent workload timer, compare_loops() asyncio vs uvloop measurement, aiohttp uvloop integration, FastAPI+uvicorn loop=uvloop configuration, asyncpg connection pool pattern, Gunicorn aiohttp/uvicorn worker config, and new_loop() for manual loop management. Start with the free tier to try uvloop async performance optimization code generation.