feat(async): add AsyncClient(max_concurrency=...)

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,

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:

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

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