diff --git a/loadtest/.gitignore b/loadtest/.gitignore new file mode 100644 index 000000000..7a60b85e1 --- /dev/null +++ b/loadtest/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/loadtest/README.md b/loadtest/README.md new file mode 100644 index 000000000..cf3e4669d --- /dev/null +++ b/loadtest/README.md @@ -0,0 +1,58 @@ +# Load tests + +Manual transport comparison scripts for the Runloop Python SDK. These hit the +real API and are not part of `pytest`; run on demand. + +## End-to-end transport comparison (real API) + +All scripts require `RUNLOOP_API_KEY`. Set `RUNLOOP_BASE_URL` to override the +default endpoint (`https://api.runloop.ai`). + +Every script sends `devboxes.create` requests against a **deliberately +nonexistent blueprint** (`bp_nonexistent_loadtest_00000`), so every request +fails fast server-side (HTTP `400`) and **no devboxes are created** — isolating +client + server _request handling_ from provisioning. + +| Script | Transport under test | +| --- | --- | +| `loadtest.py` | The **SDK** itself (`AsyncRunloop`). `USE_HTTP2=0` → HTTP/1.1; default / `USE_HTTP2=1` → HTTP/2 (shared httpx pool). Always runs against the installed package in this checkout. | +| `h2_test.py` | Raw `httpx` HTTP/2, bypassing the SDK. Configurable connection count. | +| `h2_single_conn.py` | Raw `httpx` HTTP/2 on a single warmed connection (50-request burst). | +| `raw_fetch_test.py` | Raw `httpx` HTTP/1.1 keep-alive baseline. | +| `alpn_check.py` | Confirms the origin negotiates `h2` via TLS ALPN. | + +The raw-transport probes compare httpx HTTP/2 multiplexing against HTTP/1.1 +directly — the same comparison that motivates the SDK's shared HTTP/2 pool. +They're kept so the comparison stays reproducible. + +```sh +# Install the SDK from this checkout (editable): +cd /path/to/api-client-python && uv sync + +# SDK: HTTP/2 (default) vs HTTP/1.1, 2000-request burst +source ~/env && REQUEST_COUNT=2000 uv run python loadtest/loadtest.py # HTTP/2 +source ~/env && REQUEST_COUNT=2000 USE_HTTP2=0 uv run python loadtest/loadtest.py # HTTP/1.1 + +# Raw httpx HTTP/2 vs HTTP/1.1 comparison +source ~/env && uv run python loadtest/h2_test.py +source ~/env && uv run python loadtest/raw_fetch_test.py + +# Single-connection burst +source ~/env && uv run python loadtest/h2_single_conn.py + +# ALPN check +source ~/env && uv run python loadtest/alpn_check.py +``` + +HTTP/1.1 opens a socket per in-flight request; for large bursts raise the +file-descriptor limit (`ulimit -n 65536`) or keep `REQUEST_COUNT` small. + +## Environment variables + +| Variable | Default | Description | +| --- | --- | --- | +| `RUNLOOP_API_KEY` | *(required)* | API key | +| `RUNLOOP_BASE_URL` | `https://api.runloop.ai` | Override API endpoint | +| `REQUEST_COUNT` | `100000` (`loadtest.py`) / `10000` (`h2_test.py`) / `500` (`raw_fetch_test.py`) | Total requests | +| `NUM_CONNECTIONS` | `10` (`h2_test.py`) / `20` (`raw_fetch_test.py`) | Parallel connections | +| `USE_HTTP2` | `1` | `0` to force HTTP/1.1 in `loadtest.py` | diff --git a/loadtest/alpn_check.py b/loadtest/alpn_check.py new file mode 100644 index 000000000..e0ab701cd --- /dev/null +++ b/loadtest/alpn_check.py @@ -0,0 +1,22 @@ +"""Confirm the origin negotiates h2 via TLS ALPN.""" + +from __future__ import annotations + +import os +import ssl +import socket + +BASE_URL = os.environ.get("RUNLOOP_BASE_URL", "https://api.runloop.ai") +url_parts = BASE_URL.split("://", 1) +host = url_parts[1].split("/")[0] if len(url_parts) > 1 else url_parts[0] +port = 443 + +print(f"Checking ALPN for {host}:{port}") + +ctx = ssl.create_default_context() +ctx.set_alpn_protocols(["h2", "http/1.1"]) + +with socket.create_connection((host, port)) as sock: + with ctx.wrap_socket(sock, server_hostname=host) as tls: + print(f"Negotiated protocol: {tls.selected_alpn_protocol()}") + print(f"TLS version: {tls.version()}") diff --git a/loadtest/h2_single_conn.py b/loadtest/h2_single_conn.py new file mode 100644 index 000000000..8df712f19 --- /dev/null +++ b/loadtest/h2_single_conn.py @@ -0,0 +1,59 @@ +"""Raw httpx HTTP/2 single-connection burst: 50 requests on one warmed connection.""" + +from __future__ import annotations + +import os +import time +import asyncio +from typing import cast + +import httpx + +BASE_URL = os.environ.get("RUNLOOP_BASE_URL", "https://api.runloop.ai") +API_KEY = os.environ["RUNLOOP_API_KEY"] + +BODY = { + "blueprint_id": "bp_nonexistent_loadtest_00000", + "name": "loadtest-h2s-0", + "environment_variables": {"TEST_VAR_1": "value_one"}, + "launch_parameters": {"resource_size_request": "SMALL"}, +} + + +async def send_request(client: httpx.AsyncClient) -> dict[str, object]: + start = time.perf_counter() + response = await client.post( + "/v1/devboxes", + json=BODY, + headers={"authorization": f"Bearer {API_KEY}"}, + ) + return {"latency_ms": (time.perf_counter() - start) * 1000, "status": response.status_code} + + +async def main() -> None: + client = httpx.AsyncClient( + base_url=BASE_URL, + http2=True, + limits=httpx.Limits(max_connections=1, max_keepalive_connections=1), + timeout=httpx.Timeout(120.0), + ) + + # Warmup + w = await send_request(client) + print(f"Warmup: status={w['status']}, latency={w['latency_ms']:.0f}ms") + + count = 50 + print(f"\nBursting {count} requests on 1 warmed connection...") + wall_start = time.perf_counter() + results = await asyncio.gather(*(send_request(client) for _ in range(count))) + wall_ms = (time.perf_counter() - wall_start) * 1000 + + await client.aclose() + + lats: list[float] = sorted(cast(float, r["latency_ms"]) for r in results) + print(f"{count} requests in {wall_ms:.0f}ms ({count / (wall_ms / 1000):.1f} req/s)") + print(f"Latency: min={lats[0]:.0f}ms p50={lats[count // 2]:.0f}ms max={lats[-1]:.0f}ms") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/loadtest/h2_test.py b/loadtest/h2_test.py new file mode 100644 index 000000000..5295a2723 --- /dev/null +++ b/loadtest/h2_test.py @@ -0,0 +1,111 @@ +"""Raw httpx HTTP/2 load test: REQUEST_COUNT requests across NUM_CONNECTIONS connections.""" + +from __future__ import annotations + +import os +import math +import time +import asyncio +from typing import cast + +import httpx + +REQUEST_COUNT = int(os.environ.get("REQUEST_COUNT", "10000")) +NUM_CONNECTIONS = int(os.environ.get("NUM_CONNECTIONS", "10")) +BASE_URL = os.environ.get("RUNLOOP_BASE_URL", "https://api.runloop.ai") +API_KEY = os.environ["RUNLOOP_API_KEY"] + +BODY = { + "blueprint_id": "bp_nonexistent_loadtest_00000", + "name": "loadtest-h2-0", + "environment_variables": {"TEST_VAR_1": "value_one", "TEST_VAR_2": "value_two"}, + "metadata": {"test_run": "h2", "index": "0"}, + "launch_parameters": {"resource_size_request": "SMALL", "keep_alive_time_seconds": 300}, +} + + +def percentile(sorted_vals: list[float], p: float) -> float: + idx = math.ceil(p / 100 * len(sorted_vals)) - 1 + return sorted_vals[max(0, idx)] + + +async def send_request(client: httpx.AsyncClient) -> dict[str, object]: + start = time.perf_counter() + response = await client.post( + "/v1/devboxes", + json=BODY, + headers={"authorization": f"Bearer {API_KEY}"}, + ) + return {"latency_ms": (time.perf_counter() - start) * 1000, "status": response.status_code} + + +async def main() -> None: + if NUM_CONNECTIONS < 1: + print(f'NUM_CONNECTIONS must be a positive integer (got "{os.environ.get("NUM_CONNECTIONS")}")') + return + + print(f"HTTP/2 test: {REQUEST_COUNT} requests, {NUM_CONNECTIONS} connections to {BASE_URL}") + + # One client per logical connection — each capped at 1 connection so requests + # are distributed across NUM_CONNECTIONS distinct HTTP/2 sessions. + clients = [ + httpx.AsyncClient( + base_url=BASE_URL, + http2=True, + limits=httpx.Limits(max_connections=1, max_keepalive_connections=1), + timeout=httpx.Timeout(120.0), + ) + for _ in range(NUM_CONNECTIONS) + ] + + print(f"{NUM_CONNECTIONS} connections established\n") + + completed = 0 + + async def progress_printer() -> None: + while True: + await asyncio.sleep(2) + pct = completed / REQUEST_COUNT * 100 + print(f" progress: {completed}/{REQUEST_COUNT} ({pct:.1f}%)") + + async def wrapped(idx: int) -> dict[str, object]: + nonlocal completed + r = await send_request(clients[idx % NUM_CONNECTIONS]) + completed += 1 + return r + + wall_start = time.perf_counter() + progress_task = asyncio.create_task(progress_printer()) + results = await asyncio.gather(*(wrapped(i) for i in range(REQUEST_COUNT))) + progress_task.cancel() + wall_ms = (time.perf_counter() - wall_start) * 1000 + + for c in clients: + await c.aclose() + + latencies: list[float] = sorted(cast(float, r["latency_ms"]) for r in results) + status_counts: dict[int, int] = {} + for r in results: + s = cast(int, r["status"]) + status_counts[s] = status_counts.get(s, 0) + 1 + + print(f"\n=== HTTP/2 Results ===") + print(f"Requests: {REQUEST_COUNT}") + print(f"Connections: {NUM_CONNECTIONS}") + print(f"Wall clock: {wall_ms / 1000:.2f}s") + print(f"Throughput: {REQUEST_COUNT / (wall_ms / 1000):.1f} req/s") + if latencies: + print("\nLatency (ms):") + print(f" min: {latencies[0]:.1f}") + print(f" p50: {percentile(latencies, 50):.1f}") + print(f" p90: {percentile(latencies, 90):.1f}") + print(f" p95: {percentile(latencies, 95):.1f}") + print(f" p99: {percentile(latencies, 99):.1f}") + print(f" max: {latencies[-1]:.1f}") + print("\nStatus codes:") + for s, c_count in sorted(status_counts.items()): + print(f" {s}: {c_count}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/loadtest/loadtest.py b/loadtest/loadtest.py new file mode 100644 index 000000000..9db9cc80e --- /dev/null +++ b/loadtest/loadtest.py @@ -0,0 +1,159 @@ +"""SDK load test: fires devboxes.create bursts via AsyncRunloop. + +Imports the installed package from this checkout so the benchmark always +exercises the exact code here — not a separately published build. + +USE_HTTP2=0 → HTTP/1.1 (httpx with http2=False) +default → HTTP/2 (shared httpx pool with http2=True) +REQUEST_COUNT defaults to 100 000. +""" + +from __future__ import annotations + +import os +import math +import time +import asyncio + +import httpx + +from runloop_api_client import AsyncRunloop + +REQUEST_COUNT = int(os.environ.get("REQUEST_COUNT", "100000")) +RUNLOOP_BASE_URL = os.environ.get("RUNLOOP_BASE_URL") +# HTTP/2 is the SDK default; set USE_HTTP2=0 to benchmark HTTP/1.1. +USE_HTTP2 = os.environ.get("USE_HTTP2", "1") == "1" +PROGRESS_INTERVAL = 2.0 + + +def build_client() -> AsyncRunloop: + kwargs: dict[str, object] = {"max_retries": 0, "timeout": 120.0} + if RUNLOOP_BASE_URL: + kwargs["base_url"] = RUNLOOP_BASE_URL + if not USE_HTTP2: + # A custom http_client disables the shared HTTP/2 pool. + # Set http2: false explicitly — omitting it would select HTTP/2. + kwargs["http_client"] = httpx.AsyncClient( + http2=False, + limits=httpx.Limits(max_connections=None, max_keepalive_connections=100), + timeout=httpx.Timeout(120.0), + ) + return AsyncRunloop(**kwargs) # type: ignore[arg-type] + + +async def send_request(client: AsyncRunloop, index: int, run_id: str) -> dict[str, object]: + start = time.perf_counter() + status: int | None = None + error: str | None = None + try: + await client.devboxes.create( + blueprint_id="bp_nonexistent_loadtest_00000", + name=f"loadtest-{run_id}-{index}", + environment_variables={"TEST_VAR_1": "value_one", "TEST_VAR_2": "value_two"}, + metadata={"test_run": run_id, "index": str(index)}, + launch_parameters={"resource_size_request": "SMALL", "keep_alive_time_seconds": 300}, + ) + status = 200 + except Exception as exc: + status = getattr(exc, "status_code", None) + error = str(exc) or type(exc).__name__ + return { + "index": index, + "latency_ms": (time.perf_counter() - start) * 1000, + "status": status, + "error": error, + } + + +def percentile(sorted_vals: list[float], p: float) -> float: + idx = math.ceil(p / 100 * len(sorted_vals)) - 1 + return sorted_vals[max(0, idx)] + + +def print_metrics(results: list[dict[str, object]], wall_ms: float) -> None: + latencies = sorted(float(r["latency_ms"]) for r in results) # type: ignore[arg-type] + status_counts: dict[str, int] = {} + for r in results: + key = str(r["status"]) if r["status"] is not None else "network_error" + status_counts[key] = status_counts.get(key, 0) + 1 + + print("\n=== Load Test Results ===") + print(f"Requests: {len(results)}") + print(f"Wall clock: {wall_ms / 1000:.2f}s") + print(f"Throughput: {len(results) / (wall_ms / 1000):.1f} req/s") + if latencies: + print("\nLatency (ms):") + print(f" min: {latencies[0]:.1f}") + print(f" p50: {percentile(latencies, 50):.1f}") + print(f" p90: {percentile(latencies, 90):.1f}") + print(f" p95: {percentile(latencies, 95):.1f}") + print(f" p99: {percentile(latencies, 99):.1f}") + print(f" max: {latencies[-1]:.1f}") + print("\nStatus codes:") + for s, c in sorted(status_counts.items()): + print(f" {s}: {c}") + + # Break down the opaque "network_error" bucket by exception message so + # transport-level failures (timeouts, resets, pool exhaustion) are diagnosable. + error_counts: dict[str, int] = {} + for r in results: + if r["status"] is None and r["error"] is not None: + msg = str(r["error"]) + error_counts[msg] = error_counts.get(msg, 0) + 1 + if error_counts: + print("\nErrors (network_error breakdown):") + for msg, c in sorted(error_counts.items(), key=lambda kv: kv[1], reverse=True): + print(f" {c}x {msg}") + + +async def main() -> None: + fd_limit: int | None = None + try: + import resource as _resource + + fd_limit = _resource.getrlimit(_resource.RLIMIT_NOFILE)[0] + except Exception: + pass + + if not USE_HTTP2 and fd_limit is not None and fd_limit < 10000: + print(f"\nWARNING: File descriptor limit is {fd_limit}. For large HTTP/1.1 bursts, run:") + print(" ulimit -n 65536") + print("Or use HTTP/2 multiplexing: USE_HTTP2=1\n") + + client = build_client() + run_id = f"run-{int(time.time())}" + + print(f"Starting load test: {REQUEST_COUNT} concurrent requests") + print(f"Run ID: {run_id}") + print(f"HTTP mode: {'HTTP/2 (shared httpx pool)' if USE_HTTP2 else 'HTTP/1.1 (httpx http2=False)'}") + print(f"Base URL: {RUNLOOP_BASE_URL or '(SDK default)'}") + if fd_limit is not None: + print(f"FD limit: {fd_limit}") + print() + + completed = 0 + + async def progress_printer() -> None: + while True: + await asyncio.sleep(PROGRESS_INTERVAL) + pct = completed / REQUEST_COUNT * 100 + print(f" progress: {completed}/{REQUEST_COUNT} ({pct:.1f}%)") + + async def wrapped(idx: int) -> dict[str, object]: + nonlocal completed + r = await send_request(client, idx, run_id) + completed += 1 + return r + + wall_start = time.perf_counter() + progress_task = asyncio.create_task(progress_printer()) + results = list(await asyncio.gather(*(wrapped(i) for i in range(REQUEST_COUNT)))) + progress_task.cancel() + wall_ms = (time.perf_counter() - wall_start) * 1000 + + await client.close() + print_metrics(results, wall_ms) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/loadtest/raw_fetch_test.py b/loadtest/raw_fetch_test.py new file mode 100644 index 000000000..7400f46b1 --- /dev/null +++ b/loadtest/raw_fetch_test.py @@ -0,0 +1,99 @@ +"""Raw httpx HTTP/1.1 keep-alive baseline.""" + +from __future__ import annotations + +import os +import math +import time +import asyncio +from typing import cast + +import httpx + +REQUEST_COUNT = int(os.environ.get("REQUEST_COUNT", "500")) +NUM_CONNECTIONS = int(os.environ.get("NUM_CONNECTIONS", "20")) +BASE_URL = os.environ.get("RUNLOOP_BASE_URL", "https://api.runloop.ai") +API_KEY = os.environ["RUNLOOP_API_KEY"] + +BODY = { + "blueprint_id": "bp_nonexistent_loadtest_00000", + "name": "loadtest-raw-0", + "environment_variables": {"TEST_VAR_1": "value_one", "TEST_VAR_2": "value_two"}, + "metadata": {"test_run": "raw", "index": "0"}, + "launch_parameters": {"resource_size_request": "SMALL", "keep_alive_time_seconds": 300}, +} + + +def percentile(sorted_vals: list[float], p: float) -> float: + idx = math.ceil(p / 100 * len(sorted_vals)) - 1 + return sorted_vals[max(0, idx)] + + +async def main() -> None: + print(f"HTTP/1.1 test: {REQUEST_COUNT} requests, {NUM_CONNECTIONS} keep-alive connections to {BASE_URL}") + + client = httpx.AsyncClient( + base_url=BASE_URL, + http2=False, + limits=httpx.Limits(max_connections=NUM_CONNECTIONS, max_keepalive_connections=NUM_CONNECTIONS), + timeout=httpx.Timeout(120.0), + ) + + completed = 0 + + async def progress_printer() -> None: + while True: + await asyncio.sleep(2) + pct = completed / REQUEST_COUNT * 100 + print(f" progress: {completed}/{REQUEST_COUNT} ({pct:.1f}%)") + + async def send_one() -> dict[str, object]: + nonlocal completed + start = time.perf_counter() + status: int | None = None + try: + response = await client.post( + "/v1/devboxes", + json=BODY, + headers={"authorization": f"Bearer {API_KEY}"}, + ) + status = response.status_code + except Exception: + pass + finally: + completed += 1 + return {"latency_ms": (time.perf_counter() - start) * 1000, "status": status} + + wall_start = time.perf_counter() + progress_task = asyncio.create_task(progress_printer()) + results = await asyncio.gather(*(send_one() for _ in range(REQUEST_COUNT))) + progress_task.cancel() + wall_ms = (time.perf_counter() - wall_start) * 1000 + + await client.aclose() + + latencies: list[float] = sorted(cast(float, r["latency_ms"]) for r in results) + status_counts: dict[str, int] = {} + for r in results: + key = str(r["status"]) if r["status"] is not None else "network_error" + status_counts[key] = status_counts.get(key, 0) + 1 + + print(f"\n=== HTTP/1.1 Results ===") + print(f"Requests: {REQUEST_COUNT}") + print(f"Wall clock: {wall_ms / 1000:.2f}s") + print(f"Throughput: {REQUEST_COUNT / (wall_ms / 1000):.1f} req/s") + if latencies: + print("\nLatency (ms):") + print(f" min: {latencies[0]:.1f}") + print(f" p50: {percentile(latencies, 50):.1f}") + print(f" p90: {percentile(latencies, 90):.1f}") + print(f" p95: {percentile(latencies, 95):.1f}") + print(f" p99: {percentile(latencies, 99):.1f}") + print(f" max: {latencies[-1]:.1f}") + print("\nStatus codes:") + for s, c_count in sorted(status_counts.items()): + print(f" {s}: {c_count}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 7362e5b07..c160c0fa6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -256,3 +256,4 @@ known-first-party = ["runloop_api_client", "tests"] "scripts/**.py" = ["T201", "T203"] "tests/**.py" = ["T201", "T203"] "examples/**.py" = ["T201", "T203"] +"loadtest/**.py" = ["T201", "T203"]