Handle asynchronous cancellations for HTTPConnection instances

[TheTechromancer]

• Initially raised as discussion #783

The gist of the race condition is that when a web request is cancelled within a specific time window, it will degrade the state of the connection pool, leaving it unable to issue further requests. It also seems to create a deadlock.

It can be reproduced with the following code:

import trio
import httpcore


TIMEOUT = {"read": 5, "write": 5, "connect": 5, "pool": 5}


async def send_request(client, seen_response):
 response = await client.request("GET", "http://127.0.0.1:8000", extensions={"timeout": TIMEOUT})

 # Print the first response that we see and set the "seen_response" flag.
 if not seen_response.is_set():
 print(response)
 seen_response.set()


async def main():
 async with httpcore.AsyncConnectionPool() as client:
 while 1:
 async with trio.open_nursery() as nursery:
 # This event is used to signal the first response we recieve.
 seen_response = trio.Event()

 # Kick off one hundred HTTP requests.
 for idx in range(100):
 nursery.start_soon(send_request, client, seen_response)
 
 # As soon as we see a response we can cancel out of the nursery,
 # rather than allowing all tasks to complete.
 await seen_response.wait()
 nursery.cancel_scope.cancel()

trio.run(main)

This should produce a httpcore.PoolTimeout error on the second or third iteration, after which the loop enters into a deadlock.

[karpetrosyan]

The issue appears to be HTTPConnections, which do not have the._connection attribute.

Example

import trio
import httpcore


TIMEOUT = {"read": 5, "write": 5, "connect": 5, "pool": 5}


async def send_request(client, seen_response):
 response = await client.request("GET", "http://127.0.0.1:8000", extensions={"timeout": TIMEOUT})

 # Print the first response that we see and set the "seen_response" flag.
 if not seen_response.is_set():
 seen_response.set()


async def main():
 async with httpcore.AsyncConnectionPool() as client:
 while 1:
 try:
 async with trio.open_nursery() as nursery:
 # This event is used to signal the first response we recieve.
 seen_response = trio.Event()

 # Kick off one hundred HTTP requests.
 for idx in range(100):
 nursery.start_soon(send_request, client, seen_response)
 
 # As soon as we see a response we can cancel out of the nursery,
 # rather than allowing all tasks to complete.
 await seen_response.wait()
 nursery.cancel_scope.cancel()
 except BaseException:
 print(client.connections[0].has_expired())
 print(client.connections[0].is_available())
 return

trio.run(main)

OUTPUT

False
False

This is the section of code where connection pools determine whether or not the connection can be used for the request.

httpcore/httpcore/_async/connection_pool.py

Lines 161 to 167 in b649bb0

	# Reuse an existing connection if one is currently available.
	for idx, connection in enumerate(self._pool):
	if connection.can_handle_request(origin) and connection.is_available():
	self._pool.pop(idx)
	self._pool.insert(0, connection)
	status.set_connection(connection)
	return True

It appears that these types of connections are not releasing the connection pool.
Connections that have not failed but are also not attempting to connect are not handled correctly.

How to make these kinds of connections. ?

This problem appears to occur if the newly created connection is cancelled before entering this try block.

httpcore/httpcore/_async/connection.py

Lines 75 to 99 in b649bb0

	try:
	stream = await self._connect(request)
	ssl_object = stream.get_extra_info("ssl_object")
	http2_negotiated = (
	ssl_object is not None
	and ssl_object.selected_alpn_protocol() == "h2"
	)
	if http2_negotiated or (self._http2 and not self._http1):
	from .http2 import AsyncHTTP2Connection
	self._connection = AsyncHTTP2Connection(
	origin=self._origin,
	stream=stream,
	keepalive_expiry=self._keepalive_expiry,
	)
	else:
	self._connection = AsyncHTTP11Connection(
	origin=self._origin,
	stream=stream,
	keepalive_expiry=self._keepalive_expiry,
	)
	except Exception as exc:
	self._connect_failed = True
	raise exc

I'm not sure this is what we're looking for; what do you think, @tomchristie ?

[karpetrosyan]

The simplified version of the reproducible code for this issue.

import httpcore

pool = httpcore.ConnectionPool()

for j in range(10):
 pool._pool.append(
 httpcore.HTTPConnection(httpcore.Origin(b"https", b"www.google.com", 443))
 )

response = pool.request(
 "GET", "https://www.google.com", extensions={"timeout": {"pool": 5}}
) # timeout error

[karpetrosyan]

@tomchristie Can you give me some advice here? Should I open a PR that resolves such cases?