add sync benchmark

Author: MarkusSintonen

requirements.txt

@@ -25,5 +25,6 @@ werkzeug<2.1  # See: https://github.com/psf/httpbin/issues/35
 # Benchmarking and profiling
 uvicorn==0.30.1
 aiohttp==3.9.5
+requests==2.32.3
 matplotlib==3.7.5
 pyinstrument==4.6.2

scripts/benchmark

@@ -1,5 +1,7 @@
 #!/bin/sh -e
 
+# Usage: scripts/benchmark async|sync
+
 export PREFIX=""
 if [ -d 'venv' ] ; then
     export PREFIX="venv/bin/"
@@ -10,6 +12,6 @@ set -x
 ${PREFIX}python tests/benchmark/server.py &
 SERVER_PID=$!
 EXIT_CODE=0
-${PREFIX}python tests/benchmark/client.py || EXIT_CODE=$?
+${PREFIX}python tests/benchmark/client.py "$@" || EXIT_CODE=$?
 kill $SERVER_PID
 exit $EXIT_CODE

tests/benchmark/client.py

@@ -1,13 +1,17 @@
 import asyncio
 import os
+import sys
 import time
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
-from typing import Any, Coroutine, Iterator, List
+from typing import Any, Callable, Coroutine, Iterator, List
 
 import aiohttp
 import matplotlib.pyplot as plt  # type: ignore[import-untyped]
 import pyinstrument
+import requests  # type: ignore[import-untyped]
 from matplotlib.axes import Axes  # type: ignore[import-untyped]
+from requests.adapters import HTTPAdapter  # type: ignore[import-untyped]
 
 import httpcore
 
@@ -35,19 +39,18 @@ def profile():
     profiler.open_in_browser()
 
 
-async def gather_limited_concurrency(
-    coros: Iterator[Coroutine[Any, Any, Any]], concurrency: int = CONCURRENCY
-) -> None:
-    sem = asyncio.Semaphore(concurrency)
-
-    async def coro_with_sem(coro: Coroutine[Any, Any, Any]) -> None:
-        async with sem:
-            await coro
+async def run_async_requests(axis: Axes) -> None:
+    async def gather_limited_concurrency(
+        coros: Iterator[Coroutine[Any, Any, Any]], concurrency: int = CONCURRENCY
+    ) -> None:
+        sem = asyncio.Semaphore(concurrency)
 
-    await asyncio.gather(*(coro_with_sem(c) for c in coros))
+        async def coro_with_sem(coro: Coroutine[Any, Any, Any]) -> None:
+            async with sem:
+                await coro
 
+        await asyncio.gather(*(coro_with_sem(c) for c in coros))
 
-async def run_requests(axis: Axes) -> None:
     async def httpcore_get(
         pool: httpcore.AsyncConnectionPool, timings: List[int]
     ) -> None:
@@ -99,9 +102,85 @@ async def aiohttp_get(session: aiohttp.ClientSession, timings: List[int]) -> Non
         )
 
 
+def run_sync_requests(axis: Axes) -> None:
+    def run_in_executor(
+        fns: Iterator[Callable[[], None]], executor: ThreadPoolExecutor
+    ) -> None:
+        futures = [executor.submit(fn) for fn in fns]
+        for future in futures:
+            future.result()
+
+    def httpcore_get(pool: httpcore.ConnectionPool, timings: List[int]) -> None:
+        start = time.monotonic()
+        res = pool.request("GET", URL)
+        assert len(res.read()) == 2000
+        assert res.status == 200, f"status_code={res.status}"
+        timings.append(duration(start))
+
+    def requests_get(session: requests.Session, timings: List[int]) -> None:
+        start = time.monotonic()
+        res = session.get(URL)
+        assert len(res.text) == 2000
+        assert res.status_code == 200, f"status={res.status_code}"
+        timings.append(duration(start))
+
+    with httpcore.ConnectionPool(max_connections=POOL_LIMIT) as pool:
+        # warmup
+        with ThreadPoolExecutor(max_workers=CONCURRENCY * 2) as exec:
+            run_in_executor(
+                (lambda: httpcore_get(pool, []) for _ in range(REQUESTS)),
+                exec,
+            )
+
+        timings: List[int] = []
+        exec = ThreadPoolExecutor(max_workers=CONCURRENCY)
+        start = time.monotonic()
+        with profile():
+            for _ in range(REPEATS):
+                run_in_executor(
+                    (lambda: httpcore_get(pool, timings) for _ in range(REQUESTS)), exec
+                )
+        exec.shutdown(wait=True)
+        axis.plot(
+            [*range(len(timings))], timings, label=f"httpcore (tot={duration(start)}ms)"
+        )
+
+    with requests.Session() as session:
+        session.mount(
+            "http://", HTTPAdapter(pool_connections=POOL_LIMIT, pool_maxsize=POOL_LIMIT)
+        )
+        # warmup
+        with ThreadPoolExecutor(max_workers=CONCURRENCY * 2) as exec:
+            run_in_executor(
+                (lambda: requests_get(session, []) for _ in range(REQUESTS)),
+                exec,
+            )
+
+        timings = []
+        exec = ThreadPoolExecutor(max_workers=CONCURRENCY)
+        start = time.monotonic()
+        for _ in range(REPEATS):
+            run_in_executor(
+                (lambda: requests_get(session, timings) for _ in range(REQUESTS)),
+                exec,
+            )
+        exec.shutdown(wait=True)
+        axis.plot(
+            [*range(len(timings))], timings, label=f"requests (tot={duration(start)}ms)"
+        )
+
+
 def main() -> None:
+    mode = sys.argv[1] if len(sys.argv) == 2 else None
+    assert mode in ("async", "sync"), "Usage: python client.py <async|sync>"
+
     fig, ax = plt.subplots()
-    asyncio.run(run_requests(ax))
+
+    if mode == "async":
+        asyncio.run(run_async_requests(ax))
+    else:
+        run_sync_requests(ax)
+
     plt.legend(loc="upper left")
     ax.set_xlabel("# request")
     ax.set_ylabel("[ms]")