From 9c4a4045906eed7e34562075eeede3bd67f36bb1 Mon Sep 17 00:00:00 2001 From: thomasht86 Date: Tue, 28 Jul 2026 13:23:27 +0200 Subject: [PATCH] feat(async): add AsyncClient(max_concurrency=...) AsyncClient dispatches every request with `loop.run_in_executor(None, ...)`, which selects asyncio's *default* executor. That has two consequences: 1. Concurrency is silently capped at `min(32, cpu_count + 4)` -- 16 on a 12-core machine -- no matter how many requests the caller has in flight. Downstream this made async feeding *slower* than a plain thread pool over the sync client. 2. It monopolises a pool the library does not own. `asyncio.to_thread` resolves to that same executor, so httpr requests contend with every `to_thread` call in the host application, and vice versa: a user's slow blocking work stalls HTTP requests. Give each AsyncClient its own ThreadPoolExecutor, sized by `max_concurrency` (default 64). Threads are created on demand and ThreadPoolExecutor retires them via a weakref callback once the client is collected, so an idle client costs nothing and there is nothing new to release -- `close`, `aclose` and `__aexit__` keep their existing behaviour, and a client stays reusable across several `async with` blocks. `max_concurrency=None` restores dispatch on asyncio's default executor. Concurrency only pays off when there is latency to hide. Against a local server with 50ms latency, 64 requests: max_concurrency=None (previous behaviour) 316ms peak in-flight 16 max_concurrency=64 143ms peak in-flight 64 2.2x On localhost the same comparison is ~1.0x, which is why the regression guard asserts on observed peak concurrency rather than on wall-clock. It uses a Barrier, so an undersized pool deadlocks and times out instead of merely being slower, and it fails against the previous `run_in_executor(None)` implementation. Also adds an async concurrency benchmark group. It runs against the async benchmark server rather than the httpbin fixture -- that fixture is a single-threaded WSGI server, so 64 concurrent requests queue on it and the numbers measure the server serialising, with a standard deviation ~200x the signal. The benchmark tracks dispatch overhead; it is not the regression guard, for the localhost reason above. This does not make AsyncClient natively async; it makes the thread-pool ceiling explicit and tunable. Since reqwest is already async internally, native async via pyo3-async-runtimes (plus a multi-thread tokio runtime in place of the current `new_current_thread`) remains the longer-term fix. --- httpr/__init__.py | 45 +++++++++++++-- httpr/httpr.pyi | 14 ++++- tests/benchmark/test_performance.py | 35 +++++++++++- tests/unit/test_asyncclient.py | 86 +++++++++++++++++++++++++++++ 4 files changed, 170 insertions(+), 10 deletions(-) diff --git a/httpr/__init__.py b/httpr/__init__.py index 3c7d9116..d4080577 100644 --- a/httpr/__init__.py +++ b/httpr/__init__.py @@ -30,6 +30,7 @@ import asyncio import sys from collections.abc import AsyncIterator, Generator +from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager, contextmanager from functools import partial from typing import TYPE_CHECKING, TypedDict @@ -42,6 +43,10 @@ from .httpr import CaseInsensitiveHeaderMap, RClient, Response, StreamingResponse +#: Default number of requests an :class:`AsyncClient` keeps in flight. Threads are +#: created lazily, so an idle client costs nothing. +DEFAULT_MAX_CONCURRENCY = 64 + class CaseInsensitiveDict(dict[str, str]): """A dict subclass that provides case-insensitive key access.""" @@ -641,16 +646,46 @@ async def main(): Note: AsyncClient runs synchronous Rust code in a thread executor. It provides concurrency benefits for I/O-bound tasks but is not - native async I/O. + native async I/O. `max_concurrency` sizes that executor and therefore + caps how many requests can be in flight at once. """ - def __init__(self, *args, **kwargs): + def __new__(cls, *args, max_concurrency: int | None = None, **kwargs): + # Client inherits from the Rust-backed RClient, whose __new__ consumes the + # constructor keyword arguments, so max_concurrency has to be stripped here + # as well as in __init__. + return super().__new__(cls, *args, **kwargs) + + def __init__( + self, + *args, + max_concurrency: int | None = DEFAULT_MAX_CONCURRENCY, + **kwargs, + ): """ Initialize an async HTTP client. - Accepts the same parameters as Client. + Accepts the same parameters as Client, plus: + + Args: + max_concurrency: Maximum number of requests in flight at once, i.e. the + size of this client's thread pool. Defaults to 64. Threads are + created lazily, so an idle client costs nothing. Pass ``None`` to + dispatch on asyncio's default executor instead -- note that this + shares a pool with `asyncio.to_thread` and every other + ``run_in_executor(None)`` caller in the application, and that + CPython sizes it at ``min(32, cpu_count + 4)``. """ super().__init__(*args, **kwargs) + self.max_concurrency = max_concurrency + # Threads are created on demand, and ThreadPoolExecutor retires them via a + # weakref callback once this client is collected, so there is nothing to + # release explicitly and `aclose` stays the no-op it has always been. + self._executor = ( + None + if max_concurrency is None + else ThreadPoolExecutor(max_workers=max_concurrency, thread_name_prefix="httpr") + ) async def __aenter__(self) -> AsyncClient: """Enter async context manager.""" @@ -677,9 +712,9 @@ async def aclose(self): return async def _run_sync_asyncio(self, fn, *args, **kwargs): - """Run a synchronous function in an executor.""" + """Run a synchronous function on this client's executor.""" loop = asyncio.get_running_loop() - return await loop.run_in_executor(None, partial(fn, *args, **kwargs)) + return await loop.run_in_executor(self._executor, partial(fn, *args, **kwargs)) async def request( # type: ignore[override] self, diff --git a/httpr/httpr.pyi b/httpr/httpr.pyi index 9dea45ed..2fb16d07 100644 --- a/httpr/httpr.pyi +++ b/httpr/httpr.pyi @@ -542,7 +542,8 @@ class AsyncClient(Client): Note: AsyncClient runs synchronous Rust code in a thread executor. It provides concurrency benefits for I/O-bound tasks but is not - native async I/O. + native async I/O. `max_concurrency` sizes that executor and therefore + caps how many requests can be in flight at once. """ def __init__( self, @@ -563,9 +564,18 @@ class AsyncClient(Client): client_pem_data: bytes | None = None, https_only: bool | None = False, http2_only: bool | None = False, + max_concurrency: int | None = 64, ) -> None: - """Initialize an async HTTP client. Accepts the same parameters as Client.""" + """Initialize an async HTTP client. + + Accepts the same parameters as Client, plus `max_concurrency`: the maximum + number of requests in flight at once, i.e. the size of this client's thread + pool. Threads are created lazily. Pass ``None`` to use asyncio's default + executor instead, which is shared with `asyncio.to_thread` and sized + ``min(32, cpu_count + 4)``. + """ ... + max_concurrency: int | None async def __aenter__(self) -> AsyncClient: ... async def __aexit__(self, *args: Any) -> None: ... async def aclose(self) -> None: diff --git a/tests/benchmark/test_performance.py b/tests/benchmark/test_performance.py index 58ddaad1..b0833a25 100644 --- a/tests/benchmark/test_performance.py +++ b/tests/benchmark/test_performance.py @@ -16,12 +16,14 @@ def bench_server_url(): """URL for the benchmark server (benchmark/server.py). - Set BENCHMARK_SERVER_URL env var to enable CBOR benchmarks. - The server provides /cbor/1, /cbor/10, /cbor/100 endpoints. + Set BENCHMARK_SERVER_URL env var to enable the benchmarks that need it: CBOR + and JSON decoding, and async concurrency. The server provides /cbor/{1,10,100} + and /json/{1,10,100} endpoints, and unlike the httpbin `base_url` fixture it + serves requests concurrently. """ url = os.environ.get("BENCHMARK_SERVER_URL") if not url: - pytest.skip("BENCHMARK_SERVER_URL not set - start benchmark server for CBOR tests") + pytest.skip("BENCHMARK_SERVER_URL not set - start benchmark server for these tests") return url @@ -82,6 +84,33 @@ async def run(): benchmark(lambda: asyncio.run(run())) + @pytest.mark.parametrize("max_concurrency", [8, 32, 64], ids=["8", "32", "64"]) + def test_concurrent_requests(self, benchmark, bench_server_url, max_concurrency): + """Benchmark 64 concurrent requests at a given `max_concurrency`. + + AsyncClient dispatches onto a thread pool, so `max_concurrency` sets the + in-flight ceiling. This tracks per-request overhead under concurrency, + which the single-request benchmarks cannot see. + + Deliberately uses the async benchmark server rather than the `base_url` + httpbin fixture: the latter is a single-threaded WSGI server, so 64 + concurrent requests queue up on it and the numbers measure the *server* + serialising rather than anything about httpr. + + This is overhead tracking, not a guard against a concurrency ceiling -- + on localhost there is barely any latency for concurrency to hide. The + ceiling is asserted directly in tests/unit/test_asyncclient.py:: + test_concurrency_is_not_capped_by_the_default_executor. + """ + n_requests = 64 + + async def run(): + async with httpr.AsyncClient(max_concurrency=max_concurrency) as client: + return await asyncio.gather(*[client.get(f"{bench_server_url}/json/1") for _ in range(n_requests)]) + + benchmark.group = "Async concurrency (64 requests)" + benchmark(lambda: asyncio.run(run())) + class TestResponseSizes: """Benchmark different response payload sizes.""" diff --git a/tests/unit/test_asyncclient.py b/tests/unit/test_asyncclient.py index 52040560..81c7db88 100644 --- a/tests/unit/test_asyncclient.py +++ b/tests/unit/test_asyncclient.py @@ -1,3 +1,7 @@ +import asyncio +import os +import threading + import pytest import httpr @@ -23,3 +27,85 @@ async def test_asyncclient_init(base_url_ssl, ca_bundle): assert json_data["headers"]["Cookie"] == "ccc=ddd; cccc=dddd" assert json_data["headers"]["Authorization"] == "Basic dXNlcjpwYXNzd29yZA==" assert json_data["args"] == {"x": "aaa", "y": "bbb"} + + +def test_default_max_concurrency(): + client = httpr.AsyncClient() + assert client.max_concurrency == httpr.DEFAULT_MAX_CONCURRENCY + assert client._executor._max_workers == httpr.DEFAULT_MAX_CONCURRENCY + + +def test_max_concurrency_sizes_the_pool(): + client = httpr.AsyncClient(max_concurrency=7) + assert client._executor._max_workers == 7 + + +@pytest.mark.asyncio +async def test_requests_run_on_the_clients_own_pool(base_url): + """Requests must not land on asyncio's default executor, which is shared with + asyncio.to_thread and sized min(32, cpu_count + 4).""" + client = httpr.AsyncClient(max_concurrency=4) + thread_name = await client._run_sync_asyncio(lambda: threading.current_thread().name) + assert thread_name.startswith("httpr"), f"ran on {thread_name!r}, not on the client's own pool" + # and a real request still works + response = await client.get(f"{base_url}/anything") + assert response.status_code == 200 + + +@pytest.mark.asyncio +async def test_max_concurrency_none_uses_default_executor(): + client = httpr.AsyncClient(max_concurrency=None) + assert client._executor is None + thread_name = await client._run_sync_asyncio(lambda: threading.current_thread().name) + assert thread_name.startswith("asyncio_") + + +@pytest.mark.asyncio +async def test_concurrency_is_not_capped_by_the_default_executor(): + """Regression guard for the bug this feature fixes. + + Dispatching on asyncio's default executor caps in-flight requests at + ``min(32, cpu_count + 4)``, so `max_concurrency` above that was silently + ignored. Asserts on observed peak concurrency rather than on wall-clock: + against a localhost server there is no latency for concurrency to hide, so a + throughput measurement would *not* catch a regression here. + """ + default_executor_cap = min(32, (os.cpu_count() or 1) + 4) + concurrency = default_executor_cap + 16 + + in_flight = 0 + peak = 0 + lock = threading.Lock() + all_admitted = threading.Barrier(concurrency, timeout=30) + + def occupy(): + nonlocal in_flight, peak + with lock: + in_flight += 1 + peak = max(peak, in_flight) + # Block until every task has been admitted. If the pool were smaller than + # `concurrency` this would time out rather than merely being slow. + all_admitted.wait() + with lock: + in_flight -= 1 + + client = httpr.AsyncClient(max_concurrency=concurrency) + await asyncio.gather(*[client._run_sync_asyncio(occupy) for _ in range(concurrency)]) + + assert peak == concurrency + assert peak > default_executor_cap + + +@pytest.mark.asyncio +async def test_client_stays_reusable_across_contexts(base_url): + """Reuse across several `async with` blocks must keep working -- downstream + users (pyvespa) hold one client and scope it with repeated `async with`.""" + client = httpr.AsyncClient(max_concurrency=2) + + async with client: + assert (await client.get(f"{base_url}/anything")).status_code == 200 + async with client: + assert (await client.get(f"{base_url}/anything")).status_code == 200 + + await client.aclose() + assert (await client.get(f"{base_url}/anything")).status_code == 200