Support async cancellations

[karpetrosyan]

Problem

Cancellations in the middle of a request/response cycle are not supported by the httpcore.

Something like this should brake the connection pool

importanyioimporthttpcoreasyncdefmain():
pool=httpcore.AsyncConnectionPool()
withanyio.move_on_after(1):
awaitpool.request(
"GET", "http://example.com"
) # long running requestprint(pool.connections) # One active connection that will remain active indefinitely# [<AsyncHTTPConnection ['http://example.com:80', HTTP/1.1, ACTIVE, Request Count: 1]>]anyio.run(main)

TODO

• Write a failing test
• Make a test pass

[lovelydinosaur]

I don't think this is quite correct.
The behaviour we'd want to see here is that there is a connection in the pool, but it is IDLE.

[karpetrosyan]

I thought about it and can't remember when, but I remember thinking that this is much safer.

[lovelydinosaur]

Ah yes, there's some complexities here... 🤔

Either way, this test case is a great starting point.

[karpetrosyan]

First, we must decide how we will support "CancelScope" while keeping sync and async interfaces similar. I'm not sure whether it's better to do it in another PR or in this one.

[lovelydinosaur]

First, we must decide how we will support "CancelScope" while keeping sync and async interfaces similar.

Okay. Before we get to implementation, could you start off by showing me where we need shielding rather than how we're going to implement it?

In order to have confidence that we're really getting it right I'd be interested in looking at that from the ground up...

Suppose we're just working directly with an HTTPConnection instance directly, rather than with a connection pool. Where would we need cancellation shielding?

[karpetrosyan]

httpcore/httpcore/_async/http11.py

Lines 220 to 232 in eb5957d

	asyncdef_response_closed(self) ->None:
	asyncwithself._state_lock:
	if (
	self._h11_state.our_stateish11.DONE
	andself._h11_state.their_stateish11.DONE
	):
	self._state=HTTPConnectionState.IDLE
	self._h11_state.start_next_cycle()
	ifself._keepalive_expiryisnotNone:
	now=time.monotonic()
	self._expire_at=now+self._keepalive_expiry
	else:
	awaitself.aclose()

This function appears to be the main source of the problem; the issue here is that if we call this method while cancelled, the resources will not be closed.

[lovelydinosaur]

Perhaps we should add a test case at that level of the API?

(Resolve cancellation shielding for an individual connection, and be confident we've got that sorted before moving up the stack?)

[karpetrosyan]

Sounds good

[lovelydinosaur]

Okay these are looking great - good starting point for us working through this really comprehensively.

[karpetrosyan]

If we're done with failing tests, I believe we should decide how and where we want to see shield logic.

[lovelydinosaur]

Okay, so based on the HTTP/1.1 test case I've got a nice clear example of the existing behaviour...

Here's the behaviour we see when we get a network error during a write operation on an HTTP/1.1 connection:

importtrioimporthttpcoreclassSlowStream(httpcore.AsyncNetworkStream):
asyncdefwrite(self, buffer, timeout=None):
raisehttpcore.WriteError()
asyncdefaclose(self):
passasyncdefmain():
origin=httpcore.Origin(b"http", b"example.com", 80)
stream=SlowStream()
asyncwithhttpcore.AsyncHTTP11Connection(origin, stream) asconn:
try:
awaitconn.request("GET", "http://example.com")
excepthttpcore.NetworkError:
passprint(conn)
# <AsyncHTTP11Connection ['http://example.com:80', CLOSED, Request Count: 1]>

In contrast, here's the behaviour we see when we get a cancellation exception during a write operation on an HTTP/1.1 connection:

importtrioimporthttpcoreclassSlowStream(httpcore.AsyncNetworkStream):
asyncdefwrite(self, buffer, timeout=None):
awaittrio.sleep(1)
asyncdefaclose(self):
passasyncdefmain():
origin=httpcore.Origin(b"http", b"example.com", 80)
stream=SlowStream()
asyncwithhttpcore.AsyncHTTP11Connection(origin, stream) asconn:
withtrio.move_on_after(0.001):
awaitconn.request("GET", "http://example.com")
print(conn)
# <AsyncHTTP11Connection ['http://example.com:80', ACTIVE, Request Count: 1]>trio.run(main)

[lovelydinosaur]

Starting with the HTTP/1.1 case it looks to me like we want...

• Two test cases - the existing one that we currently have for a timeout during writing, and an additional one for a timeout during reading.
• Shielding semantics around our exception handling in the handle_async_request method...

exceptBaseExceptionasexc:
withShieldFromCancellation():
asyncwithTrace("response_closed", logger, request) astrace:
awaitself._response_closed()
raiseexc

• Shielding semantics around our exception handling in the byte stream __aiter__ method...

exceptBaseExceptionasexc:
# If we get an exception while streaming the response,# we want to close the response (and possibly the connection)# before raising that exception.withShieldFromCancellation():
awaitself.aclose()
raiseexc

I'm not presuming how we implement the ShieldFromCancellation here, just presenting what I think the semantics ought to look like.

(Based on trio's design in this problem space. Because structured concurrency FTW.)

[karpetrosyan]

To me, it appears to be completely correct.

[karpetrosyan]

Two test cases - the existing one that we currently have for a timeout during writing, and an additional one for a timeout during reading.

Are you referring to slow-write test for ConnectionPool or HTTP11Connection?

[lovelydinosaur]

Are you referring to slow-write test for ConnectionPool or HTTP11Connection?

I'm suggesting that we'll also want slow-read tests.

Here's how that'd look for the HTTP/1.1 case...

classSlowReadStream(httpcore.AsyncNetworkStream):
def__init__(self, buffer):
self.buffer=bufferasyncdefwrite(self, buffer, timeout=None):
passasyncdefread(self, max_bytes: int, timeout: typing.Optional[float] =None) ->bytes:
ifnotself._buffer:
awaittrio.sleep(1)
else:
returnself._buffer.pop(0)
asyncdefaclose(self):
passstream=SlowReadStream([
b"HTTP/1.1 200 OK\r\n",
b"Content-Type: plain/text\r\n",
b"Content-Length: 1000\r\n",
b"\r\n",
b"Hello, world!...", # The stream will hang after this portion of the incomplete response,# so that we're able to test our async cancellation semantics.
])

[karpetrosyan]

Is a buffer really necessary?

[lovelydinosaur]

Is a buffer really necessary?

You'd need this or something like this in order to test the cancellation handling for HTTP11ConnectionByteStream.__aiter__.

[lovelydinosaur]

Okay, want to try...

• Adding the ShieldCancellation from Support async cancellations. #726
• Making the first HTTP/1.1 test pass.
• Making the second HTTP/1.1 test pass.

[lovelydinosaur]

Thanks so much for your work on this.
I've started pulling it into #726, so I think we can close this one as being superseeded at this point.