feat(loadtest): add USE_SYNC to benchmark the blocking client

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). |

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()