Repository: encode/httpx

Issue:
HTTP/2 download speeds / flow control settings
Which having an initial look at HTTP/3, @jlaine dug out that we seem to have signiifcantly slow uploading right now.

First thing to do would be to investigate and replicate, by eg. compare and contrast a simple upload from `httpx` vs `requests/urllib3` - does it replicate trivially and reliably? This really *shouldn't* be an issue for us so will need some looking at.

Initial thoughts on where issues could be:

* We're sending too-small chunks, and always waiting for the network to fully drain on each chunk is negatively impacting performance.
* We're sending too-small chunks, and ending up with largish computation overhead relative to data transfer as a result.
* We've got some big unseen overhead in the sync-to-async marshalling.
* Something else?

Discussion:

Patch:
diff --git a/docs/compatibility.md b/docs/compatibility.md
index 8f4a19b..506340c 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -4,7 +4,7 @@ HTTPX aims to be compatible with the `requests` API wherever possible.
 
 This documentation outlines places where the API differs...
 
-## Request URLS
+## Request URLs
 
 Accessing `response.url` will return a `URL` instance, rather than a string.
 Use `str(response.url)` if you need a string instance.
diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
index c73a138..bb795b1 100644
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -242,6 +242,10 @@ class AsyncioBackend(ConcurrencyBackend):
 
         return SocketStream(stream_reader=stream_reader, stream_writer=stream_writer)
 
+    def time(self) -> float:
+        loop = asyncio.get_event_loop()
+        return loop.time()
+
     async def run_in_threadpool(
         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:
diff --git a/httpx/concurrency/auto.py b/httpx/concurrency/auto.py
index c90ce24..c11c064 100644
--- a/httpx/concurrency/auto.py
+++ b/httpx/concurrency/auto.py
@@ -41,6 +41,9 @@ class AutoBackend(ConcurrencyBackend):
     ) -> BaseSocketStream:
         return await self.backend.open_uds_stream(path, hostname, ssl_context, timeout)
 
+    def time(self) -> float:
+        return self.backend.time()
+
     def get_semaphore(self, max_value: int) -> BasePoolSemaphore:
         return self.backend.get_semaphore(max_value)
 
diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py
index da6dd98..366e574 100644
--- a/httpx/concurrency/base.py
+++ b/httpx/concurrency/base.py
@@ -99,6 +99,9 @@ class ConcurrencyBackend:
     ) -> BaseSocketStream:
         raise NotImplementedError()  # pragma: no cover
 
+    def time(self) -> float:
+        raise NotImplementedError()  # pragma: no cover
+
     def get_semaphore(self, max_value: int) -> BasePoolSemaphore:
         raise NotImplementedError()  # pragma: no cover
 
diff --git a/httpx/concurrency/trio.py b/httpx/concurrency/trio.py
index 6f2ab2d..0c64988 100644
--- a/httpx/concurrency/trio.py
+++ b/httpx/concurrency/trio.py
@@ -156,6 +156,9 @@ class TrioBackend(ConcurrencyBackend):
             functools.partial(coroutine, **kwargs) if kwargs else coroutine, *args
         )
 
+    def time(self) -> float:
+        return trio.current_time()
+
     def get_semaphore(self, max_value: int) -> BasePoolSemaphore:
         return PoolSemaphore(max_value)
 
diff --git a/httpx/dispatch/connection.py b/httpx/dispatch/connection.py
index 009895a..15ed455 100644
--- a/httpx/dispatch/connection.py
+++ b/httpx/dispatch/connection.py
@@ -38,6 +38,7 @@ class HTTPConnection(Dispatcher):
         self.release_func = release_func
         self.uds = uds
         self.open_connection: typing.Optional[OpenConnection] = None
+        self.expires_at: typing.Optional[float] = None
 
     async def send(
         self,
diff --git a/httpx/dispatch/connection_pool.py b/httpx/dispatch/connection_pool.py
index 4fe3998..db23640 100644
--- a/httpx/dispatch/connection_pool.py
+++ b/httpx/dispatch/connection_pool.py
@@ -78,6 +78,8 @@ class ConnectionStore:
 
 
 class ConnectionPool(Dispatcher):
+    KEEP_ALIVE_EXPIRY = 5.0
+
     def __init__(
         self,
         *,
@@ -101,6 +103,7 @@ class ConnectionPool(Dispatcher):
         self.active_connections = ConnectionStore()
 
         self.backend = lookup_backend(backend)
+        self.next_keepalive_check = 0.0
 
     @property
     def max_connections(self) -> BasePoolSemaphore:
@@ -118,6 +121,21 @@ class ConnectionPool(Dispatcher):
     def num_connections(self) -> int:
         return len(self.keepalive_connections) + len(self.active_connections)
 
+    async def check_keepalive_expiry(self) -> None:
+        now = self.backend.time()
+        if now < self.next_keepalive_check:
+            return
+        self.next_keepalive_check = now + 1.0
+
+        # Iterate through all the keep alive connections.
+        # We create a list here to avoid any 'changed during iteration' errors.
+        keepalives = list(self.keepalive_connections.all.keys())
+        for connection in keepalives:
+            if connection.expires_at is not None and now > connection.expires_at:
+                self.keepalive_connections.remove(connection)
+                self.max_connections.release()
+                await connection.close()
+
     async def send(
         self,
         request: Request,
@@ -125,6 +143,7 @@ class ConnectionPool(Dispatcher):
         cert: CertTypes = None,
         timeout: Timeout = None,
     ) -> Response:
+        await self.check_keepalive_expiry()
         connection = await self.acquire_connection(
             origin=request.url.origin, timeout=timeout
         )
@@ -180,6 +199,8 @@ class ConnectionPool(Dispatcher):
             self.max_connections.release()
             await connection.close()
         else:
+            now = self.backend.time()
+            connection.expires_at = now + self.KEEP_ALIVE_EXPIRY
             self.active_connections.remove(connection)
             self.keepalive_connections.add(connection)
 
diff --git a/httpx/dispatch/http2.py b/httpx/dispatch/http2.py
index e0ddf23..43ea8df 100644
--- a/httpx/dispatch/http2.py
+++ b/httpx/dispatch/http2.py
@@ -96,6 +96,7 @@ class HTTP2Connection(OpenConnection):
         del self.state.local_settings[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL]
 
         self.state.initiate_connection()
+        self.state.increment_flow_control_window(2 ** 24)
         data_to_send = self.state.data_to_send()
         await self.socket.write(data_to_send, timeout)
 
@@ -165,6 +166,7 @@ class HTTP2Connection(OpenConnection):
         timeout: Timeout,
     ) -> None:
         self.state.send_headers(stream_id, headers)
+        self.state.increment_flow_control_window(2 ** 24, stream_id=stream_id)
         data_to_send = self.state.data_to_send()
         await self.socket.write(data_to_send, timeout)
 


Test patch:
diff --git a/tests/dispatch/test_connection_pools.py b/tests/dispatch/test_connection_pools.py
index f10b16a..447b8d6 100644
--- a/tests/dispatch/test_connection_pools.py
+++ b/tests/dispatch/test_connection_pools.py
@@ -18,6 +18,37 @@ async def test_keepalive_connections(server, backend):
         assert len(http.keepalive_connections) == 1
 
 
+async def test_keepalive_timeout(server, backend):
+    """
+    Keep-alive connections should timeout.
+    """
+    async with ConnectionPool() as http:
+        response = await http.request("GET", server.url)
+        await response.read()
+        assert len(http.active_connections) == 0
+        assert len(http.keepalive_connections) == 1
+
+        http.next_keepalive_check = 0.0
+        await http.check_keepalive_expiry()
+
+        assert len(http.active_connections) == 0
+        assert len(http.keepalive_connections) == 1
+
+    async with ConnectionPool() as http:
+        http.KEEP_ALIVE_EXPIRY = 0.0
+
+        response = await http.request("GET", server.url)
+        await response.read()
+        assert len(http.active_connections) == 0
+        assert len(http.keepalive_connections) == 1
+
+        http.next_keepalive_check = 0.0
+        await http.check_keepalive_expiry()
+
+        assert len(http.active_connections) == 0
+        assert len(http.keepalive_connections) == 0
+
+
 async def test_differing_connection_keys(server, backend):
     """
     Connections to differing connection keys should result in multiple connections.
diff --git a/tests/dispatch/utils.py b/tests/dispatch/utils.py
index 0056528..3d24ac7 100644
--- a/tests/dispatch/utils.py
+++ b/tests/dispatch/utils.py
@@ -41,7 +41,6 @@ class MockHTTP2Server(BaseSocketStream):
         self.requests = {}
         self.close_connection = False
         self.return_data = {}
-        self.returning = {}
         self.settings_changed = []
 
     # Socket stream interface
@@ -75,14 +74,6 @@ class MockHTTP2Server(BaseSocketStream):
                     self.buffer += self.conn.data_to_send()
             elif isinstance(event, h2.events.StreamEnded):
                 self.stream_complete(event.stream_id)
-            elif isinstance(event, h2.events.WindowUpdated):
-                if event.stream_id == 0:
-                    for key, value in self.returning.items():
-                        if value:
-                            self.send_return_data(key)
-                # This will throw an error if the event is for a not-yet created stream
-                elif self.returning[event.stream_id]:
-                    self.send_return_data(event.stream_id)
             elif isinstance(event, h2.events.RemoteSettingsChanged):
                 self.settings_changed.append(event)
 
@@ -141,7 +132,6 @@ class MockHTTP2Server(BaseSocketStream):
         self.conn.send_headers(stream_id, response_headers)
         self.buffer += self.conn.data_to_send()
         self.return_data[stream_id] = response.content
-        self.returning[stream_id] = True
         self.send_return_data(stream_id)
 
     def send_return_data(self, stream_id):
@@ -161,7 +151,6 @@ class MockHTTP2Server(BaseSocketStream):
                 )
                 self.conn.send_data(stream_id, chunk)
                 self.buffer += self.conn.data_to_send()
-        self.returning[stream_id] = False
         self.conn.end_stream(stream_id)
         self.buffer += self.conn.data_to_send()