Some checks failed
Lint Code / lint (push) Failing after 44s
CI/CD Pipeline / lint (push) Successful in 0s
Run Tests / test (3.12) (push) Successful in 3m48s
Run Tests / test (3.13) (push) Successful in 3m7s
CI/CD Pipeline / test (push) Successful in 1s
CI/CD Pipeline / build-and-release (push) Has been skipped
CI/CD Pipeline / notify (push) Successful in 1s
154 lines
4.7 KiB
Python
154 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Benchmark script for routing performance comparison.
|
|
|
|
Compares:
|
|
- Pure Python implementation with standard re (_routing_py)
|
|
- Cython implementation with PCRE2 JIT (_routing)
|
|
|
|
Usage:
|
|
python benchmarks/bench_routing.py
|
|
"""
|
|
|
|
import re
|
|
import time
|
|
import statistics
|
|
from typing import Callable, Tuple
|
|
|
|
from pyserve._routing_py import (
|
|
FastRouter as PyFastRouter,
|
|
FastRouteMatch as PyFastRouteMatch,
|
|
)
|
|
|
|
try:
|
|
from pyserve._routing import (
|
|
FastRouter as CyFastRouter,
|
|
FastRouteMatch as CyFastRouteMatch,
|
|
)
|
|
CYTHON_AVAILABLE = True
|
|
except ImportError:
|
|
CYTHON_AVAILABLE = False
|
|
print("Cython module not compiled. Run: poetry run python scripts/build_cython.py\n")
|
|
|
|
|
|
def benchmark(func: Callable, iterations: int = 100000) -> Tuple[float, float]:
|
|
"""Benchmark a function and return mean/stdev in nanoseconds."""
|
|
times = []
|
|
|
|
# Warmup
|
|
for _ in range(1000):
|
|
func()
|
|
|
|
# Actual benchmark
|
|
for _ in range(iterations):
|
|
start = time.perf_counter_ns()
|
|
func()
|
|
end = time.perf_counter_ns()
|
|
times.append(end - start)
|
|
|
|
return statistics.mean(times), statistics.stdev(times)
|
|
|
|
|
|
def format_time(ns: float) -> str:
|
|
"""Format time in nanoseconds to human readable format."""
|
|
if ns < 1000:
|
|
return f"{ns:.1f} ns"
|
|
elif ns < 1_000_000:
|
|
return f"{ns/1000:.2f} µs"
|
|
else:
|
|
return f"{ns/1_000_000:.2f} ms"
|
|
|
|
|
|
def setup_router(router_class):
|
|
"""Setup a router with typical routes."""
|
|
router = router_class()
|
|
|
|
# Exact routes
|
|
router.add_route("=/health", {"return": "200 OK"})
|
|
router.add_route("=/api/status", {"return": "200 OK"})
|
|
router.add_route("=/favicon.ico", {"return": "204"})
|
|
|
|
# Regex routes
|
|
router.add_route("~^/api/v1/users/(?P<user_id>\\d+)$", {"proxy_pass": "http://users-service"})
|
|
router.add_route("~^/api/v1/posts/(?P<post_id>\\d+)$", {"proxy_pass": "http://posts-service"})
|
|
router.add_route("~\\.(css|js|png|jpg|gif|svg|woff2?)$", {"root": "./static"})
|
|
router.add_route("~^/api/", {"proxy_pass": "http://api-gateway"})
|
|
|
|
# Default route
|
|
router.add_route("__default__", {"spa_fallback": True, "root": "./dist"})
|
|
|
|
return router
|
|
|
|
|
|
def run_benchmarks():
|
|
print("=" * 70)
|
|
print("ROUTING BENCHMARK")
|
|
print("=" * 70)
|
|
print()
|
|
|
|
# Test paths with different matching scenarios
|
|
test_cases = [
|
|
("/health", "Exact match (first)"),
|
|
("/api/status", "Exact match (middle)"),
|
|
("/api/v1/users/12345", "Regex match with groups"),
|
|
("/static/app.js", "Regex match (file extension)"),
|
|
("/api/v2/other", "Regex match (simple prefix)"),
|
|
("/some/random/path", "Default route (fallback)"),
|
|
("/nonexistent", "Default route (fallback)"),
|
|
]
|
|
|
|
iterations = 100000
|
|
|
|
print(f"Iterations: {iterations:,}")
|
|
print()
|
|
|
|
# Setup routers
|
|
py_router = setup_router(PyFastRouter)
|
|
cy_router = setup_router(CyFastRouter) if CYTHON_AVAILABLE else None
|
|
|
|
results = {}
|
|
|
|
for path, description in test_cases:
|
|
print(f"Path: {path}")
|
|
print(f" {description}")
|
|
|
|
# Python implementation (standard re)
|
|
py_mean, py_std = benchmark(lambda p=path: py_router.match(p), iterations)
|
|
results[(path, "Python (re)")] = py_mean
|
|
print(f" Python (re): {format_time(py_mean):>12} ± {format_time(py_std)}")
|
|
|
|
# Cython implementation (PCRE2 JIT)
|
|
if CYTHON_AVAILABLE and cy_router:
|
|
cy_mean, cy_std = benchmark(lambda p=path: cy_router.match(p), iterations)
|
|
results[(path, "Cython (PCRE2)")] = cy_mean
|
|
speedup = py_mean / cy_mean if cy_mean > 0 else 0
|
|
print(f" Cython (PCRE2): {format_time(cy_mean):>12} ± {format_time(cy_std)} ({speedup:.2f}x faster)")
|
|
|
|
print()
|
|
|
|
# Summary
|
|
if CYTHON_AVAILABLE:
|
|
print("=" * 70)
|
|
print("SUMMARY")
|
|
print("=" * 70)
|
|
|
|
py_total = sum(v for k, v in results.items() if k[1] == "Python (re)")
|
|
cy_total = sum(v for k, v in results.items() if k[1] == "Cython (PCRE2)")
|
|
|
|
print(f" Python (re) total: {format_time(py_total)}")
|
|
print(f" Cython (PCRE2) total: {format_time(cy_total)}")
|
|
print(f" Overall speedup: {py_total / cy_total:.2f}x")
|
|
|
|
# Show JIT compilation status
|
|
print()
|
|
print("PCRE2 JIT Status:")
|
|
for route in cy_router.list_routes(): # type: ignore False linter error
|
|
if route["type"] == "regex":
|
|
jit = route.get("jit_compiled", False)
|
|
status = "✓ JIT" if jit else "✗ No JIT"
|
|
print(f" {status}: {route['pattern']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_benchmarks()
|