Repository: encode/httpx

Issue:
Python 3.7: RuntimeError: read() called while another coroutine is already waiting for incoming data 
As we already discussed with @florimondmanca in https://github.com/encode/httpx/issues/382, I am reporting this issue here. 

I get the following error: `RuntimeError: read() called while another coroutine is already waiting for incoming data` when I am sending images to the server. As you can see from the code it is implemented that only max 100 requests are sent to the server at the same time. I know there could be lazy loading used but the limitation is made custom because before we send images to the server we check whether they are in the local cache or not. The limit of 100 is set that not all images are loaded in the memory at the same time (there can be also 10000 images).

Here is the trace:
```
Traceback (most recent call last):
  File "/Users/primoz/PycharmProjects/orange3-imageanalytics/example.py", line 63, in <module>
    asyncio.get_event_loop().run_until_complete(cl.embedd_batch())
  File "/Users/primoz/miniconda3/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "/Users/primoz/PycharmProjects/orange3-imageanalytics/example.py", line 18, in embedd_batch
    embeddings = await asyncio.gather(*requests)
  File "/Users/primoz/PycharmProjects/orange3-imageanalytics/example.py", line 41, in _send_to_server
    emb = await self._send_request(client, im, url)
  File "/Users/primoz/PycharmProjects/orange3-imageanalytics/example.py", line 56, in _send_request
    data=image
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/client.py", line 484, in post
    trust_env=trust_env,
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/client.py", line 626, in request
    trust_env=trust_env,
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/client.py", line 650, in send
    trust_env=trust_env,
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/client.py", line 265, in _get_response
    return await get_response(request)
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/middleware/redirect.py", line 31, in __call__
    response = await get_response(request)
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/client.py", line 226, in get_response
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/dispatch/connection_pool.py", line 126, in send
    raise exc
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/dispatch/connection_pool.py", line 121, in send
    request, verify=verify, cert=cert, timeout=timeout
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/dispatch/connection.py", line 62, in send
    response = await self.h2_connection.send(request, timeout=timeout)
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/dispatch/http2.py", line 52, in send
    status_code, headers = await self.receive_response(stream_id, timeout)
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/dispatch/http2.py", line 171, in receive_response
    event = await self.receive_event(stream_id, timeout)
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/dispatch/http2.py", line 206, in receive_event
    data = await self.stream.read(self.READ_NUM_BYTES, timeout, flag=flag)
  File "/Users/primoz/venv/orange/lib/python3.7/site-packages/httpx/concurrency/asyncio.py", line 114, in read
    data = await asyncio.wait_for(self.stream_reader.read(n), read_timeout)
  File "/Users/primoz/miniconda3/lib/python3.7/asyncio/tasks.py", line 416, in wait_for
    return fut.result()
  File "/Users/primoz/miniconda3/lib/python3.7/asyncio/streams.py", line 640, in read
    await self._wait_for_data('read')
  File "/Users/primoz/miniconda3/lib/python3.7/asyncio/streams.py", line 460, in _wait_for_data
    f'{func_name}() called while another coroutine is '
RuntimeError: read() called while another coroutine is already waiting for incoming data
```

The minimum example that reproduces the error is here:

```
import asyncio
import pickle

import httpx


class SendRequests:
    num_parallel_requests = 0
    MAX_PARALLEL = 100

    async def embedd_batch(self):
        requests = []
        async with httpx.AsyncClient(
                timeout=httpx.TimeoutConfig(timeout=60),
                base_url="https://api.garaza.io/") as client:
            for i in range(1000):
                requests.append(self._send_to_server(client))

            embeddings = await asyncio.gather(*requests)

        return embeddings

    async def __wait_until_released(self):
        while self.num_parallel_requests >= self.MAX_PARALLEL:
            await asyncio.sleep(0.1)

    async def _send_to_server(self, client):
        await self.__wait_until_released()

        self.num_parallel_requests += 1
        # simplified image loading 
        with open("image.pkl", "rb") as f:
            im = pickle.load(f)
        url = "/image/inception-v3?machine=1&session=1&retry=0"
        emb = await self._send_request(client, im, url)
        self.num_parallel_requests -= 1
        return emb

    async def _send_request(self, client, image, url):
        headers = {'Content-Type': 'image/jpeg',
                   'Content-Length': str(len(image))}
        response = await client.post(url, headers=headers, data=image)
        print(response.content)
        return response


if __name__ == "__main__":
    cl = SendRequests()
    asyncio.get_event_loop().run_until_complete(cl.embedd_batch())
```

All advice about this error is welcome. I tried to debug this error by myself but was not successful. 

The pickle file required in the code snippet is here: 
[image.pkl.zip](https://github.com/encode/httpx/files/3847317/image.pkl.zip)

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 34d47ec..3734b1f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## 0.7.8 (November 17, 2019)
+
+### Added
+
+- Add support for proxy tunnels for Python 3.6 + asyncio. (Pull #521)
+
 ## 0.7.7 (November 15, 2019)
 
 ### Fixed
diff --git a/docs/advanced.md b/docs/advanced.md
index 3e73f5b..1365645 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -179,6 +179,21 @@ password example-password
 ...
 ```
 
+## Unix Domain Sockets
+
+You can configure an `httpx` client to connect through a unix domain socket via the `uds` parameter. This is useful when making requests to a server that is bound to a socket file rather than an IP address.
+
+Here's an example requesting the Docker Engine API:
+
+```python
+import httpx
+
+
+with httpx.Client(uds="/var/run/docker.sock") as client:
+    # This request will connect through the socket file.
+    resp = client.get("http://localhost/version")
+```
+
 ## HTTP Proxying
 
 HTTPX supports setting up proxies the same way that Requests does via the `proxies` parameter.
diff --git a/docs/async.md b/docs/async.md
index 45620bf..c631d54 100644
--- a/docs/async.md
+++ b/docs/async.md
@@ -105,3 +105,38 @@ trio.run(main)
 
 !!! important
     `trio` must be installed to import and use the `TrioBackend`.
+
+## FAQ
+
+### When should I use an `AsyncClient`?
+
+You should use an `AsyncClient` whenever you are inside an *async environment*.
+
+In particular, using `httpx.get()` or `httpx.Client()` in an async environment is **not** supported. There are several technical reasons to this, but the rationale is that you shouldn't be doing blocking-style network calls in the context of an event loop (for more discussion, see [#179](https://github.com/encode/httpx/issues/179)).
+
+For example, this won't work:
+
+```python
+import asyncio
+import httpx
+
+async def main():
+    r = httpx.get("https://example.org")
+
+asyncio.run(main())
+```
+
+Instead, you should use an `AsyncClient`:
+
+```python
+import asyncio
+import httpx
+
+async def main():
+    async with httpx.AsyncClient() as client:
+        r = await client.get("https://example.org")
+
+asyncio.run(main())
+```
+
+If you *need* to make synchronous requests, or otherwise run into issues related to sync usage, you should probably consider using a regular threaded client such as [Requests](https://github.com/psf/requests) instead.
diff --git a/docs/compatibility.md b/docs/compatibility.md
index 250a02c..7b2b391 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -32,3 +32,7 @@ with httpx.Client(**kwargs) as client:
 ```
 
 More detailed documentation and usage of `Client` can be found in [Advanced Usage](advanced.md).
+
+## Mocking
+
+If you need to mock HTTPX the same way that test utilities like `responses` and `requests-mock` does for `requests`, see [RESPX](https://github.com/lundberg/respx).
diff --git a/httpx/__version__.py b/httpx/__version__.py
index 5783d84..2f47ae7 100644
--- a/httpx/__version__.py
+++ b/httpx/__version__.py
@@ -1,3 +1,3 @@
 __title__ = "httpx"
 __description__ = "A next generation HTTP client, for Python 3."
-__version__ = "0.7.7"
+__version__ = "0.7.8"
diff --git a/httpx/client.py b/httpx/client.py
index 45589b9..c327c2d 100644
--- a/httpx/client.py
+++ b/httpx/client.py
@@ -74,6 +74,7 @@ class BaseClient:
         app: typing.Callable = None,
         backend: ConcurrencyBackend = None,
         trust_env: bool = True,
+        uds: str = None,
     ):
         if backend is None:
             backend = AsyncioBackend()
@@ -99,6 +100,7 @@ class BaseClient:
                 pool_limits=pool_limits,
                 backend=backend,
                 trust_env=self.trust_env,
+                uds=uds,
             )
         elif isinstance(dispatch, Dispatcher):
             async_dispatch = ThreadedDispatcher(dispatch, backend)
@@ -721,6 +723,7 @@ class Client(BaseClient):
     async requests.
     * **trust_env** - *(optional)* Enables or disables usage of environment
     variables for configuration.
+    * **uds** - *(optional)* A path to a Unix domain socket to connect through.
     """
 
     def check_concurrency_backend(self, backend: ConcurrencyBackend) -> None:
diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
index 010d821..d3febbb 100644
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -51,6 +51,7 @@ class SocketStream(BaseSocketStream):
         self.stream_reader = stream_reader
         self.stream_writer = stream_writer
         self.timeout = timeout
+        self.read_lock = asyncio.Lock()
 
         self._inner: typing.Optional[SocketStream] = None
 
@@ -58,16 +59,49 @@ class SocketStream(BaseSocketStream):
         self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig
     ) -> "SocketStream":
         loop = asyncio.get_event_loop()
-        if not hasattr(loop, "start_tls"):  # pragma: no cover
-            raise NotImplementedError(
-                "asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+"
-            )
 
         stream_reader = asyncio.StreamReader()
         protocol = asyncio.StreamReaderProtocol(stream_reader)
         transport = self.stream_writer.transport
 
-        loop_start_tls = loop.start_tls  # type: ignore
+        if hasattr(loop, "start_tls"):
+            loop_start_tls = loop.start_tls  # type: ignore
+        else:
+
+            async def loop_start_tls(
+                transport: asyncio.BaseTransport,
+                protocol: asyncio.BaseProtocol,
+                sslcontext: ssl.SSLContext = None,
+                *,
+                server_side: bool = False,
+                server_hostname: str = None,
+                ssl_handshake_timeout: float = None,
+            ) -> asyncio.Transport:
+                """Python 3.6 asyncio doesn't have a start_tls() method on the loop
+                so we use this function in place of the loop's start_tls() method.
+                Adapted from this comment:
+                https://github.com/urllib3/urllib3/issues/1323#issuecomment-362494839
+                """
+                import asyncio.sslproto
+
+                waiter = loop.create_future()
+                ssl_protocol = asyncio.sslproto.SSLProtocol(
+                    loop,
+                    protocol,
+                    sslcontext,
+                    waiter,
+                    server_side=False,
+                    server_hostname=server_hostname,
+                    call_connection_made=False,
+                )
+
+                transport.set_protocol(ssl_protocol)
+                loop.call_soon(ssl_protocol.connection_made, transport)
+                loop.call_soon(transport.resume_reading)  # type: ignore
+
+                await waiter
+                return ssl_protocol._app_transport
+
         transport = await asyncio.wait_for(
             loop_start_tls(
                 transport=transport,
@@ -111,8 +145,10 @@ class SocketStream(BaseSocketStream):
             should_raise = flag is None or flag.raise_on_read_timeout
             read_timeout = timeout.read_timeout if should_raise else 0.01
             try:
-                data = await asyncio.wait_for(self.stream_reader.read(n), read_timeout)
-                break
+                async with self.read_lock:
+                    data = await asyncio.wait_for(
+                        self.stream_reader.read(n), read_timeout
+                    )
             except asyncio.TimeoutError:
                 if should_raise:
                     raise ReadTimeout() from None
@@ -122,6 +158,8 @@ class SocketStream(BaseSocketStream):
                 # doesn't seem to allow on 3.6.
                 # See: https://github.com/encode/httpx/issues/382
                 await asyncio.sleep(0)
+            else:
+                break
 
         return data
 
@@ -242,6 +280,29 @@ class AsyncioBackend(ConcurrencyBackend):
             stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout
         )
 
+    async def open_uds_stream(
+        self,
+        path: str,
+        hostname: typing.Optional[str],
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> SocketStream:
+        server_hostname = hostname if ssl_context else None
+
+        try:
+            stream_reader, stream_writer = await asyncio.wait_for(  # type: ignore
+                asyncio.open_unix_connection(
+                    path, ssl=ssl_context, server_hostname=server_hostname
+                ),
+                timeout.connect_timeout,
+            )
+        except asyncio.TimeoutError:
+            raise ConnectTimeout()
+
+        return SocketStream(
+            stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout
+        )
+
     async def run_in_threadpool(
         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:
diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py
index 9d5bffd..2109c21 100644
--- a/httpx/concurrency/base.py
+++ b/httpx/concurrency/base.py
@@ -124,6 +124,15 @@ class ConcurrencyBackend:
     ) -> BaseSocketStream:
         raise NotImplementedError()  # pragma: no cover
 
+    async def open_uds_stream(
+        self,
+        path: str,
+        hostname: typing.Optional[str],
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> BaseSocketStream:
+        raise NotImplementedError()  # pragma: no cover
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         raise NotImplementedError()  # pragma: no cover
 
diff --git a/httpx/concurrency/trio.py b/httpx/concurrency/trio.py
index 5d3b50d..f2ab919 100644
--- a/httpx/concurrency/trio.py
+++ b/httpx/concurrency/trio.py
@@ -32,6 +32,7 @@ class SocketStream(BaseSocketStream):
         self.stream = stream
         self.timeout = timeout
         self.write_buffer = b""
+        self.read_lock = trio.Lock()
         self.write_lock = trio.Lock()
 
     async def start_tls(
@@ -74,7 +75,8 @@ class SocketStream(BaseSocketStream):
             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)
+                async with self.read_lock:
+                    return await self.stream.receive_some(max_bytes=n)
 
             if should_raise:
                 raise ReadTimeout() from None
@@ -191,6 +193,26 @@ class TrioBackend(ConcurrencyBackend):
 
         return SocketStream(stream=stream, timeout=timeout)
 
+    async def open_uds_stream(
+        self,
+        path: str,
+        hostname: typing.Optional[str],
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> SocketStream:
+        connect_timeout = _or_inf(timeout.connect_timeout)
+
+        with trio.move_on_after(connect_timeout) as cancel_scope:
+            stream: trio.SocketStream = await trio.open_unix_socket(path)
+            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 SocketStream(stream=stream, timeout=timeout)
+
     async def run_in_threadpool(
         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:
diff --git a/httpx/dispatch/connection.py b/httpx/dispatch/connection.py
index 91feb97..0612bcc 100644
--- a/httpx/dispatch/connection.py
+++ b/httpx/dispatch/connection.py
@@ -38,6 +38,7 @@ class HTTPConnection(AsyncDispatcher):
         http_versions: HTTPVersionTypes = None,
         backend: ConcurrencyBackend = None,
         release_func: typing.Optional[ReleaseCallback] = None,
+        uds: typing.Optional[str] = None,
     ):
         self.origin = Origin(origin) if isinstance(origin, str) else origin
         self.ssl = SSLConfig(cert=cert, verify=verify, trust_env=trust_env)
@@ -45,6 +46,7 @@ class HTTPConnection(AsyncDispatcher):
         self.http_versions = HTTPVersionConfig(http_versions)
         self.backend = AsyncioBackend() if backend is None else backend
         self.release_func = release_func
+        self.uds = uds
         self.h11_connection = None  # type: typing.Optional[HTTP11Connection]
         self.h2_connection = None  # type: typing.Optional[HTTP2Connection]
 
@@ -84,8 +86,21 @@ class HTTPConnection(AsyncDispatcher):
         else:
             on_release = functools.partial(self.release_func, self)
 
-        logger.trace(f"start_connect host={host!r} port={port!r} timeout={timeout!r}")
-        stream = await self.backend.open_tcp_stream(host, port, ssl_context, timeout)
+        if self.uds is None:
+            logger.trace(
+                f"start_connect tcp host={host!r} port={port!r} timeout={timeout!r}"
+            )
+            stream = await self.backend.open_tcp_stream(
+                host, port, ssl_context, timeout
+            )
+        else:
+            logger.trace(
+                f"start_connect uds path={self.uds!r} host={host!r} timeout={timeout!r}"
+            )
+            stream = await self.backend.open_uds_stream(
+                self.uds, host, ssl_context, timeout
+            )
+
         http_version = stream.get_http_version()
         logger.trace(f"connected http_version={http_version!r}")
 
diff --git a/httpx/dispatch/connection_pool.py b/httpx/dispatch/connection_pool.py
index 189fcff..5d7d886 100644
--- a/httpx/dispatch/connection_pool.py
+++ b/httpx/dispatch/connection_pool.py
@@ -89,6 +89,7 @@ class ConnectionPool(AsyncDispatcher):
         pool_limits: PoolLimits = DEFAULT_POOL_LIMITS,
         http_versions: HTTPVersionTypes = None,
         backend: ConcurrencyBackend = None,
+        uds: typing.Optional[str] = None,
     ):
         self.verify = verify
         self.cert = cert
@@ -97,6 +98,7 @@ class ConnectionPool(AsyncDispatcher):
         self.http_versions = http_versions
         self.is_closed = False
         self.trust_env = trust_env
+        self.uds = uds
 
         self.keepalive_connections = ConnectionStore()
         self.active_connections = ConnectionStore()
@@ -142,6 +144,7 @@ class ConnectionPool(AsyncDispatcher):
                 backend=self.backend,
                 release_func=self.release_connection,
                 trust_env=self.trust_env,
+                uds=self.uds,
             )
             logger.trace(f"new_connection connection={connection!r}")
         else:
diff --git a/httpx/dispatch/http11.py b/httpx/dispatch/http11.py
index b1781bf..188d56d 100644
--- a/httpx/dispatch/http11.py
+++ b/httpx/dispatch/http11.py
@@ -4,6 +4,7 @@ import h11
 
 from ..concurrency.base import BaseSocketStream, ConcurrencyBackend, TimeoutFlag
 from ..config import TimeoutConfig, TimeoutTypes
+from ..exceptions import ConnectionClosed, ProtocolError
 from ..models import AsyncRequest, AsyncResponse
 from ..utils import get_logger
 
@@ -161,7 +162,17 @@ class HTTP11Connection:
         Read a single `h11` event, reading more data from the network if needed.
         """
         while True:
-            event = self.h11_state.next_event()
+            try:
+                event = self.h11_state.next_event()
+            except h11.RemoteProtocolError as e:
+                logger.debug(
+                    "h11.RemoteProtocolError exception "
+                    + f"their_state={self.h11_state.their_state} "
+                    + f"error_status_hint={e.error_status_hint}"
+                )
+                if self.stream.is_connection_dropped():
+                    raise ConnectionClosed(e)
+                raise ProtocolError(e)
 
             if isinstance(event, h11.Data):
                 logger.trace(f"receive_event event=Data(<{len(event.data)} bytes>)")
diff --git a/httpx/exceptions.py b/httpx/exceptions.py
index 81df38b..419a21a 100644
--- a/httpx/exceptions.py
+++ b/httpx/exceptions.py
@@ -150,6 +150,12 @@ class InvalidURL(HTTPError):
     """
 
 
+class ConnectionClosed(HTTPError):
+    """
+    Expected more data from peer, but connection was closed.
+    """
+
+
 class CookieConflict(HTTPError):
     """
     Attempted to lookup a cookie by name, but multiple cookies existed.


Test patch:
diff --git a/tests/client/test_async_client.py b/tests/client/test_async_client.py
index eaac5ef..42202aa 100644
--- a/tests/client/test_async_client.py
+++ b/tests/client/test_async_client.py
@@ -146,3 +146,14 @@ async def test_100_continue(server, backend):
 
     assert response.status_code == 200
     assert response.content == data
+
+
+async def test_uds(uds_server, backend):
+    url = uds_server.url
+    uds = uds_server.config.uds
+    assert uds is not None
+    async with httpx.AsyncClient(backend=backend, uds=uds) as client:
+        response = await client.get(url)
+    assert response.status_code == 200
+    assert response.text == "Hello, world!"
+    assert response.encoding == "iso-8859-1"
diff --git a/tests/client/test_client.py b/tests/client/test_client.py
index f7be607..5dc196d 100644
--- a/tests/client/test_client.py
+++ b/tests/client/test_client.py
@@ -138,6 +138,17 @@ def test_base_url(server):
     assert response.url == base_url
 
 
+def test_uds(uds_server):
+    url = uds_server.url
+    uds = uds_server.config.uds
+    assert uds is not None
+    with httpx.Client(uds=uds) as http:
+        response = http.get(url)
+    assert response.status_code == 200
+    assert response.text == "Hello, world!"
+    assert response.encoding == "iso-8859-1"
+
+
 def test_merge_url():
     client = httpx.Client(base_url="https://www.paypal.com/")
     url = client.merge_url("http://www.paypal.com")
diff --git a/tests/concurrency.py b/tests/concurrency.py
index 99d5d3f..0d7d135 100644
--- a/tests/concurrency.py
+++ b/tests/concurrency.py
@@ -5,8 +5,12 @@ required as part of the ConcurrencyBackend API.
 
 import asyncio
 import functools
+import typing
+
+import trio
 
 from httpx import AsyncioBackend
+from httpx.concurrency.trio import TrioBackend
 
 
 @functools.singledispatch
@@ -19,13 +23,24 @@ 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)
+
+
+@functools.singledispatch
+async def run_concurrently(backend, *coroutines: typing.Callable[[], typing.Awaitable]):
+    raise NotImplementedError  # pragma: no cover
+
+
+@run_concurrently.register(AsyncioBackend)
+async def _run_concurrently_asyncio(backend, *coroutines):
+    coros = (coroutine() for coroutine in coroutines)
+    await asyncio.gather(*coros)
+
 
-    @sleep.register(TrioBackend)
-    async def _sleep_trio(backend, seconds: int):
-        await trio.sleep(seconds)
+@run_concurrently.register(TrioBackend)
+async def _run_concurrently_trio(backend, *coroutines):
+    async with trio.open_nursery() as nursery:
+        for coroutine in coroutines:
+            nursery.start_soon(coroutine)
diff --git a/tests/conftest.py b/tests/conftest.py
index ef57cae..540b013 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -63,6 +63,8 @@ async def app(scope, receive, send):
     assert scope["type"] == "http"
     if scope["path"].startswith("/slow_response"):
         await slow_response(scope, receive, send)
+    elif scope["path"].startswith("/premature_close"):
+        await premature_close(scope, receive, send)
     elif scope["path"].startswith("/status"):
         await status_code(scope, receive, send)
     elif scope["path"].startswith("/echo_body"):
@@ -103,6 +105,16 @@ async def slow_response(scope, receive, send):
     await send({"type": "http.response.body", "body": b"Hello, world!"})
 
 
+async def premature_close(scope, receive, send):
+    await send(
+        {
+            "type": "http.response.start",
+            "status": 200,
+            "headers": [[b"content-type", b"text/plain"]],
+        }
+    )
+
+
 async def status_code(scope, receive, send):
     status_code = int(scope["path"].replace("/status/", ""))
     await send(
@@ -288,6 +300,15 @@ def server():
     yield from serve_in_thread(server)
 
 
+@pytest.fixture(scope=SERVER_SCOPE)
+def uds_server():
+    uds = "test_server.sock"
+    config = Config(app=app, lifespan="off", loop="asyncio", uds=uds)
+    server = TestServer(config=config)
+    yield from serve_in_thread(server)
+    os.remove(uds)
+
+
 @pytest.fixture(scope=SERVER_SCOPE)
 def https_server(cert_pem_file, cert_private_key_file):
     config = Config(
@@ -301,3 +322,19 @@ def https_server(cert_pem_file, cert_private_key_file):
     )
     server = TestServer(config=config)
     yield from serve_in_thread(server)
+
+
+@pytest.fixture(scope=SERVER_SCOPE)
+def https_uds_server(cert_pem_file, cert_private_key_file):
+    uds = "https_test_server.sock"
+    config = Config(
+        app=app,
+        lifespan="off",
+        ssl_certfile=cert_pem_file,
+        ssl_keyfile=cert_private_key_file,
+        uds=uds,
+        loop="asyncio",
+    )
+    server = TestServer(config=config)
+    yield from serve_in_thread(server)
+    os.remove(uds)
diff --git a/tests/dispatch/test_connections.py b/tests/dispatch/test_connections.py
index b4aea17..f39ac4f 100644
--- a/tests/dispatch/test_connections.py
+++ b/tests/dispatch/test_connections.py
@@ -1,4 +1,6 @@
-from httpx import HTTPConnection
+import pytest
+
+from httpx import HTTPConnection, exceptions
 
 
 async def test_get(server, backend):
@@ -15,6 +17,15 @@ async def test_post(server, backend):
         assert response.status_code == 200
 
 
+async def test_premature_close(server, backend):
+    with pytest.raises(exceptions.ConnectionClosed):
+        async with HTTPConnection(origin=server.url, backend=backend) as conn:
+            response = await conn.request(
+                "GET", server.url.copy_with(path="/premature_close")
+            )
+            await response.read()
+
+
 async def test_https_get_with_ssl_defaults(https_server, ca_cert_pem_file, backend):
     """
     An HTTPS request, with default SSL configuration set on the client.
diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py
index 8bb933b..878d6a8 100644
--- a/tests/test_concurrency.py
+++ b/tests/test_concurrency.py
@@ -1,10 +1,9 @@
-import sys
-
 import pytest
 import trio
 
 from httpx import AsyncioBackend, HTTPVersionConfig, SSLConfig, TimeoutConfig
 from httpx.concurrency.trio import TrioBackend
+from tests.concurrency import run_concurrently
 
 
 @pytest.mark.parametrize(
@@ -26,20 +25,26 @@ from httpx.concurrency.trio import TrioBackend
         ),
     ],
 )
-async def test_start_tls_on_socket_stream(https_server, backend, get_cipher):
+@pytest.mark.parametrize("use_uds", (False, True))
+async def test_start_tls_on_socket_stream(
+    https_server, https_uds_server, backend, get_cipher, use_uds
+):
     """
     See that the concurrency backend can make a connection without TLS then
     start TLS on an existing connection.
     """
-    if isinstance(backend, AsyncioBackend) and sys.version_info < (3, 7):
-        pytest.xfail(reason="Requires Python 3.7+ for AbstractEventLoop.start_tls()")
-
     ctx = SSLConfig().load_ssl_context_no_verify(HTTPVersionConfig())
     timeout = TimeoutConfig(5)
 
-    stream = await backend.open_tcp_stream(
-        https_server.url.host, https_server.url.port, None, timeout
-    )
+    if use_uds:
+        assert https_uds_server.config.uds is not None
+        stream = await backend.open_uds_stream(
+            https_uds_server.config.uds, https_uds_server.url.host, None, timeout
+        )
+    else:
+        stream = await backend.open_tcp_stream(
+            https_server.url.host, https_server.url.port, None, timeout
+        )
 
     try:
         assert stream.is_connection_dropped() is False
@@ -68,3 +73,19 @@ async def test_start_tls_on_socket_stream(https_server, backend, get_cipher):
 
     finally:
         await stream.close()
+
+
+async def test_concurrent_read(server, backend):
+    """
+    Regression test for: https://github.com/encode/httpx/issues/527
+    """
+    stream = await backend.open_tcp_stream(
+        server.url.host, server.url.port, ssl_context=None, timeout=TimeoutConfig(5)
+    )
+    try:
+        await stream.write(b"GET / HTTP/1.1\r\n\r\n")
+        await run_concurrently(
+            backend, lambda: stream.read(10), lambda: stream.read(10)
+        )
+    finally:
+        await stream.close()