|
| 1 | +"""SDK load test: fires devboxes.create bursts via AsyncRunloop. |
| 2 | +
|
| 3 | +Imports the installed package from this checkout so the benchmark always |
| 4 | +exercises the exact code here — not a separately published build. |
| 5 | +
|
| 6 | +USE_HTTP2=0 → HTTP/1.1 (httpx with http2=False) |
| 7 | +default → HTTP/2 (shared httpx pool with http2=True) |
| 8 | +REQUEST_COUNT defaults to 100 000. |
| 9 | +""" |
| 10 | + |
| 11 | +from __future__ import annotations |
| 12 | + |
| 13 | +import asyncio |
| 14 | +import math |
| 15 | +import os |
| 16 | +import time |
| 17 | + |
| 18 | +import httpx |
| 19 | + |
| 20 | +from runloop_api_client import AsyncRunloop |
| 21 | + |
| 22 | +REQUEST_COUNT = int(os.environ.get("REQUEST_COUNT", "100000")) |
| 23 | +RUNLOOP_BASE_URL = os.environ.get("RUNLOOP_BASE_URL") |
| 24 | +# HTTP/2 is the SDK default; set USE_HTTP2=0 to benchmark HTTP/1.1. |
| 25 | +USE_HTTP2 = os.environ.get("USE_HTTP2", "1") == "1" |
| 26 | +PROGRESS_INTERVAL = 2.0 |
| 27 | + |
| 28 | + |
| 29 | +def build_client() -> AsyncRunloop: |
| 30 | + kwargs: dict[str, object] = {"max_retries": 0, "timeout": 120.0} |
| 31 | + if RUNLOOP_BASE_URL: |
| 32 | + kwargs["base_url"] = RUNLOOP_BASE_URL |
| 33 | + if not USE_HTTP2: |
| 34 | + # A custom http_client disables the shared HTTP/2 pool. |
| 35 | + # Set http2: false explicitly — omitting it would select HTTP/2. |
| 36 | + kwargs["http_client"] = httpx.AsyncClient( |
| 37 | + http2=False, |
| 38 | + limits=httpx.Limits(max_connections=None, max_keepalive_connections=100), |
| 39 | + timeout=httpx.Timeout(120.0), |
| 40 | + ) |
| 41 | + return AsyncRunloop(**kwargs) # type: ignore[arg-type] |
| 42 | + |
| 43 | + |
| 44 | +async def send_request(client: AsyncRunloop, index: int, run_id: str) -> dict[str, object]: |
| 45 | + start = time.perf_counter() |
| 46 | + status: int | None = None |
| 47 | + try: |
| 48 | + await client.devboxes.create( |
| 49 | + blueprint_id="bp_nonexistent_loadtest_00000", |
| 50 | + name=f"loadtest-{run_id}-{index}", |
| 51 | + environment_variables={"TEST_VAR_1": "value_one", "TEST_VAR_2": "value_two"}, |
| 52 | + metadata={"test_run": run_id, "index": str(index)}, |
| 53 | + launch_parameters={"resource_size_request": "SMALL", "keep_alive_time_seconds": 300}, |
| 54 | + ) |
| 55 | + status = 200 |
| 56 | + except Exception as exc: |
| 57 | + status = getattr(exc, "status_code", None) |
| 58 | + return {"index": index, "latency_ms": (time.perf_counter() - start) * 1000, "status": status} |
| 59 | + |
| 60 | + |
| 61 | +def percentile(sorted_vals: list[float], p: float) -> float: |
| 62 | + idx = math.ceil(p / 100 * len(sorted_vals)) - 1 |
| 63 | + return sorted_vals[max(0, idx)] |
| 64 | + |
| 65 | + |
| 66 | +def print_metrics(results: list[dict[str, object]], wall_ms: float) -> None: |
| 67 | + latencies = sorted(float(r["latency_ms"]) for r in results) # type: ignore[arg-type] |
| 68 | + status_counts: dict[str, int] = {} |
| 69 | + for r in results: |
| 70 | + key = str(r["status"]) if r["status"] is not None else "network_error" |
| 71 | + status_counts[key] = status_counts.get(key, 0) + 1 |
| 72 | + |
| 73 | + print("\n=== Load Test Results ===") |
| 74 | + print(f"Requests: {len(results)}") |
| 75 | + print(f"Wall clock: {wall_ms / 1000:.2f}s") |
| 76 | + print(f"Throughput: {len(results) / (wall_ms / 1000):.1f} req/s") |
| 77 | + if latencies: |
| 78 | + print("\nLatency (ms):") |
| 79 | + print(f" min: {latencies[0]:.1f}") |
| 80 | + print(f" p50: {percentile(latencies, 50):.1f}") |
| 81 | + print(f" p90: {percentile(latencies, 90):.1f}") |
| 82 | + print(f" p95: {percentile(latencies, 95):.1f}") |
| 83 | + print(f" p99: {percentile(latencies, 99):.1f}") |
| 84 | + print(f" max: {latencies[-1]:.1f}") |
| 85 | + print("\nStatus codes:") |
| 86 | + for s, c in sorted(status_counts.items()): |
| 87 | + print(f" {s}: {c}") |
| 88 | + |
| 89 | + |
| 90 | +async def main() -> None: |
| 91 | + fd_limit: int | None = None |
| 92 | + try: |
| 93 | + import resource as _resource |
| 94 | + |
| 95 | + fd_limit = _resource.getrlimit(_resource.RLIMIT_NOFILE)[0] |
| 96 | + except Exception: |
| 97 | + pass |
| 98 | + |
| 99 | + if not USE_HTTP2 and fd_limit is not None and fd_limit < 10000: |
| 100 | + print(f"\nWARNING: File descriptor limit is {fd_limit}. For large HTTP/1.1 bursts, run:") |
| 101 | + print(" ulimit -n 65536") |
| 102 | + print("Or use HTTP/2 multiplexing: USE_HTTP2=1\n") |
| 103 | + |
| 104 | + client = build_client() |
| 105 | + run_id = f"run-{int(time.time())}" |
| 106 | + |
| 107 | + print(f"Starting load test: {REQUEST_COUNT} concurrent requests") |
| 108 | + print(f"Run ID: {run_id}") |
| 109 | + print(f"HTTP mode: {'HTTP/2 (shared httpx pool)' if USE_HTTP2 else 'HTTP/1.1 (httpx http2=False)'}") |
| 110 | + print(f"Base URL: {RUNLOOP_BASE_URL or '(SDK default)'}") |
| 111 | + if fd_limit is not None: |
| 112 | + print(f"FD limit: {fd_limit}") |
| 113 | + print() |
| 114 | + |
| 115 | + completed = 0 |
| 116 | + |
| 117 | + async def progress_printer() -> None: |
| 118 | + while True: |
| 119 | + await asyncio.sleep(PROGRESS_INTERVAL) |
| 120 | + pct = completed / REQUEST_COUNT * 100 |
| 121 | + print(f" progress: {completed}/{REQUEST_COUNT} ({pct:.1f}%)") |
| 122 | + |
| 123 | + async def wrapped(idx: int) -> dict[str, object]: |
| 124 | + nonlocal completed |
| 125 | + r = await send_request(client, idx, run_id) |
| 126 | + completed += 1 |
| 127 | + return r |
| 128 | + |
| 129 | + wall_start = time.perf_counter() |
| 130 | + progress_task = asyncio.create_task(progress_printer()) |
| 131 | + results = list(await asyncio.gather(*(wrapped(i) for i in range(REQUEST_COUNT)))) |
| 132 | + progress_task.cancel() |
| 133 | + wall_ms = (time.perf_counter() - wall_start) * 1000 |
| 134 | + |
| 135 | + await client.close() |
| 136 | + print_metrics(results, wall_ms) |
| 137 | + |
| 138 | + |
| 139 | +if __name__ == "__main__": |
| 140 | + asyncio.run(main()) |
0 commit comments