diff --git a/loadtest/README.md b/loadtest/README.md index cf3e4669d..cdad9be95 100644 --- a/loadtest/README.md +++ b/loadtest/README.md @@ -15,7 +15,7 @@ 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. | +| `loadtest.py` | The **SDK** itself. `USE_HTTP2=0` → HTTP/1.1; default / `USE_HTTP2=1` → HTTP/2 (shared httpx pool). `USE_SYNC=0` (default) → async `AsyncRunloop` via asyncio; `USE_SYNC=1` → blocking `Runloop` via a thread 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. | @@ -33,6 +33,10 @@ cd /path/to/api-client-python && uv sync 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 +# SDK: async (default) vs sync client +source ~/env && REQUEST_COUNT=2000 USE_SYNC=1 uv run python loadtest/loadtest.py # sync, HTTP/2 +source ~/env && REQUEST_COUNT=2000 USE_SYNC=1 CONCURRENCY=500 uv run python loadtest/loadtest.py # cap the thread pool + # 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 @@ -56,3 +60,5 @@ file-descriptor limit (`ulimit -n 65536`) or keep `REQUEST_COUNT` small. | `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` | +| `USE_SYNC` | `0` | `1` to benchmark the blocking `Runloop` client (thread pool) instead of async `AsyncRunloop` | +| `CONCURRENCY` | `500` | Worker-thread cap for the sync path. Effective workers = `min(REQUEST_COUNT, CONCURRENCY)`; above the cap, requests queue through the pool. The sync client is GIL-bound, so raising this past a few hundred adds contention without improving throughput (a 500-worker pool beat 5000 at 20k requests: 260 vs 199 req/s, ~10x lower p50 latency). | diff --git a/loadtest/loadtest.py b/loadtest/loadtest.py index 9db9cc80e..fffb7feb3 100644 --- a/loadtest/loadtest.py +++ b/loadtest/loadtest.py @@ -1,11 +1,36 @@ -"""SDK load test: fires devboxes.create bursts via AsyncRunloop. +"""SDK load test: fires devboxes.create bursts via the Runloop SDK. 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) + +USE_SYNC=1 → blocking sync `Runloop` client driven by a thread pool +default → async `AsyncRunloop` client driven by asyncio + REQUEST_COUNT defaults to 100 000. + +Sync concurrency model +---------------------- +The sync `Runloop` client is blocking, so a concurrent burst is driven by a +`ThreadPoolExecutor` rather than `asyncio.gather`. One OS thread per request is +infeasible at high request counts (memory + scheduler overhead, and on most +hosts the open-file-descriptor limit caps real socket concurrency well below +that anyway), so the worker pool is capped: + + workers = min(REQUEST_COUNT, CONCURRENCY) # CONCURRENCY default 500 + +At request counts above the cap, requests queue through the pool — so for the +sync path "concurrency" means the worker count, not REQUEST_COUNT. + +The default cap is 500. The blocking client is GIL-bound, so adding threads +past a few hundred does not raise throughput and actively hurts: in testing +against the dev cluster (20 000 HTTP/1.1 requests), a 500-worker pool beat a +5000-worker pool on throughput (260 vs 199 req/s) with ~10x lower p50 latency +(1.6s vs 16.9s), because 5000 threads add GIL contention and context-switching +with no upside. Raise `CONCURRENCY` only if you have evidence it helps on your +host; lower it if thread/FD pressure dominates. """ from __future__ import annotations @@ -14,19 +39,27 @@ import math import time import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed import httpx -from runloop_api_client import AsyncRunloop +from runloop_api_client import Runloop, 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" +# Async (asyncio) by default; set USE_SYNC=1 to benchmark the blocking client. +USE_SYNC = os.environ.get("USE_SYNC", "0") == "1" +# Worker-thread cap for the sync path (see module docstring). Default 500: +# the blocking client is GIL-bound, so more threads add contention without +# raising throughput (500 beat 5000 in testing — see docstring). +SYNC_WORKER_CAP = int(os.environ.get("CONCURRENCY", "500")) PROGRESS_INTERVAL = 2.0 -def build_client() -> AsyncRunloop: +def build_async_client() -> AsyncRunloop: kwargs: dict[str, object] = {"max_retries": 0, "timeout": 120.0} if RUNLOOP_BASE_URL: kwargs["base_url"] = RUNLOOP_BASE_URL @@ -41,17 +74,57 @@ def build_client() -> AsyncRunloop: return AsyncRunloop(**kwargs) # type: ignore[arg-type] -async def send_request(client: AsyncRunloop, index: int, run_id: str) -> dict[str, object]: +def build_sync_client() -> Runloop: + kwargs: dict[str, object] = {"max_retries": 0, "timeout": 120.0} + if RUNLOOP_BASE_URL: + kwargs["base_url"] = RUNLOOP_BASE_URL + if not USE_HTTP2: + kwargs["http_client"] = httpx.Client( + http2=False, + limits=httpx.Limits(max_connections=None, max_keepalive_connections=100), + timeout=httpx.Timeout(120.0), + ) + return Runloop(**kwargs) # type: ignore[arg-type] + + +_CREATE_KWARGS = { + "blueprint_id": "bp_nonexistent_loadtest_00000", + "environment_variables": {"TEST_VAR_1": "value_one", "TEST_VAR_2": "value_two"}, + "launch_parameters": {"resource_size_request": "SMALL", "keep_alive_time_seconds": 300}, +} + + +async def send_request_async(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}, + **_CREATE_KWARGS, # type: ignore[arg-type] + ) + 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 send_request_sync(client: Runloop, index: int, run_id: str) -> dict[str, object]: + start = time.perf_counter() + status: int | None = None + error: str | None = None + try: + client.devboxes.create( + name=f"loadtest-{run_id}-{index}", + metadata={"test_run": run_id, "index": str(index)}, + **_CREATE_KWARGS, # type: ignore[arg-type] ) status = 200 except Exception as exc: @@ -106,31 +179,34 @@ def print_metrics(results: list[dict[str, object]], wall_ms: float) -> None: print(f" {c}x {msg}") -async def main() -> None: - fd_limit: int | None = None +def _fd_limit() -> int | None: try: import resource as _resource - fd_limit = _resource.getrlimit(_resource.RLIMIT_NOFILE)[0] + return _resource.getrlimit(_resource.RLIMIT_NOFILE)[0] except Exception: - pass + return None - 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") +def _print_header(run_id: str, fd_limit: int | None, workers: int | None) -> None: + print(f"Starting load test: {REQUEST_COUNT} requests") print(f"Run ID: {run_id}") + mode = "sync (ThreadPoolExecutor)" if USE_SYNC else "async (asyncio)" + print(f"Concurrency: {mode}") + if USE_SYNC: + print(f"Workers: {workers} threads") 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() + +async def run_async() -> None: + client = build_async_client() + run_id = f"run-{int(time.time())}" + _print_header(run_id, _fd_limit(), workers=None) + completed = 0 async def progress_printer() -> None: @@ -141,7 +217,7 @@ async def progress_printer() -> None: async def wrapped(idx: int) -> dict[str, object]: nonlocal completed - r = await send_request(client, idx, run_id) + r = await send_request_async(client, idx, run_id) completed += 1 return r @@ -155,5 +231,49 @@ async def wrapped(idx: int) -> dict[str, object]: print_metrics(results, wall_ms) +def run_sync() -> None: + workers = max(1, min(REQUEST_COUNT, SYNC_WORKER_CAP)) + client = build_sync_client() + run_id = f"run-{int(time.time())}" + _print_header(run_id, _fd_limit(), workers=workers) + + completed = 0 + stop = threading.Event() + + def progress_printer() -> None: + while not stop.wait(PROGRESS_INTERVAL): + pct = completed / REQUEST_COUNT * 100 + print(f" progress: {completed}/{REQUEST_COUNT} ({pct:.1f}%)") + + progress_thread = threading.Thread(target=progress_printer, daemon=True) + progress_thread.start() + + results: list[dict[str, object]] = [] + wall_start = time.perf_counter() + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = [pool.submit(send_request_sync, client, i, run_id) for i in range(REQUEST_COUNT)] + for fut in as_completed(futures): + results.append(fut.result()) + completed += 1 + wall_ms = (time.perf_counter() - wall_start) * 1000 + + stop.set() + client.close() + print_metrics(results, wall_ms) + + +def main() -> None: + fd_limit = _fd_limit() + 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") + + if USE_SYNC: + run_sync() + else: + asyncio.run(run_async()) + + if __name__ == "__main__": - asyncio.run(main()) + main()