Repository: encode/httpx

Issue:
Add support for a Trio ConcurrencyBackend

Discussion:
sethmlarson: :tada: Very exciting!
florimondmanca: Yup, good idea @sethmlarson. As you mentioned this can (should?) be implemented in a separate PR. :-)
florimondmanca: I'm encountering a bug when performing a real-life HTTP/2  + TLS request with the `TrioBackend`:

```python
# example.py
import trio
import httpx
from httpx.concurrency.trio import TrioBackend


async def main(url):
    async with httpx.AsyncClient(backend=TrioBackend()) as client:
        await client.get(url)


trio.run(main, "https://example.com")
```

```console
$ python example.py
Traceback (most recent call last):
  File "example.py", line 11, in <module>
    trio.run(main, "https://example.com")
  File "/Users/florimond/Developer/python-projects/httpx/venv/lib/python3.7/site-packages/trio/_core/_run.py", line 1783, in run
    raise runner.main_task_outcome.error
  File "example.py", line 8, in main
    await client.get(url)
  File "/Users/florimond/Developer/python-projects/httpx/httpx/client.py", line 363, in get
    trust_env=trust_env,
  File "/Users/florimond/Developer/python-projects/httpx/httpx/client.py", line 610, in request
    trust_env=trust_env,
  File "/Users/florimond/Developer/python-projects/httpx/httpx/client.py", line 181, in send
    allow_redirects=allow_redirects,
  File "/Users/florimond/Developer/python-projects/httpx/httpx/client.py", line 218, in send_handling_redirects
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/florimond/Developer/python-projects/httpx/httpx/dispatch/connection_pool.py", line 120, in send
    raise exc
  File "/Users/florimond/Developer/python-projects/httpx/httpx/dispatch/connection_pool.py", line 115, in send
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/florimond/Developer/python-projects/httpx/httpx/dispatch/connection.py", line 57, in send
    response = await self.h2_connection.send(request, timeout=timeout)
  File "/Users/florimond/Developer/python-projects/httpx/httpx/dispatch/http2.py", line 45, in send
    status_code, headers = await self.receive_response(stream_id, timeout)
  File "/Users/florimond/Developer/python-projects/httpx/httpx/dispatch/http2.py", line 119, in receive_response
    event = await self.receive_event(stream_id, timeout)
  File "/Users/florimond/Developer/python-projects/httpx/httpx/dispatch/http2.py", line 158, in receive_event
    await self.stream.write(data_to_send, timeout)
  File "/Users/florimond/Developer/python-projects/httpx/httpx/concurrency/trio.py", line 86, in write
    await self.stream.send_all(data)
  File "/Users/florimond/Developer/python-projects/httpx/venv/lib/python3.7/site-packages/trio/_ssl.py", line 684, in send_all
    with self._outer_send_conflict_detector:
  File "/Users/florimond/Developer/python-projects/httpx/venv/lib/python3.7/site-packages/trio/_util.py", line 122, in __enter__
    raise trio.BusyResourceError(self._msg)
trio.BusyResourceError: another task is currently sending data on this SSLStream
```

Everything is fine if I force `http_versions=["HTTP/1.1"]` on `AsyncClient`, so HTTP/2 is indeed failing here.

The place where this error is raised alternates between `.receive_event()` (as above) and `.end_stream()`. They run in parallel via the background manager, respectively as part of receiving the response (which also requires sending some data back to the server) and sending request data.

But, quoting the description of [`BusyResourceError`](https://trio.readthedocs.io/en/latest/reference-core.html#trio.BusyResourceError) in the Trio docs:

> If two tasks try to send data through the same socket at the same time, Trio will raise `BusyResourceError` instead of letting the data get scrambled.

And there's no way around it — probably a way for Trio to tell us to do things differently.

…This is where the "read/write for a while" approach that @tomchristie mentioned [here](https://github.com/encode/httpx/issues/246#issuecomment-522930937) as an alternative to `BackgroundMananger` starts to make sense to me. I don't know exactly how it would look if implemented, but hopefully it ensure that we're not writing to the stream at the same time from two different places?

So from my understanding, right now the Trio backend can't be made compatible with HTTP/2, and we need to rework the sending/reception of request/response data to completely unblock this.

Since HTTP/1.1 is working though, what would be everyone's thoughts on disabling HTTP/2 on trio for now temporarily, so we can get Trio + HTTP/1.1 in now, and add HTTP/2 support later?

(I managed to force usage of HTTP/1.1 by executing:

```python
ssl_context.set_alpn_protocols(["h1"])
ssl_context.set_npn_protocols(["h1"])
```

before opening the `SSLStream`.)
florimondmanca: @sethmlarson Yes, if asyncio does not have guards in place it’s very possible that we’re writing to the stream writer at the same time as well, potentially sending corrupted data (although up to now it doesn’t seem to have surfaced?).
florimondmanca: I pushed a commit demonstrating reversing the way we do things: serving the server in a separate thread, which allows us to do all async stuff in the main thread and the proper async environment.

(I've made pytest share the server during the whole test session, so as a benefit tests are now super fast since we don't need to set it up/tear it down on every test. Also we don't need to run sync tests in the threadpool anymore.)

The commit also turns the server restart into a no-op to prove it has no effect on the tests.

There's still a failing tests for the case where an exception is raised within the ASGI app. The failure comes from how we're handling that case — mostly not feeding the exception back into the background manager in `response.on_close()`. I tried fiddling around with it but it's tricky — the comment we put there stating that the background manager is a weak design solution is definitely relevant.
tomchristie: > it allowed to remove the need for write_no_block() on the concurrency backend (which trio does not have an obvious equivalent for). Things seem to be okay, but if I missed any reasons why it was not async in the first place, let me know!

So the reasoning there was because `initiate_connection` needs to be an atomic operation, that's performed exactly once on the first stream, before any other streams start sending messages.

If we can ensure that it's not async then we know that'll be the case since we can't task switch around it (an alternative would be to introduce a lock around it.)
tomchristie: > I pushed a commit demonstrating reversing the way we do things: serving the server in a separate thread, which allows us to do all async stuff in the main thread and a "native" async environment.

Would there be any sense in creating a pull request that *only* does that, without the rest of this, so that we can get that into the test suite and confirm it, independently of the trio work?
florimondmanca: Yup @tomchristie, I’ll isolate it into its own PR. :)
sethmlarson: Just wondering since I just found the project, would `anyio` work for us?
florimondmanca: @sethmlarson https://github.com/encode/httpx/issues/296 😄 

florimondmanca: I managed to have the failing ASGI test pass — we needed to close the background manager before re-raising the app exception if it occurred.

The remaining blockers are:

- Even with the threaded Uvicorn server, we're getting a `ConnectionResetError` when it restarts in the connection pool tests. Could be unblocked by #290.
- We need to rethink the HTTP/2 read/write handling to comply with Trio's sensible "only one task can write to a stream at a time" policy.
tomchristie: > We need to rethink the HTTP/2 read/write handling to comply with Trio's sensible "only one task can write to a stream at a time" policy.

Could you expand on that aspect?
florimondmanca: @tomchristie I tried to explain that in the comment above: https://github.com/encode/httpx/pull/276#issuecomment-525469745

Let me know if there’s anything else to clarify. Didn’t really have the time to think about a solution yet, but it relates to some ideas you’ve been mentioning lately.
tomchristie: Okay - I guess the answer there is to put locking around the read/writes.
florimondmanca: So, an update on my investigations regarding the HTTP/2 issue…

> Okay - I guess the answer there is to put locking around the read/writes.

It was the right thing to do — I added a write lock around the failing `.send_all()` call.

After that I was receiving a `PROTOCOL_ERROR` event. (The program was hanging because we didn't handle error events, so I added a fix for that as well — perhaps worth its own PR?). I fixed it by sending buffered data in a separate `.write()` call instead of concatenating it to the data to send.

**tl;dr: HTTP/2 is now operational on trio :tada:**

Updated example I used to compare the flow between asyncio and trio via `HTTPX_DEBUG`:

```python
# example.py
import asyncio
import trio
import httpx
from httpx.concurrency.trio import TrioBackend


async def main(url, backend):
    async with httpx.AsyncClient(backend=backend) as client:
        await client.get(url)


URL = "https://example.com"
asyncio.run(main(URL, httpx.AsyncioBackend()))
input("\nEnter to continue...\n")
trio.run(main, URL, TrioBackend())
```
florimondmanca: Last item to tackle from my point of view is the failing tests due to restarts of the uvicorn server, which *should* be solvable by using Hypercorn — #290.

Here's the traceback showing the `ConnectionResetError` in one of the tests:

```console
___________________________ test_connection_closed_free_semaphore_on_acquire[TrioBackend] ___________________________

    @contextmanager
    def _translate_socket_errors_to_stream_errors():
        try:
>           yield

venv/lib/python3.7/site-packages/trio/_highlevel_socket.py:30: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <trio.SocketStream object at 0x10de7d2e8>, max_bytes = 4096

    async def receive_some(self, max_bytes=None):
        if max_bytes is None:
            max_bytes = DEFAULT_RECEIVE_SIZE
        if max_bytes < 1:
            raise ValueError("max_bytes must be >= 1")
        with _translate_socket_errors_to_stream_errors():
>           return await self.socket.recv(max_bytes)

venv/lib/python3.7/site-packages/trio/_highlevel_socket.py:144: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <trio.socket.socket fd=54, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 55977)>
args = (4096,), kwargs = {}

    @_wraps(fn, assigned=("__name__",), updated=())
    async def wrapper(self, *args, **kwargs):
>       return await self._nonblocking_helper(fn, args, kwargs, wait_fn)

venv/lib/python3.7/site-packages/trio/_socket.py:348: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <trio.socket.socket fd=54, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 55977)>
fn = <method 'recv' of '_socket.socket' objects>, args = (4096,), kwargs = {}
wait_fn = <function wait_readable at 0x1064b31e0>

    async def _nonblocking_helper(self, fn, args, kwargs, wait_fn):
        # We have to reconcile two conflicting goals:
        # - We want to make it look like we always blocked in doing these
        #   operations. The obvious way is to always do an IO wait before
        #   calling the function.
        # - But, we also want to provide the correct semantics, and part
        #   of that means giving correct errors. So, for example, if you
        #   haven't called .listen(), then .accept() raises an error
        #   immediately. But in this same circumstance, then on macOS, the
        #   socket does not register as readable. So if we block waiting
        #   for read *before* we call accept, then we'll be waiting
        #   forever instead of properly raising an error. (On Linux,
        #   interestingly, AFAICT a socket that can't possible read/write
        #   *does* count as readable/writable for select() purposes. But
        #   not on macOS.)
        #
        # So, we have to call the function once, with the appropriate
        # cancellation/yielding sandwich if it succeeds, and if it gives
        # BlockingIOError *then* we fall back to IO wait.
        #
        # XX think if this can be combined with the similar logic for IOCP
        # submission...
        async with _try_sync():
>           return fn(self._sock, *args, **kwargs)
E           ConnectionResetError: [Errno 54] Connection reset by peer

venv/lib/python3.7/site-packages/trio/_socket.py:588: ConnectionResetError

The above exception was the direct cause of the following exception:

value = <trio.Nursery object at 0x10df149e8>

    async def yield_(value=None):
>       return await _yield_(value)

venv/lib/python3.7/site-packages/async_generator/_impl.py:106: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv/lib/python3.7/site-packages/async_generator/_impl.py:99: in _yield_
    return (yield _wrap(value))
tests/dispatch/test_connection_pools.py:180: in test_connection_closed_free_semaphore_on_acquire
    response = await http.request("GET", server.url)
httpx/dispatch/base.py:40: in request
    return await self.send(request, verify=verify, cert=cert, timeout=timeout)
httpx/dispatch/connection_pool.py:126: in send
    raise exc
httpx/dispatch/connection_pool.py:121: in send
    request, verify=verify, cert=cert, timeout=timeout
httpx/dispatch/connection.py:65: in send
    response = await self.h11_connection.send(request, timeout=timeout)
httpx/dispatch/http11.py:53: in send
    http_version, status_code, headers = await self._receive_response(timeout)
httpx/dispatch/http11.py:133: in _receive_response
    event = await self._receive_event(timeout)
httpx/dispatch/http11.py:174: in _receive_event
    self.READ_NUM_BYTES, timeout, flag=self.timeout_flag
httpx/concurrency/trio.py:57: in read
    data = await self.stream.receive_some(max_bytes=n)
venv/lib/python3.7/site-packages/trio/_highlevel_socket.py:144: in receive_some
    return await self.socket.recv(max_bytes)
../../../.pyenv/versions/3.7.3/lib/python3.7/contextlib.py:130: in __exit__
    self.gen.throw(type, value, traceback)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    @contextmanager
    def _translate_socket_errors_to_stream_errors():
        try:
            yield
        except OSError as exc:
            if exc.errno in _closed_stream_errnos:
                raise trio.ClosedResourceError(
                    "this socket was already closed"
                ) from None
            else:
                raise trio.BrokenResourceError(
                    "socket connection broken: {}".format(exc)
>               ) from exc
E               trio.BrokenResourceError: socket connection broken: [Errno 54] Connection reset by peer

venv/lib/python3.7/site-packages/trio/_highlevel_socket.py:39: BrokenResourceError
```
tomchristie: > failing tests due to restarts of the uvicorn server

Could you link to where that's discussed? I've lost track of it. Ta!
florimondmanca: @tomchristie I don’t think I discussed it in much more detail, but tl;dr is that the server restart that we’re doing in three of the connection pool tests to close keep alive connections is seemingly causing trio to lose track of the socket.

It makes sense because I *think* shutting down the uvicorn server closes the socket since the underlying asyncio server is shut down entirely.

Is there a way to close the connections in server.shutdown() without shutting down the whole server, and to then leave the sockets untouched? That might do the trick and not require us to switch to hypercorn — which might actually not even solve this problem.

Perhaps this is actually a problem related to uvicorn not closing the connections gracefully enough. The ConnectionResetError seems pretty rough — is uvicorn missing sending some kind of connection termination packet before closing it?
florimondmanca: Aaaaalright. I pushed a fix for the server restart issue in the latest commit.

I first realized that `.is_connection_dropped()` was returning `False` after restart when it should return `True`, causing the connetion pool to reuse the actually broken connection, leading to the very much relevant `BrokenResourceError` I saw.

Then, after investigating why `.is_eof` was not set properly, I realized that the sentinel value `b""` [that trio expects implementations to return](https://trio.readthedocs.io/en/stable/reference-io.html#trio.abc.ReceiveStream.receive_some) wasn't always returned. (Though it's not clear in this context who plays the role of the "implementation" here?)

More precisely:

- uvicorn + httptools has a final line of the body with a `b"0\r\n\r\n"` suffix:

```console
21:58:29.411 - httpx.dispatch.http11 - receive_event event=NEED_DATA
DATA: b'HTTP/1.1 200 OK\r\ndate: Fri, 06 Sep 2019 19:58:29 GMT\r\nserver: uvicorn\r\ncontent-type: text/plain\r\ntransfer-encoding: chunked\r\n\r\nd\r\nHello, world!\r\n0\r\n\r\n'
21:58:29.412 - httpx.dispatch.http11 - receive_event event=Response(status_code=200, headers=[(b'date', b'Fri, 06 Sep 2019 19:58:29 GMT'), (b'server', b'uvicorn'), (b'content-type', b'text/plain'), (b'transfer-encoding', b'chunked')], http_version=b'1.1', reason=b'OK')
21:58:29.413 - httpx.dispatch.http11 - receive_event event=Data(<13 bytes>)
21:58:29.413 - httpx.dispatch.http11 - receive_event event=EndOfMessage(headers=[])
```

- OTOH, uvicorn + h11 includes that final string of bytes in a separate line, which triggers an extra `NEED_DATA` event:

```console
22:01:14.662 - httpx.dispatch.http11 - receive_event event=NEED_DATA
DATA: b'HTTP/1.1 200 OK\r\ndate: Fri, 06 Sep 2019 20:01:14 GMT\r\nserver: uvicorn\r\ncontent-type: text/plain\r\ntransfer-encoding: chunked\r\n\r\nd\r\nHello, world!\r\n'
22:01:14.666 - httpx.dispatch.http11 - receive_event event=Response(status_code=200, headers=[(b'date', b'Fri, 06 Sep 2019 20:01:14 GMT'), (b'server', b'uvicorn'), (b'content-type', b'text/plain'), (b'transfer-encoding', b'chunked')], http_version=b'1.1', reason=b'OK')
22:01:14.667 - httpx.dispatch.http11 - receive_event event=Data(<13 bytes>)
22:01:14.667 - httpx.dispatch.http11 - receive_event event=NEED_DATA
DATA: b'0\r\n\r\n'
22:01:14.668 - httpx.dispatch.http11 - receive_event event=EndOfMessage(headers=[])
```

So I figured, I just need to say "we reached EOF" if we get `b""` OR something that ends with `b"0\r\n\r\n"`.

I'm really not sure what that final string of bytes represents though — is it standard in HTTP? — so not sure whether this is a dirty hack, a proper fix, or an indication that something might be wrong somewhere (us, h11/httptools, uvicorn, etc).

ANYWAY…

The build is still failing because of the uvloop compilation issue on 3.8-dev, BUT all other tests are passing. :tada: :tada: :tada:
florimondmanca: I think this is finally ready for review from everyone. :)
yeraydiazdiaz: I gave this branch a go with the top 10 Alexa sites: 9 of them returned a response just fine with most of them using HTTP2. There was a single ConnectionTimeout on qq.com and some sites returned `413 Request Entity Too Large`.

Really cool stuff 🥇 
florimondmanca: Thanks for trying it out! @yeraydiazdiaz! Do you get similar results with the asyncio backend?
yeraydiazdiaz: Yep, results are the same for the top 10 sites. Here's the quick and dirty script 😅 

```python
import asyncio
import logging
import sys
from pathlib import Path

import trio

import httpx
from httpx.concurrency.trio import TrioBackend

logging.basicConfig(
    format="[%(levelname)s %(asctime)s] (%(name)s) %(message)s", datefmt="%H:%M:%S"
)
log = logging.getLogger("httpx")
log.setLevel(logging.DEBUG)


async def do_request(url: str) -> None:
    cl = httpx.AsyncClient()
    resp = await cl.get(url)
    print("Success:", resp)


async def do_trio_request(url: str) -> None:
    async with httpx.AsyncClient(backend=TrioBackend()) as client:
        resp = await client.get(url)
        print("Success:", resp)


def do_run(backend, url):
    if backend == "asyncio":
        asyncio.run(do_request(url))
    else:
        trio.run(do_trio_request, url)


if __name__ == "__main__":
    limit = 1_000
    try:
        limit = int(sys.argv[1])
    except:
        pass

    backend = "asyncio"
    try:
        backend = sys.argv[2]
    except:
        pass

    print(f"Limiting to {limit} requests...")

    with open(Path(__file__).parent / "alexa-1m.txt") as fd:
        for i, url in enumerate(fd.readlines(), 1):
            url = url.strip()
            if input(f"\n\nAbout to fetch {url}, is that ok? [Y/n]: ").lower() in (
                "n",
                "no",
            ):
                break
            try:
                do_run(backend, url)
            except httpx.exceptions.HTTPError as e:
                log.exception(e)
                print("\n\n\n ERROR \n\n\n")
                pass

            if i == limit:
                break
```

it assumes the [top 1M Alexa sites text file](https://github.com/encode/httpx/files/3521345/alexa-1m.txt.gz) is in the same directory
florimondmanca: So, I think I got confused about what a *readable* socket means — which means the implementation originally pushed last week was correct. I added a bunch of code comments along with the last commit to clarify the asyncio and trio implementations of `.is_connection_dropped()`.

If all looks well, I'd love a new review from @sethmlarson, and @tomchristie's approval for merging this. (I know there were past discussions about whether trio support should land in core, or be released as a third-party package. My position is that we should merge this in core *first*, and then possibly extract it into a separate package once we have enough live experience.)
sethmlarson: Now that 0.7.3 is out you ready to merge this?
florimondmanca: Yup, if this sounds good to you then let's merge. I'll add docs in a subsequent PR. :)
sethmlarson: It's happening!!! :boom: 

Patch:
diff --git a/httpx/client.py b/httpx/client.py
index 266d980..653e7d8 100644
--- a/httpx/client.py
+++ b/httpx/client.py
@@ -81,7 +81,7 @@ class BaseClient:
             if param_count == 2:
                 dispatch = WSGIDispatch(app=app)
             else:
-                dispatch = ASGIDispatch(app=app)
+                dispatch = ASGIDispatch(app=app, backend=backend)
 
         self.trust_env = True if trust_env is None else trust_env
 
diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
index 1a145be..696e32c 100644
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -112,6 +112,20 @@ class TCPStream(BaseTCPStream):
                     raise WriteTimeout() from None
 
     def is_connection_dropped(self) -> bool:
+        # Counter-intuitively, what we really want to know here is whether the socket is
+        # *readable*, i.e. whether it would return immediately with empty bytes if we
+        # called `.recv()` on it, indicating that the other end has closed the socket.
+        # See: https://github.com/encode/httpx/pull/143#issuecomment-515181778
+        #
+        # As it turns out, asyncio checks for readability in the background
+        # (see: https://github.com/encode/httpx/pull/276#discussion_r322000402),
+        # so checking for EOF or readability here would yield the same result.
+        #
+        # At the cost of rigour, we check for EOF instead of readability because asyncio
+        # does not expose any public API to check for readability.
+        # (For a solution that uses private asyncio APIs, see:
+        # https://github.com/encode/httpx/pull/143#issuecomment-515202982)
+
         return self.stream_reader.at_eof()
 
     async def close(self) -> None:
diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py
index 63fb143..fc784b3 100644
--- a/httpx/concurrency/base.py
+++ b/httpx/concurrency/base.py
@@ -187,3 +187,10 @@ class BaseBackgroundManager:
         traceback: TracebackType = None,
     ) -> None:
         raise NotImplementedError()  # pragma: no cover
+
+    async def close(self, exception: BaseException = None) -> None:
+        if exception is None:
+            await self.__aexit__(None, None, None)
+        else:
+            traceback = exception.__traceback__  # type: ignore
+            await self.__aexit__(type(exception), exception, traceback)
diff --git a/httpx/concurrency/trio.py b/httpx/concurrency/trio.py
new file mode 100644
index 0000000..14d7986
--- /dev/null
+++ b/httpx/concurrency/trio.py
@@ -0,0 +1,255 @@
+import functools
+import math
+import ssl
+import typing
+from types import TracebackType
+
+import trio
+
+from ..config import PoolLimits, TimeoutConfig
+from ..exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout
+from .base import (
+    BaseBackgroundManager,
+    BaseEvent,
+    BasePoolSemaphore,
+    BaseQueue,
+    BaseTCPStream,
+    ConcurrencyBackend,
+    TimeoutFlag,
+)
+
+
+def _or_inf(value: typing.Optional[float]) -> float:
+    return value if value is not None else float("inf")
+
+
+class TCPStream(BaseTCPStream):
+    def __init__(
+        self,
+        stream: typing.Union[trio.SocketStream, trio.SSLStream],
+        timeout: TimeoutConfig,
+    ) -> None:
+        self.stream = stream
+        self.timeout = timeout
+        self.write_buffer = b""
+        self.write_lock = trio.Lock()
+
+    def get_http_version(self) -> str:
+        if not isinstance(self.stream, trio.SSLStream):
+            return "HTTP/1.1"
+
+        ident = self.stream.selected_alpn_protocol()
+        if ident is None:
+            return "HTTP/1.1"
+
+        return "HTTP/2" if ident == "h2" else "HTTP/1.1"
+
+    async def read(
+        self, n: int, timeout: TimeoutConfig = None, flag: TimeoutFlag = None
+    ) -> bytes:
+        if timeout is None:
+            timeout = self.timeout
+
+        while True:
+            # Check our flag at the first possible moment, and use a fine
+            # grained retry loop if we're not yet in read-timeout mode.
+            should_raise = flag is None or flag.raise_on_read_timeout
+            read_timeout = _or_inf(timeout.read_timeout if should_raise else 0.01)
+
+            with trio.move_on_after(read_timeout):
+                return await self.stream.receive_some(max_bytes=n)
+
+            if should_raise:
+                raise ReadTimeout() from None
+
+    def is_connection_dropped(self) -> bool:
+        # Adapted from: https://github.com/encode/httpx/pull/143#issuecomment-515202982
+        stream = self.stream
+
+        # Peek through any SSLStream wrappers to get the underlying SocketStream.
+        while hasattr(stream, "transport_stream"):
+            stream = stream.transport_stream
+        assert isinstance(stream, trio.SocketStream)
+
+        # Counter-intuitively, what we really want to know here is whether the socket is
+        # *readable*, i.e. whether it would return immediately with empty bytes if we
+        # called `.recv()` on it, indicating that the other end has closed the socket.
+        # See: https://github.com/encode/httpx/pull/143#issuecomment-515181778
+        return stream.socket.is_readable()
+
+    def write_no_block(self, data: bytes) -> None:
+        self.write_buffer += data
+
+    async def write(
+        self, data: bytes, timeout: TimeoutConfig = None, flag: TimeoutFlag = None
+    ) -> None:
+        if self.write_buffer:
+            previous_data = self.write_buffer
+            # Reset before recursive call, otherwise we'll go through
+            # this branch indefinitely.
+            self.write_buffer = b""
+            try:
+                await self.write(previous_data, timeout=timeout, flag=flag)
+            except WriteTimeout:
+                self.writer_buffer = previous_data
+                raise
+
+        if not data:
+            return
+
+        if timeout is None:
+            timeout = self.timeout
+
+        write_timeout = _or_inf(timeout.write_timeout)
+
+        while True:
+            with trio.move_on_after(write_timeout):
+                async with self.write_lock:
+                    await self.stream.send_all(data)
+                break
+            # We check our flag at the first possible moment, in order to
+            # allow us to suppress write timeouts, if we've since
+            # switched over to read-timeout mode.
+            should_raise = flag is None or flag.raise_on_write_timeout
+            if should_raise:
+                raise WriteTimeout() from None
+
+    async def close(self) -> None:
+        await self.stream.aclose()
+
+
+class PoolSemaphore(BasePoolSemaphore):
+    def __init__(self, pool_limits: PoolLimits):
+        self.pool_limits = pool_limits
+
+    @property
+    def semaphore(self) -> typing.Optional[trio.Semaphore]:
+        if not hasattr(self, "_semaphore"):
+            max_connections = self.pool_limits.hard_limit
+            if max_connections is None:
+                self._semaphore = None
+            else:
+                self._semaphore = trio.Semaphore(
+                    max_connections, max_value=max_connections
+                )
+        return self._semaphore
+
+    async def acquire(self) -> None:
+        if self.semaphore is None:
+            return
+
+        timeout = _or_inf(self.pool_limits.pool_timeout)
+
+        with trio.move_on_after(timeout):
+            await self.semaphore.acquire()
+            return
+
+        raise PoolTimeout()
+
+    def release(self) -> None:
+        if self.semaphore is None:
+            return
+
+        self.semaphore.release()
+
+
+class TrioBackend(ConcurrencyBackend):
+    async def open_tcp_stream(
+        self,
+        hostname: str,
+        port: int,
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> TCPStream:
+        connect_timeout = _or_inf(timeout.connect_timeout)
+
+        with trio.move_on_after(connect_timeout) as cancel_scope:
+            stream: trio.SocketStream = await trio.open_tcp_stream(hostname, port)
+            if ssl_context is not None:
+                stream = trio.SSLStream(stream, ssl_context, server_hostname=hostname)
+                await stream.do_handshake()
+
+        if cancel_scope.cancelled_caught:
+            raise ConnectTimeout()
+
+        return TCPStream(stream=stream, timeout=timeout)
+
+    async def run_in_threadpool(
+        self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
+    ) -> typing.Any:
+        return await trio.to_thread.run_sync(
+            functools.partial(func, **kwargs) if kwargs else func, *args
+        )
+
+    def run(
+        self, coroutine: typing.Callable, *args: typing.Any, **kwargs: typing.Any
+    ) -> typing.Any:
+        return trio.run(
+            functools.partial(coroutine, **kwargs) if kwargs else coroutine, *args
+        )
+
+    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()
+
+    def background_manager(
+        self, coroutine: typing.Callable, *args: typing.Any
+    ) -> "BackgroundManager":
+        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()
+
+    def set(self) -> None:
+        self._event.set()
+
+    def is_set(self) -> bool:
+        return self._event.is_set()
+
+    async def wait(self) -> None:
+        await self._event.wait()
+
+    def clear(self) -> None:
+        # trio.Event.clear() was deprecated in Trio 0.12.
+        # https://github.com/python-trio/trio/issues/637
+        self._event = trio.Event()
+
+
+class BackgroundManager(BaseBackgroundManager):
+    def __init__(self, coroutine: typing.Callable, *args: typing.Any) -> None:
+        self.coroutine = coroutine
+        self.args = args
+        self.nursery_manager = trio.open_nursery()
+        self.nursery: typing.Optional[trio.Nursery] = None
+
+    async def __aenter__(self) -> "BackgroundManager":
+        self.nursery = await self.nursery_manager.__aenter__()
+        self.nursery.start_soon(self.coroutine, *self.args)
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: typing.Type[BaseException] = None,
+        exc_value: BaseException = None,
+        traceback: TracebackType = None,
+    ) -> None:
+        assert self.nursery is not None
+        await self.nursery_manager.__aexit__(exc_type, exc_value, traceback)
diff --git a/httpx/dispatch/asgi.py b/httpx/dispatch/asgi.py
index c56d757..0633ea8 100644
--- a/httpx/dispatch/asgi.py
+++ b/httpx/dispatch/asgi.py
@@ -130,6 +130,7 @@ class ASGIDispatch(AsyncDispatcher):
         await response_started_or_failed.wait()
 
         if app_exc is not None and self.raise_app_exceptions:
+            await background.close(app_exc)
             raise app_exc
 
         assert status_code is not None, "application did not return a response."
@@ -138,7 +139,7 @@ class ASGIDispatch(AsyncDispatcher):
         async def on_close() -> None:
             nonlocal response_body
             await response_body.drain()
-            await background.__aexit__(None, None, None)
+            await background.close(app_exc)
             if app_exc is not None and self.raise_app_exceptions:
                 raise app_exc
 
diff --git a/httpx/dispatch/http2.py b/httpx/dispatch/http2.py
index 4ddd29f..b1efe6a 100644
--- a/httpx/dispatch/http2.py
+++ b/httpx/dispatch/http2.py
@@ -6,6 +6,7 @@ import h2.events
 
 from ..concurrency.base import BaseEvent, BaseTCPStream, ConcurrencyBackend, TimeoutFlag
 from ..config import TimeoutConfig, TimeoutTypes
+from ..exceptions import ProtocolError
 from ..models import AsyncRequest, AsyncResponse
 from ..utils import get_logger
 
@@ -187,6 +188,10 @@ class HTTP2Connection:
                 logger.debug(
                     f"receive_event stream_id={event_stream_id} event={event!r}"
                 )
+
+                if hasattr(event, "error_code"):
+                    raise ProtocolError(event)
+
                 if isinstance(event, h2.events.WindowUpdated):
                     if event_stream_id == 0:
                         for window_update_event in self.window_update_received.values():
diff --git a/setup.cfg b/setup.cfg
index a9fc1a8..299bac6 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -11,7 +11,7 @@ combine_as_imports = True
 force_grid_wrap = 0
 include_trailing_comma = True
 known_first_party = httpx,tests
-known_third_party = brotli,certifi,chardet,cryptography,h11,h2,hstspreload,nox,pytest,requests,rfc3986,setuptools,trustme,uvicorn
+known_third_party = brotli,certifi,chardet,cryptography,h11,h2,hstspreload,nox,pytest,requests,rfc3986,setuptools,trio,trustme,uvicorn
 line_length = 88
 multi_line_output = 3
 
diff --git a/setup.py b/setup.py
index 8bd9d60..298d06f 100644
--- a/setup.py
+++ b/setup.py
@@ -59,6 +59,7 @@ setup(
         "idna==2.*",
         "rfc3986==1.*",
     ],
+    extras_require={"trio": ["trio"]},
     classifiers=[
         "Development Status :: 3 - Alpha",
         "Environment :: Web Environment",


Test patch:
diff --git a/test-requirements.txt b/test-requirements.txt
index cbef387..870e558 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -1,4 +1,4 @@
--e .
+-e .[trio]
 
 # Optional
 brotlipy==0.7.*
@@ -11,6 +11,7 @@ isort
 mypy
 pytest
 pytest-asyncio
+pytest-trio
 pytest-cov
 trustme
 uvicorn
diff --git a/tests/concurrency.py b/tests/concurrency.py
index 1240034..99d5d3f 100644
--- a/tests/concurrency.py
+++ b/tests/concurrency.py
@@ -17,3 +17,15 @@ async def sleep(backend, seconds: int):
 @sleep.register(AsyncioBackend)
 async def _sleep_asyncio(backend, seconds: int):
     await asyncio.sleep(seconds)
+
+
+try:
+    import trio
+    from httpx.concurrency.trio import TrioBackend
+except ImportError:  # pragma: no cover
+    pass
+else:
+
+    @sleep.register(TrioBackend)
+    async def _sleep_trio(backend, seconds: int):
+        await trio.sleep(seconds)
diff --git a/tests/conftest.py b/tests/conftest.py
index 658a6f9..ed9ea34 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -47,7 +47,17 @@ def clean_environ() -> typing.Dict[str, typing.Any]:
     os.environ.update(original_environ)
 
 
-@pytest.fixture(params=[pytest.param(AsyncioBackend, marks=pytest.mark.asyncio)])
+backend_params = [pytest.param(AsyncioBackend, marks=pytest.mark.asyncio)]
+
+try:
+    from httpx.concurrency.trio import TrioBackend
+except ImportError:  # pragma: no cover
+    pass
+else:
+    backend_params.append(pytest.param(TrioBackend, marks=pytest.mark.trio))
+
+
+@pytest.fixture(params=backend_params)
 def backend(request):
     backend_cls = request.param
     return backend_cls()
diff --git a/tests/dispatch/test_connection_pools.py b/tests/dispatch/test_connection_pools.py
index 2765808..6f4361d 100644
--- a/tests/dispatch/test_connection_pools.py
+++ b/tests/dispatch/test_connection_pools.py
@@ -168,7 +168,9 @@ async def test_connection_closed_free_semaphore_on_acquire(server, restart, back
     Verify that max_connections semaphore is released
     properly on a disconnected connection.
     """
-    async with httpx.ConnectionPool(pool_limits=httpx.PoolLimits(hard_limit=1)) as http:
+    async with httpx.ConnectionPool(
+        pool_limits=httpx.PoolLimits(hard_limit=1), backend=backend
+    ) as http:
         response = await http.request("GET", server.url)
         await response.read()
 
diff --git a/tests/test_timeouts.py b/tests/test_timeouts.py
index 765c8f7..8dc1e08 100644
--- a/tests/test_timeouts.py
+++ b/tests/test_timeouts.py
@@ -38,7 +38,7 @@ async def test_connect_timeout(server, backend):
 
 
 async def test_pool_timeout(server, backend):
-    pool_limits = PoolLimits(hard_limit=1, pool_timeout=1e-6)
+    pool_limits = PoolLimits(hard_limit=1, pool_timeout=1e-4)
 
     async with AsyncClient(pool_limits=pool_limits, backend=backend) as client:
         response = await client.get(server.url, stream=True)