Repository: encode/httpx

Issue:
Drop background manager (in favour of a two-function fork primitive).
I think we're running against the grain with our current `read`, `write`, `background_manager` API on the backends.

Switching to the same style as the `hip` project is using for its concurrency backends is likely a smart thing for us to do here... https://github.com/python-trio/hip/tree/master/src/urllib3/_backends

What would it take for us to switch over to a `send_and_receive_for_a_while` API?

@florimondmanca - I seem to remember you did some work on scoping this out at some point?


Discussion:
tomchristie: Nice one!

I think we should probably look at using `anyio`'s TaskGroups for the asyncio implementation... https://anyio.readthedocs.io/en/latest/tasks.html - that ensures we've got a *proper* nursery like primitive there.

(I essentially want to start *only* using TaskGroups for any `asyncio` branching, since *its the right approach*™️

We can also be stronger on our wording wrt. `MultiError`, it's not that "we don't support it yet" it's that we *specifically only care about raising one exception type*. It's enough for the API to expose that eg. "A read timeout occured" or "A write timeout occured". There's no need for us to support exposing the case that "A read timeout and a write timeout occured". *either one of those* is sufficient for the operation as a whole to be cancelled, and it's actually okay in this circumstance for us to just return either indeterminatly.
florimondmanca: Are we okay adding a dependency on anyio for this? We don’t *need* it — correctly handling errors in the case of 2 coroutines is easy enough, and the current implementation yields the expected behavior.
tomchristie: Yeah, I'm actually pretty okay with it...

I'm strongly enough of the camp that `asyncio` circa Python 3.8 does *not* provide suitable primitives for async task branching, and that a reasonable policy at this point in time is to *never* use the low level APIs there.

There's been talk of introducing a `TaskSupervisor` class into Python, which might also have a backport-installable package, in which case we could switch to that. Until that point in time the only sensible nursery-like-primitive for asyncio is the `anyio` implementation.

If we stick with that then we *know* we're properly adhering to the structued concurrency constraints, even in our asyncio implementation.

Once we've got some specifc async docs back in again I'd also like to help us lead our users down the structured concurrency path, and have us recommend `anyio`'s TaskGroups to end users for task-branching with `asyncio`.
tomchristie: If we think that's being overly zealous, then I *guess we could* stick with a plain asyncio implementation there, but I'd be more comfortable with us leaning on `anyio`'s TaskGroups.
tomchristie: Let's leave it *as an open option*.

This is looking great as it stands so we'll get this in, the later tweak the comment note on `MutliError`, and consider `anyio`. (It could also be that we don't strictly need it ourselves, but could highly recommend TaskGroups to end users)

Patch:
diff --git a/httpx/__init__.py b/httpx/__init__.py
index 686359a..b6cd6df 100644
--- a/httpx/__init__.py
+++ b/httpx/__init__.py
@@ -3,12 +3,7 @@ from .api import delete, get, head, options, patch, post, put, request, stream
 from .auth import BasicAuth, DigestAuth
 from .client import Client
 from .concurrency.asyncio import AsyncioBackend
-from .concurrency.base import (
-    BaseBackgroundManager,
-    BasePoolSemaphore,
-    BaseSocketStream,
-    ConcurrencyBackend,
-)
+from .concurrency.base import BasePoolSemaphore, BaseSocketStream, ConcurrencyBackend
 from .config import (
     USER_AGENT,
     CertTypes,
@@ -89,7 +84,6 @@ __all__ = [
     "VerifyTypes",
     "HTTPConnection",
     "BasePoolSemaphore",
-    "BaseBackgroundManager",
     "ConnectionPool",
     "HTTPProxy",
     "HTTPProxyMode",
diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
index a759716..eee4ec3 100644
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -3,12 +3,10 @@ import functools
 import ssl
 import sys
 import typing
-from types import TracebackType
 
 from ..config import PoolLimits, Timeout
 from ..exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout
 from .base import (
-    BaseBackgroundManager,
     BaseEvent,
     BasePoolSemaphore,
     BaseSocketStream,
@@ -317,34 +315,30 @@ class AsyncioBackend(ConcurrencyBackend):
         finally:
             self._loop = loop
 
+    async def fork(
+        self,
+        coroutine1: typing.Callable,
+        args1: typing.Sequence,
+        coroutine2: typing.Callable,
+        args2: typing.Sequence,
+    ) -> None:
+        task1 = self.loop.create_task(coroutine1(*args1))
+        task2 = self.loop.create_task(coroutine2(*args2))
+
+        try:
+            await asyncio.gather(task1, task2)
+        finally:
+            pending: typing.Set[asyncio.Future[typing.Any]]  # Please mypy.
+            _, pending = await asyncio.wait({task1, task2}, timeout=0)
+            for task in pending:
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         return PoolSemaphore(limits)
 
     def create_event(self) -> BaseEvent:
         return typing.cast(BaseEvent, asyncio.Event())
-
-    def background_manager(
-        self, coroutine: typing.Callable, *args: typing.Any
-    ) -> "BackgroundManager":
-        return BackgroundManager(coroutine, args)
-
-
-class BackgroundManager(BaseBackgroundManager):
-    def __init__(self, coroutine: typing.Callable, args: typing.Any) -> None:
-        self.coroutine = coroutine
-        self.args = args
-
-    async def __aenter__(self) -> "BackgroundManager":
-        loop = asyncio.get_event_loop()
-        self.task = loop.create_task(self.coroutine(*self.args))
-        return self
-
-    async def __aexit__(
-        self,
-        exc_type: typing.Type[BaseException] = None,
-        exc_value: BaseException = None,
-        traceback: TracebackType = None,
-    ) -> None:
-        await self.task
-        if exc_type is None:
-            self.task.result()
diff --git a/httpx/concurrency/auto.py b/httpx/concurrency/auto.py
index 3dd31a8..3b57e56 100644
--- a/httpx/concurrency/auto.py
+++ b/httpx/concurrency/auto.py
@@ -5,7 +5,6 @@ import sniffio
 
 from ..config import PoolLimits, Timeout
 from .base import (
-    BaseBackgroundManager,
     BaseEvent,
     BasePoolSemaphore,
     BaseSocketStream,
@@ -52,8 +51,3 @@ class AutoBackend(ConcurrencyBackend):
 
     def create_event(self) -> BaseEvent:
         return self.backend.create_event()
-
-    def background_manager(
-        self, coroutine: typing.Callable, *args: typing.Any
-    ) -> BaseBackgroundManager:
-        return self.backend.background_manager(coroutine, *args)
diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py
index f32501a..ff5f72f 100644
--- a/httpx/concurrency/base.py
+++ b/httpx/concurrency/base.py
@@ -1,6 +1,5 @@
 import ssl
 import typing
-from types import TracebackType
 
 from ..config import PoolLimits, Timeout
 
@@ -154,27 +153,20 @@ class ConcurrencyBackend:
     def create_event(self) -> BaseEvent:
         raise NotImplementedError()  # pragma: no cover
 
-    def background_manager(
-        self, coroutine: typing.Callable, *args: typing.Any
-    ) -> "BaseBackgroundManager":
-        raise NotImplementedError()  # pragma: no cover
-
-
-class BaseBackgroundManager:
-    async def __aenter__(self) -> "BaseBackgroundManager":
-        raise NotImplementedError()  # pragma: no cover
-
-    async def __aexit__(
+    async def fork(
         self,
-        exc_type: typing.Type[BaseException] = None,
-        exc_value: BaseException = None,
-        traceback: TracebackType = None,
+        coroutine1: typing.Callable,
+        args1: typing.Sequence,
+        coroutine2: typing.Callable,
+        args2: typing.Sequence,
     ) -> None:
-        raise NotImplementedError()  # pragma: no cover
+        """
+        Run two coroutines concurrently.
+
+        This should start 'coroutine1' with '*args1' and 'coroutine2' with '*args2',
+        and wait for them to finish.
 
-    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)
+        In case one of the coroutines raises an exception, cancel the other one then
+        raise. If the other coroutine had also raised an exception, ignore it (for now).
+        """
+        raise NotImplementedError()  # pragma: no cover
diff --git a/httpx/concurrency/trio.py b/httpx/concurrency/trio.py
index 4af4242..f1fc7c4 100644
--- a/httpx/concurrency/trio.py
+++ b/httpx/concurrency/trio.py
@@ -1,14 +1,12 @@
 import functools
 import ssl
 import typing
-from types import TracebackType
 
 import trio
 
 from ..config import PoolLimits, Timeout
 from ..exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout
 from .base import (
-    BaseBackgroundManager,
     BaseEvent,
     BasePoolSemaphore,
     BaseSocketStream,
@@ -204,17 +202,31 @@ class TrioBackend(ConcurrencyBackend):
             functools.partial(coroutine, **kwargs) if kwargs else coroutine, *args
         )
 
+    async def fork(
+        self,
+        coroutine1: typing.Callable,
+        args1: typing.Sequence,
+        coroutine2: typing.Callable,
+        args2: typing.Sequence,
+    ) -> None:
+        try:
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(coroutine1, *args1)
+                nursery.start_soon(coroutine2, *args2)
+        except trio.MultiError as exc:
+            # NOTE: asyncio doesn't handle multi-errors yet, so we must align on its
+            # behavior here, and need to arbitrarily decide which exception to raise.
+            # We may want to add an 'httpx.MultiError', manually add support
+            # for this situation in the asyncio backend, and re-raise
+            # an 'httpx.MultiError' from trio's here.
+            raise exc.exceptions[0]
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         return PoolSemaphore(limits)
 
     def create_event(self) -> BaseEvent:
         return Event()
 
-    def background_manager(
-        self, coroutine: typing.Callable, *args: typing.Any
-    ) -> "BackgroundManager":
-        return BackgroundManager(coroutine, *args)
-
 
 class Event(BaseEvent):
     def __init__(self) -> None:
@@ -233,25 +245,3 @@ class Event(BaseEvent):
         # 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/http2.py b/httpx/dispatch/http2.py
index 09368c8..471ba9b 100644
--- a/httpx/dispatch/http2.py
+++ b/httpx/dispatch/http2.py
@@ -65,9 +65,23 @@ class HTTP2Connection:
         self.timeout_flags[stream_id] = TimeoutFlag()
         self.window_update_received[stream_id] = self.backend.create_event()
 
-        task, args = self.send_request_data, [stream_id, request.stream(), timeout]
-        async with self.backend.background_manager(task, *args):
+        status_code: typing.Optional[int] = None
+        headers: typing.Optional[list] = None
+
+        async def receive_response(stream_id: int, timeout: Timeout) -> None:
+            nonlocal status_code, headers
             status_code, headers = await self.receive_response(stream_id, timeout)
+
+        await self.backend.fork(
+            self.send_request_data,
+            [stream_id, request.stream(), timeout],
+            receive_response,
+            [stream_id, timeout],
+        )
+
+        assert status_code is not None
+        assert headers is not None
+
         content = self.body_iter(stream_id, timeout)
         on_close = functools.partial(self.response_closed, stream_id=stream_id)
 


Test patch:
diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py
index 4fecaa6..0898dd4 100644
--- a/tests/test_concurrency.py
+++ b/tests/test_concurrency.py
@@ -3,7 +3,7 @@ import trio
 
 from httpx import AsyncioBackend, SSLConfig, Timeout
 from httpx.concurrency.trio import TrioBackend
-from tests.concurrency import run_concurrently
+from tests.concurrency import run_concurrently, sleep
 
 
 def get_asyncio_cipher(stream):
@@ -110,3 +110,49 @@ async def test_concurrent_read(server, backend):
         )
     finally:
         await stream.close()
+
+
+async def test_fork(backend):
+    ok_counter = 0
+
+    async def ok(delay: int) -> None:
+        nonlocal ok_counter
+        await sleep(backend, delay)
+        ok_counter += 1
+
+    async def fail(message: str, delay: int) -> None:
+        await sleep(backend, delay)
+        raise RuntimeError(message)
+
+    await backend.fork(ok, [0], ok, [0])
+    assert ok_counter == 2
+
+    with pytest.raises(RuntimeError, match="Oops"):
+        await backend.fork(ok, [0], fail, ["Oops", 0.01])
+
+    assert ok_counter == 3
+
+    with pytest.raises(RuntimeError, match="Oops"):
+        await backend.fork(ok, [0.01], fail, ["Oops", 0])
+
+    assert ok_counter == 3
+
+    with pytest.raises(RuntimeError, match="Oops"):
+        await backend.fork(fail, ["Oops", 0.01], ok, [0])
+
+    assert ok_counter == 4
+
+    with pytest.raises(RuntimeError, match="Oops"):
+        await backend.fork(fail, ["Oops", 0], ok, [0.01])
+
+    assert ok_counter == 4
+
+    with pytest.raises(RuntimeError, match="My bad"):
+        await backend.fork(fail, ["My bad", 0], fail, ["Oops", 0.01])
+
+    with pytest.raises(RuntimeError, match="Oops"):
+        await backend.fork(fail, ["My bad", 0.01], fail, ["Oops", 0])
+
+    # No 'match', since we can't know which will be raised first.
+    with pytest.raises(RuntimeError):
+        await backend.fork(fail, ["My bad", 0], fail, ["Oops", 0])