Repository: encode/httpx

Issue:
Timeout from ConnectionPool on number of concurrent requests larger than max_connections
While investigating https://github.com/encode/httpx/issues/551 I noticed that when issuing a high number of concurrent requests to a server an `TimeoutError`.

Running this example adapted from #551:

```python
import asyncio
import os
import random
from typing import NamedTuple, Optional as Opt

import click
import httpx

class Config(NamedTuple):
    requests: Opt[int]
    min: Opt[int]
    max: Opt[int]


async def send_requests(config):
    requests = []
    async with httpx.Client(
        timeout=httpx.TimeoutConfig(timeout=60), base_url="http://localhost:5000/"
    ) as client:
        for _ in range(config.requests):
            requests.append(_send_to_server(client, config))

        results = await asyncio.gather(*requests)

    return results


async def _send_to_server(client, config):
    size = random.randint(config.min, config.max)
    payload = os.urandom(size)
    headers = {"Content-Type": "application/octet-stream", "Content-Length": str(size)}
    response = await client.post("/", headers=headers, data=payload)
    print(response.content)
    return response


@click.command()
@click.option("--min", "min_", default=1, help="Minimum size of payload")
@click.option("--max", "max_", default=(1024 ** 2), help="Maximum size of payload")
@click.option("--requests", default=1000, help="Number of requests")
def run(min_, max_, requests):
    config = Config(requests=requests, min=min_, max=max_)
    asyncio.run(send_requests(config))


if __name__ == "__main__":
    run()
```

Making requests to this simple Quart server:

```python
from quart import Quart, request

app = Quart(__name__)

@app.route("/", methods=["POST"])
async def post():
    data = await request.get_data()
    return {"size": len(data)}

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)
```

And with logging active (raised some lines to DEBUG) a long string of the following line is produced:

`DEBUG [2019-11-27 20:00:31] httpx.dispatch.connection_pool - acquire_connection origin=Origin(scheme='http' host='localhost' port=5000)`

Followed after a few seconds with another string of:

`DEBUG [2019-11-27 20:00:31] httpx.dispatch.connection_pool - new_connection connection=HTTPConnection(origin=Origin(scheme='http' host='localhost' port=5000))`

But shortly after crashing with the following traceback:

```
Traceback (most recent call last):
  File "/Users/yeray/code/personal/_forks/httpx/httpx/concurrency/asyncio.py", line 236, in acquire
    await asyncio.wait_for(self.semaphore.acquire(), timeout)
  File "/Users/yeray/.pyenv/versions/3.7.4/lib/python3.7/asyncio/tasks.py", line 449, in wait_for
    raise futures.TimeoutError()
concurrent.futures._base.TimeoutError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "__dev__/concurrent_writes.py", line 48, in <module>
    run()
  File "/Users/yeray/.pyenv/versions/httpx/lib/python3.7/site-packages/click/core.py", line 764, in __call__
    return self.main(*args, **kwargs)
  File "/Users/yeray/.pyenv/versions/httpx/lib/python3.7/site-packages/click/core.py", line 717, in main
    rv = self.invoke(ctx)
  File "/Users/yeray/.pyenv/versions/httpx/lib/python3.7/site-packages/click/core.py", line 956, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/yeray/.pyenv/versions/httpx/lib/python3.7/site-packages/click/core.py", line 555, in invoke
    return callback(*args, **kwargs)
  File "__dev__/concurrent_writes.py", line 44, in run
    asyncio.run(send_requests(config))
  File "/Users/yeray/.pyenv/versions/3.7.4/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Users/yeray/.pyenv/versions/3.7.4/lib/python3.7/asyncio/base_events.py", line 579, in run_until_complete
    return future.result()
  File "__dev__/concurrent_writes.py", line 24, in send_requests
    results = await asyncio.gather(*requests)
  File "__dev__/concurrent_writes.py", line 33, in _send_to_server
    response = await client.post("/", headers=headers, data=payload)
  File "/Users/yeray/code/personal/_forks/httpx/httpx/client.py", line 718, in post
    trust_env=trust_env,
  File "/Users/yeray/code/personal/_forks/httpx/httpx/client.py", line 246, in request
    trust_env=trust_env,
  File "/Users/yeray/code/personal/_forks/httpx/httpx/client.py", line 354, in send
    allow_redirects=allow_redirects,
  File "/Users/yeray/code/personal/_forks/httpx/httpx/client.py", line 417, in send_handling_redirects
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/yeray/code/personal/_forks/httpx/httpx/client.py", line 546, in send_single_request
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/yeray/code/personal/_forks/httpx/httpx/client.py", line 546, in send_single_request
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/yeray/code/personal/_forks/httpx/httpx/dispatch/connection_pool.py", line 120, in send
    connection = await self.acquire_connection(origin=request.url.origin)
  File "/Users/yeray/code/personal/_forks/httpx/httpx/dispatch/connection_pool.py", line 137, in acquire_connection
    await self.max_connections.acquire()
  File "/Users/yeray/code/personal/_forks/httpx/httpx/concurrency/asyncio.py", line 238, in acquire
    raise PoolTimeout()
httpx.exceptions.PoolTimeout
```

This only happens when the number of requests is much larger than the `max_connections`, in my Mac it starts happening at around 400 requests.

I believe the issue is around the fact that all coroutines will block on the semaphore roughly at the same time with the same timeout, but it takes longer than the timeout for the queue to clear in front of them, causing the exception to be raised.

Discussion:

Patch:
diff --git a/httpx/client.py b/httpx/client.py
index 30e2181..4c594f2 100644
--- a/httpx/client.py
+++ b/httpx/client.py
@@ -535,7 +535,7 @@ class Client:
         Sends a single request, without handling any redirections.
         """
 
-        dispatcher = self._dispatcher_for_request(request, self.proxies)
+        dispatcher = self.dispatcher_for_url(request.url)
 
         try:
             with ElapsedTimer() as timer:
@@ -560,12 +560,12 @@ class Client:
 
         return response
 
-    def _dispatcher_for_request(
-        self, request: Request, proxies: typing.Dict[str, Dispatcher]
-    ) -> Dispatcher:
-        """Gets the Dispatcher instance that should be used for a given Request"""
-        if proxies:
-            url = request.url
+    def dispatcher_for_url(self, url: URL) -> Dispatcher:
+        """
+        Returns the Dispatcher instance that should be used for a given URL.
+        This will either be the standard connection pool, or a proxy.
+        """
+        if self.proxies:
             is_default_port = (url.scheme == "http" and url.port == 80) or (
                 url.scheme == "https" and url.port == 443
             )
@@ -579,8 +579,8 @@ class Client:
                 "all",
             )
             for proxy_key in proxy_keys:
-                if proxy_key and proxy_key in proxies:
-                    dispatcher = proxies[proxy_key]
+                if proxy_key and proxy_key in self.proxies:
+                    dispatcher = self.proxies[proxy_key]
                     return dispatcher
 
         return self.dispatch
diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
index d3febbb..c3fcae0 100644
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -11,7 +11,6 @@ from .base import (
     BaseBackgroundManager,
     BaseEvent,
     BasePoolSemaphore,
-    BaseQueue,
     BaseSocketStream,
     ConcurrencyBackend,
     TimeoutFlag,
@@ -227,11 +226,10 @@ class PoolSemaphore(BasePoolSemaphore):
                 self._semaphore = asyncio.BoundedSemaphore(value=max_connections)
         return self._semaphore
 
-    async def acquire(self) -> None:
+    async def acquire(self, timeout: float = None) -> None:
         if self.semaphore is None:
             return
 
-        timeout = self.pool_limits.pool_timeout
         try:
             await asyncio.wait_for(self.semaphore.acquire(), timeout)
         except asyncio.TimeoutError:
@@ -325,9 +323,6 @@ class AsyncioBackend(ConcurrencyBackend):
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         return PoolSemaphore(limits)
 
-    def create_queue(self, max_size: int) -> BaseQueue:
-        return typing.cast(BaseQueue, asyncio.Queue(maxsize=max_size))
-
     def create_event(self) -> BaseEvent:
         return typing.cast(BaseEvent, asyncio.Event())
 
diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py
index 2109c21..6bbeb07 100644
--- a/httpx/concurrency/base.py
+++ b/httpx/concurrency/base.py
@@ -70,18 +70,6 @@ class BaseSocketStream:
         raise NotImplementedError()  # pragma: no cover
 
 
-class BaseQueue:
-    """
-    A FIFO queue. Abstracts away any asyncio-specific interfaces.
-    """
-
-    async def get(self) -> typing.Any:
-        raise NotImplementedError()  # pragma: no cover
-
-    async def put(self, value: typing.Any) -> None:
-        raise NotImplementedError()  # pragma: no cover
-
-
 class BaseEvent:
     """
     An event object. Abstracts away any asyncio-specific interfaces.
@@ -107,7 +95,7 @@ class BasePoolSemaphore:
     Abstracts away any asyncio-specific interfaces.
     """
 
-    async def acquire(self) -> None:
+    async def acquire(self, timeout: float = None) -> None:
         raise NotImplementedError()  # pragma: no cover
 
     def release(self) -> None:
@@ -141,37 +129,11 @@ class ConcurrencyBackend:
     ) -> typing.Any:
         raise NotImplementedError()  # pragma: no cover
 
-    async def iterate_in_threadpool(self, iterator):  # type: ignore
-        class IterationComplete(Exception):
-            pass
-
-        def next_wrapper(iterator):  # type: ignore
-            try:
-                return next(iterator)
-            except StopIteration:
-                raise IterationComplete()
-
-        while True:
-            try:
-                yield await self.run_in_threadpool(next_wrapper, iterator)
-            except IterationComplete:
-                break
-
     def run(
         self, coroutine: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:
         raise NotImplementedError()  # pragma: no cover
 
-    def iterate(self, async_iterator):  # type: ignore
-        while True:
-            try:
-                yield self.run(async_iterator.__anext__)
-            except StopAsyncIteration:
-                break
-
-    def create_queue(self, max_size: int) -> BaseQueue:
-        raise NotImplementedError()  # pragma: no cover
-
     def create_event(self) -> BaseEvent:
         raise NotImplementedError()  # pragma: no cover
 
diff --git a/httpx/concurrency/trio.py b/httpx/concurrency/trio.py
index f2ab919..0169f54 100644
--- a/httpx/concurrency/trio.py
+++ b/httpx/concurrency/trio.py
@@ -1,5 +1,4 @@
 import functools
-import math
 import ssl
 import typing
 from types import TracebackType
@@ -12,7 +11,6 @@ from .base import (
     BaseBackgroundManager,
     BaseEvent,
     BasePoolSemaphore,
-    BaseQueue,
     BaseSocketStream,
     ConcurrencyBackend,
     TimeoutFlag,
@@ -153,11 +151,11 @@ class PoolSemaphore(BasePoolSemaphore):
                 )
         return self._semaphore
 
-    async def acquire(self) -> None:
+    async def acquire(self, timeout: float = None) -> None:
         if self.semaphore is None:
             return
 
-        timeout = _or_inf(self.pool_limits.pool_timeout)
+        timeout = _or_inf(timeout)
 
         with trio.move_on_after(timeout):
             await self.semaphore.acquire()
@@ -230,9 +228,6 @@ class TrioBackend(ConcurrencyBackend):
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         return PoolSemaphore(limits)
 
-    def create_queue(self, max_size: int) -> BaseQueue:
-        return Queue(max_size=max_size)
-
     def create_event(self) -> BaseEvent:
         return Event()
 
@@ -242,17 +237,6 @@ class TrioBackend(ConcurrencyBackend):
         return BackgroundManager(coroutine, *args)
 
 
-class Queue(BaseQueue):
-    def __init__(self, max_size: int) -> None:
-        self.send_channel, self.receive_channel = trio.open_memory_channel(math.inf)
-
-    async def get(self) -> typing.Any:
-        return await self.receive_channel.receive()
-
-    async def put(self, value: typing.Any) -> None:
-        await self.send_channel.send(value)
-
-
 class Event(BaseEvent):
     def __init__(self) -> None:
         self._event = trio.Event()
diff --git a/httpx/config.py b/httpx/config.py
index 910f91b..a32ae84 100644
--- a/httpx/config.py
+++ b/httpx/config.py
@@ -10,7 +10,9 @@ from .utils import get_ca_bundle_from_env, get_logger
 
 CertTypes = typing.Union[str, typing.Tuple[str, str], typing.Tuple[str, str, str]]
 VerifyTypes = typing.Union[str, bool, ssl.SSLContext]
-TimeoutTypes = typing.Union[float, typing.Tuple[float, float, float], "TimeoutConfig"]
+TimeoutTypes = typing.Union[
+    float, typing.Tuple[float, float, float, float], "TimeoutConfig"
+]
 HTTPVersionTypes = typing.Union[
     str, typing.List[str], typing.Tuple[str], "HTTPVersionConfig"
 ]
@@ -227,28 +229,34 @@ class TimeoutConfig:
         connect_timeout: float = None,
         read_timeout: float = None,
         write_timeout: float = None,
+        pool_timeout: float = None,
     ):
         if timeout is None:
             self.connect_timeout = connect_timeout
             self.read_timeout = read_timeout
             self.write_timeout = write_timeout
+            self.pool_timeout = pool_timeout
         else:
             # Specified as a single timeout value
             assert connect_timeout is None
             assert read_timeout is None
             assert write_timeout is None
+            assert pool_timeout is None
             if isinstance(timeout, TimeoutConfig):
                 self.connect_timeout = timeout.connect_timeout
                 self.read_timeout = timeout.read_timeout
                 self.write_timeout = timeout.write_timeout
+                self.pool_timeout = timeout.pool_timeout
             elif isinstance(timeout, tuple):
                 self.connect_timeout = timeout[0]
                 self.read_timeout = timeout[1]
-                self.write_timeout = timeout[2]
+                self.write_timeout = None if len(timeout) < 3 else timeout[2]
+                self.pool_timeout = None if len(timeout) < 4 else timeout[3]
             else:
                 self.connect_timeout = timeout
                 self.read_timeout = timeout
                 self.write_timeout = timeout
+                self.pool_timeout = timeout
 
     def __eq__(self, other: typing.Any) -> bool:
         return (
@@ -256,15 +264,27 @@ class TimeoutConfig:
             and self.connect_timeout == other.connect_timeout
             and self.read_timeout == other.read_timeout
             and self.write_timeout == other.write_timeout
+            and self.pool_timeout == other.pool_timeout
         )
 
     def __repr__(self) -> str:
         class_name = self.__class__.__name__
-        if len({self.connect_timeout, self.read_timeout, self.write_timeout}) == 1:
+        if (
+            len(
+                {
+                    self.connect_timeout,
+                    self.read_timeout,
+                    self.write_timeout,
+                    self.pool_timeout,
+                }
+            )
+            == 1
+        ):
             return f"{class_name}(timeout={self.connect_timeout})"
         return (
             f"{class_name}(connect_timeout={self.connect_timeout}, "
-            f"read_timeout={self.read_timeout}, write_timeout={self.write_timeout})"
+            f"read_timeout={self.read_timeout}, write_timeout={self.write_timeout}, "
+            f"pool_timeout={self.pool_timeout})"
         )
 
 
@@ -320,34 +340,27 @@ class PoolLimits:
     """
 
     def __init__(
-        self,
-        *,
-        soft_limit: int = None,
-        hard_limit: int = None,
-        pool_timeout: float = None,
+        self, *, soft_limit: int = None, hard_limit: int = None,
     ):
         self.soft_limit = soft_limit
         self.hard_limit = hard_limit
-        self.pool_timeout = pool_timeout
 
     def __eq__(self, other: typing.Any) -> bool:
         return (
             isinstance(other, self.__class__)
             and self.soft_limit == other.soft_limit
             and self.hard_limit == other.hard_limit
-            and self.pool_timeout == other.pool_timeout
         )
 
     def __repr__(self) -> str:
         class_name = self.__class__.__name__
         return (
-            f"{class_name}(soft_limit={self.soft_limit}, "
-            f"hard_limit={self.hard_limit}, pool_timeout={self.pool_timeout})"
+            f"{class_name}(soft_limit={self.soft_limit}, hard_limit={self.hard_limit})"
         )
 
 
 DEFAULT_SSL_CONFIG = SSLConfig(cert=None, verify=True)
 DEFAULT_TIMEOUT_CONFIG = TimeoutConfig(timeout=5.0)
-DEFAULT_POOL_LIMITS = PoolLimits(soft_limit=10, hard_limit=100, pool_timeout=5.0)
+DEFAULT_POOL_LIMITS = PoolLimits(soft_limit=10, hard_limit=100)
 DEFAULT_CA_BUNDLE_PATH = Path(certifi.where())
 DEFAULT_MAX_REDIRECTS = 20
diff --git a/httpx/dispatch/connection_pool.py b/httpx/dispatch/connection_pool.py
index d64a193..3217330 100644
--- a/httpx/dispatch/connection_pool.py
+++ b/httpx/dispatch/connection_pool.py
@@ -8,6 +8,7 @@ from ..config import (
     CertTypes,
     HTTPVersionTypes,
     PoolLimits,
+    TimeoutConfig,
     TimeoutTypes,
     VerifyTypes,
 )
@@ -93,7 +94,7 @@ class ConnectionPool(Dispatcher):
     ):
         self.verify = verify
         self.cert = cert
-        self.timeout = timeout
+        self.timeout = TimeoutConfig(timeout)
         self.pool_limits = pool_limits
         self.http_versions = http_versions
         self.is_closed = False
@@ -117,7 +118,9 @@ class ConnectionPool(Dispatcher):
         cert: CertTypes = None,
         timeout: TimeoutTypes = None,
     ) -> Response:
-        connection = await self.acquire_connection(origin=request.url.origin)
+        connection = await self.acquire_connection(
+            origin=request.url.origin, timeout=timeout
+        )
         try:
             response = await connection.send(
                 request, verify=verify, cert=cert, timeout=timeout
@@ -129,12 +132,19 @@ class ConnectionPool(Dispatcher):
 
         return response
 
-    async def acquire_connection(self, origin: Origin) -> HTTPConnection:
+    async def acquire_connection(
+        self, origin: Origin, timeout: TimeoutTypes = None
+    ) -> HTTPConnection:
         logger.trace(f"acquire_connection origin={origin!r}")
         connection = self.pop_connection(origin)
 
         if connection is None:
-            await self.max_connections.acquire()
+            if timeout is None:
+                pool_timeout = self.timeout.pool_timeout
+            else:
+                pool_timeout = TimeoutConfig(timeout).pool_timeout
+
+            await self.max_connections.acquire(timeout=pool_timeout)
             connection = HTTPConnection(
                 origin,
                 verify=self.verify,
diff --git a/httpx/dispatch/proxy_http.py b/httpx/dispatch/proxy_http.py
index 5451634..13eb423 100644
--- a/httpx/dispatch/proxy_http.py
+++ b/httpx/dispatch/proxy_http.py
@@ -81,12 +81,14 @@ class HTTPProxy(ConnectionPool):
         token = b64encode(b":".join(userpass)).decode().strip()
         return f"Basic {token}"
 
-    async def acquire_connection(self, origin: Origin) -> HTTPConnection:
+    async def acquire_connection(
+        self, origin: Origin, timeout: TimeoutTypes = None
+    ) -> HTTPConnection:
         if self.should_forward_origin(origin):
             logger.trace(
                 f"forward_connection proxy_url={self.proxy_url!r} origin={origin!r}"
             )
-            return await super().acquire_connection(self.proxy_url.origin)
+            return await super().acquire_connection(self.proxy_url.origin, timeout)
         else:
             logger.trace(
                 f"tunnel_connection proxy_url={self.proxy_url!r} origin={origin!r}"


Test patch:
diff --git a/tests/client/test_proxies.py b/tests/client/test_proxies.py
index 1c02ce8..fc1e57a 100644
--- a/tests/client/test_proxies.py
+++ b/tests/client/test_proxies.py
@@ -102,8 +102,7 @@ PROXY_URL = "http://[::1]"
 )
 def test_dispatcher_for_request(url, proxies, expected):
     client = httpx.Client(proxies=proxies)
-    request = httpx.Request("GET", url)
-    dispatcher = client._dispatcher_for_request(request, client.proxies)
+    dispatcher = client.dispatcher_for_url(httpx.URL(url))
 
     if expected is None:
         assert isinstance(dispatcher, httpx.ConnectionPool)
diff --git a/tests/test_config.py b/tests/test_config.py
index 498116c..d8d429d 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -160,9 +160,7 @@ def test_empty_http_version():
 
 def test_limits_repr():
     limits = httpx.PoolLimits(hard_limit=100)
-    assert (
-        repr(limits) == "PoolLimits(soft_limit=None, hard_limit=100, pool_timeout=None)"
-    )
+    assert repr(limits) == "PoolLimits(soft_limit=None, hard_limit=100)"
 
 
 def test_ssl_eq():
@@ -185,6 +183,7 @@ def test_timeout_from_nothing():
     assert timeout.connect_timeout is None
     assert timeout.read_timeout is None
     assert timeout.write_timeout is None
+    assert timeout.pool_timeout is None
 
 
 def test_timeout_from_none():
@@ -198,7 +197,7 @@ def test_timeout_from_one_none_value():
 
 
 def test_timeout_from_tuple():
-    timeout = httpx.TimeoutConfig(timeout=(5.0, 5.0, 5.0))
+    timeout = httpx.TimeoutConfig(timeout=(5.0, 5.0, 5.0, 5.0))
     assert timeout == httpx.TimeoutConfig(timeout=5.0)
 
 
@@ -212,9 +211,9 @@ def test_timeout_repr():
     assert repr(timeout) == "TimeoutConfig(timeout=5.0)"
 
     timeout = httpx.TimeoutConfig(read_timeout=5.0)
-    assert (
-        repr(timeout)
-        == "TimeoutConfig(connect_timeout=None, read_timeout=5.0, write_timeout=None)"
+    assert repr(timeout) == (
+        "TimeoutConfig(connect_timeout=None, read_timeout=5.0, "
+        "write_timeout=None, pool_timeout=None)"
     )
 
 
diff --git a/tests/test_timeouts.py b/tests/test_timeouts.py
index bd6bf8c..89cdc1e 100644
--- a/tests/test_timeouts.py
+++ b/tests/test_timeouts.py
@@ -38,9 +38,12 @@ async def test_connect_timeout(server, backend):
 
 
 async def test_pool_timeout(server, backend):
-    pool_limits = PoolLimits(hard_limit=1, pool_timeout=1e-4)
+    pool_limits = PoolLimits(hard_limit=1)
+    timeout = TimeoutConfig(pool_timeout=1e-4)
 
-    async with Client(pool_limits=pool_limits, backend=backend) as client:
+    async with Client(
+        pool_limits=pool_limits, timeout=timeout, backend=backend
+    ) as client:
         response = await client.get(server.url, stream=True)
 
         with pytest.raises(PoolTimeout):