Timeout from ConnectionPool on number of concurrent requests larger than max_connections

[yeraydiazdiaz]

While investigating #551 I noticed that when issuing a high number of concurrent requests to a server an TimeoutError.

Running this example adapted from #551:

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:

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.

[tomchristie]

Right, yup. This looks like a UX issue to me.

We've got pool_limits configuration which is seperate to the timeout configuration.
Counter-intuitively, the pool_limits includes a pool_timeout, which the default config sets to 5.

You'd be getting the behavior you're expecting if you used...

pool_limits = PoolLimits(soft_limit=10, hard_limit=100, timeout=60)
async with httpx.Client(
    timeout=60, pool_limits=pool_limits, base_url="http://localhost:5000/"
) as client:

But that's rather gross.

We should look at switching this behavior, so that all timeouts are on the the Timeout config, and so just setting timeout=60 on it's own does exactly what you'd expect.