Refactoring... context managed timeouts.

[tomchristie]

Prompted by review @karpetrosyan's work on #936.

I think there's some preceding work we could do here to improve timeouts in httpcore so that we've got an easier-to-work-with set of primitives. I think we probably want to tackle this refactoring before dealing with introducing overall timeouts, since it deals with the underlying issue.

Here's where we currently are...

We need a consistent approach to timeouts across both sync and async, which is currently awkward because the existing primitives in each case are different. Standard threaded codebases use .set_timeout on the socket. Async codebases use context managed timeout blocks, such as trio.fail_after().

Currently we mediate between these two differing styles by ensuring that all socket operations take a timeout=... parameter. Within the async backends we use .fail_after() within these operations. This ensures that we've got an API that's consistent across both sync & async cases...

sync...

content = stream.read(4096, timeout=1.0)

async...

content = await stream.read(4096, timeout=1.0)

Here's where I think we should be...

The thing is, trio gets the design of timeouts right. Timeouts as context managed blocks is easier to reason about, and allows us to handle cases such as "overall" timeouts in an obvious way.

Really we'd like to have the alternate style here, all the way through our stack.

sync...

with timeout(1.0):
    ...
    content = stream.read(4096)

async...

async with timeout(1.0):
    ...
    content = await stream.read(4096)

We should be aiming to drop the timeout parameter from the NetworkStream API, and have context-managed timeouts that are dealt with outside of that layer.

An example to demonstrate...

Here's a little sync API with context managed timeouts...

import socket
import time
import types
import contextvars
import contextlib

_timeout_stack = contextvars.ContextVar("timeout_context", default=())

def _get_current_timeout() -> float | None:
    stack = _timeout_stack.get()
    if not stack:
        return None
    soonest = min(stack)
    now = time.monotonic()
    remaining = soonest - now
    if remaining <= 0.0:
        raise TimeoutError()
    return remaining


class Stream:
    def __init__(self, sock: socket.socket) -> None:
        self.socket = sock

    def read(self, max_bytes: int = 64 * 1024) -> bytes:
        timeout = _get_current_timeout()
        self.socket.settimeout(timeout)
        return self.socket.recv(max_bytes)

    def write(self, buffer: bytes) -> None:
        while buffer:
            timeout = _get_current_timeout()
            self.socket.settimeout(timeout)
            n = self.socket.send(buffer)
            buffer = buffer[n:]

    def close(self) -> None:
        timeout = _get_current_timeout()
        self.socket.settimeout(timeout)
        self.socket.close()

    # Context managed usage...
    def __enter__(self) -> "Stream":
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None = None,
        exc_value: BaseException | None = None,
        traceback: types.TracebackType | None = None,
    ):
        self.close()


def connect(host: str, port: int) -> Stream:
    """
    Connect to the given address, returning a Stream instance.

    ```
    with connect("example.com", 80) as stream:
        ...
    ```
    """
    addr = (host, port)
    timeout = _get_current_timeout()
    sock = socket.create_connection(addr, timeout=timeout)
    return Stream(sock)


@contextlib.contextmanager
def timeout(duration: float) -> None:
    """
    A context managed timeout API.

    ```
    with timeout(1.0):
        ...
    ```
    """
    now = time.monotonic()
    until = now + duration
    stack = _timeout_stack.get()
    token = _timeout_stack.set((until,) + stack)
    try:
        yield
    finally:
        _timeout_stack.reset(token)

Using the API then ~looks like this...

def send_request() -> bytes:
    # Connect.
    with timeout(1.0):
        stream = connect("www.example.com", 80)

    # Ensure we close the stream properly on completion.
    with stream:
        # Send the request.
        with timeout(5.0):
            request = b"GET / HTTP/1.1\r\nHost: www.example.com\r\nConnection: close\r\n\r\n"
            stream.write(request)    
        # Read the response.
        with timeout(5.0):
            response = b''
            while True:
                content = stream.read()
                if content:
                    response += content
                else:
                    break
 
    # Done.
    return response


content = send_request()
print(content)

How we can apply this in HTTP core...

• I'd suggest that we attempt a refactoring where each of the NetworkBackend classes implement a timeout context manager, rather than having a timeout=... parameter on the socket operations.
• Our "read" and "write" timeouts would be applied at the HTTP/1.1 and HTTP/2 layer.
• This then puts us in a position to implement overall timeouts using the same API.

[graingert]

It should probably be possible to cancel the timeout or change the deadline from another thread by using https://github.com/python-trio/trio/wiki/notes-to-self#blocking-read-hackpy