Adjust max streams when remote settings change (#652)

* Adjust max streams when remote settings change

If the remote server sends a settings update for
`max_concurrent_streams`, check and potentially update the max number
of streams we will use for a connection.

* Update sync version of http2

* Add aiter_stream sub to iter_stream

* Add test for remote max streams update

Improve testing for remote max streams update

This new test checks whether a change in the max streams setting is
properly handled. The max streams value is changed twice throughout the
mock stream. The first update increases the value while the second
update decreases it.

---------

Co-authored-by: Tom Christie <tom@tomchristie.com>

Author: dunkmann00

httpcore/_async/http2.py

@@ -88,8 +88,18 @@ async def handle_async_request(self, request: Request) -> Response:
                 async with Trace("http2.send_connection_init", request, kwargs):
                     await self._send_connection_init(**kwargs)
                 self._sent_connection_init = True
-                max_streams = self._h2_state.local_settings.max_concurrent_streams
-                self._max_streams_semaphore = AsyncSemaphore(max_streams)
+
+                # Initially start with just 1 until the remote server provides
+                # its max_concurrent_streams value
+                self._max_streams = 1
+
+                local_settings_max_streams = (
+                    self._h2_state.local_settings.max_concurrent_streams
+                )
+                self._max_streams_semaphore = AsyncSemaphore(local_settings_max_streams)
+
+                for _ in range(local_settings_max_streams - self._max_streams):
+                    await self._max_streams_semaphore.acquire()
 
         await self._max_streams_semaphore.acquire()
 
@@ -280,6 +290,13 @@ async def _receive_events(
             if stream_id is None or not self._events.get(stream_id):
                 events = await self._read_incoming_data(request)
                 for event in events:
+                    if isinstance(event, h2.events.RemoteSettingsChanged):
+                        async with Trace(
+                            "http2.receive_remote_settings", request
+                        ) as trace:
+                            await self._receive_remote_settings_change(event)
+                            trace.return_value = event
+
                     event_stream_id = getattr(event, "stream_id", 0)
 
                     # The ConnectionTerminatedEvent applies to the entire connection,
@@ -293,6 +310,23 @@ async def _receive_events(
 
         await self._write_outgoing_data(request)
 
+    async def _receive_remote_settings_change(self, event: h2.events.Event) -> None:
+        max_concurrent_streams = event.changed_settings.get(
+            h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS
+        )
+        if max_concurrent_streams:
+            new_max_streams = min(
+                max_concurrent_streams.new_value,
+                self._h2_state.local_settings.max_concurrent_streams,
+            )
+            if new_max_streams and new_max_streams != self._max_streams:
+                while new_max_streams > self._max_streams:
+                    await self._max_streams_semaphore.release()
+                    self._max_streams += 1
+                while new_max_streams < self._max_streams:
+                    await self._max_streams_semaphore.acquire()
+                    self._max_streams -= 1
+
     async def _response_closed(self, stream_id: int) -> None:
         await self._max_streams_semaphore.release()
         del self._events[stream_id]

httpcore/_sync/http2.py

@@ -88,8 +88,18 @@ def handle_request(self, request: Request) -> Response:
                 with Trace("http2.send_connection_init", request, kwargs):
                     self._send_connection_init(**kwargs)
                 self._sent_connection_init = True
-                max_streams = self._h2_state.local_settings.max_concurrent_streams
-                self._max_streams_semaphore = Semaphore(max_streams)
+
+                # Initially start with just 1 until the remote server provides
+                # its max_concurrent_streams value
+                self._max_streams = 1
+
+                local_settings_max_streams = (
+                    self._h2_state.local_settings.max_concurrent_streams
+                )
+                self._max_streams_semaphore = Semaphore(local_settings_max_streams)
+
+                for _ in range(local_settings_max_streams - self._max_streams):
+                    self._max_streams_semaphore.acquire()
 
         self._max_streams_semaphore.acquire()
 
@@ -280,6 +290,13 @@ def _receive_events(
             if stream_id is None or not self._events.get(stream_id):
                 events = self._read_incoming_data(request)
                 for event in events:
+                    if isinstance(event, h2.events.RemoteSettingsChanged):
+                        with Trace(
+                            "http2.receive_remote_settings", request
+                        ) as trace:
+                            self._receive_remote_settings_change(event)
+                            trace.return_value = event
+
                     event_stream_id = getattr(event, "stream_id", 0)
 
                     # The ConnectionTerminatedEvent applies to the entire connection,
@@ -293,6 +310,23 @@ def _receive_events(
 
         self._write_outgoing_data(request)
 
+    def _receive_remote_settings_change(self, event: h2.events.Event) -> None:
+        max_concurrent_streams = event.changed_settings.get(
+            h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS
+        )
+        if max_concurrent_streams:
+            new_max_streams = min(
+                max_concurrent_streams.new_value,
+                self._h2_state.local_settings.max_concurrent_streams,
+            )
+            if new_max_streams and new_max_streams != self._max_streams:
+                while new_max_streams > self._max_streams:
+                    self._max_streams_semaphore.release()
+                    self._max_streams += 1
+                while new_max_streams < self._max_streams:
+                    self._max_streams_semaphore.acquire()
+                    self._max_streams -= 1
+
     def _response_closed(self, stream_id: int) -> None:
         self._max_streams_semaphore.release()
         del self._events[stream_id]

tests/_async/test_http2.py

@@ -294,3 +294,55 @@ async def test_http2_request_to_incorrect_origin():
     async with AsyncHTTP2Connection(origin=origin, stream=stream) as conn:
         with pytest.raises(RuntimeError):
             await conn.request("GET", "https://other.com/")
+
+
+@pytest.mark.anyio
+async def test_http2_remote_max_streams_update():
+    """
+    If the remote server updates the maximum concurrent streams value, we should
+    be adjusting how many streams we will allow.
+    """
+    origin = Origin(b"https", b"example.com", 443)
+    stream = AsyncMockStream(
+        [
+            hyperframe.frame.SettingsFrame(
+                settings={hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 1000}
+            ).serialize(),
+            hyperframe.frame.HeadersFrame(
+                stream_id=1,
+                data=hpack.Encoder().encode(
+                    [
+                        (b":status", b"200"),
+                        (b"content-type", b"plain/text"),
+                    ]
+                ),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(stream_id=1, data=b"Hello, world!").serialize(),
+            hyperframe.frame.SettingsFrame(
+                settings={hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 50}
+            ).serialize(),
+            hyperframe.frame.DataFrame(
+                stream_id=1, data=b"Hello, world...again!", flags=["END_STREAM"]
+            ).serialize(),
+        ]
+    )
+    async with AsyncHTTP2Connection(origin=origin, stream=stream) as conn:
+        async with conn.stream("GET", "https://example.com/") as response:
+            i = 0
+            async for chunk in response.aiter_stream():
+                if i == 0:
+                    assert chunk == b"Hello, world!"
+                    assert conn._h2_state.remote_settings.max_concurrent_streams == 1000
+                    assert conn._max_streams == min(
+                        conn._h2_state.remote_settings.max_concurrent_streams,
+                        conn._h2_state.local_settings.max_concurrent_streams,
+                    )
+                elif i == 1:
+                    assert chunk == b"Hello, world...again!"
+                    assert conn._h2_state.remote_settings.max_concurrent_streams == 50
+                    assert conn._max_streams == min(
+                        conn._h2_state.remote_settings.max_concurrent_streams,
+                        conn._h2_state.local_settings.max_concurrent_streams,
+                    )
+                i += 1

tests/_sync/test_http2.py

@@ -294,3 +294,55 @@ def test_http2_request_to_incorrect_origin():
     with HTTP2Connection(origin=origin, stream=stream) as conn:
         with pytest.raises(RuntimeError):
             conn.request("GET", "https://other.com/")
+
+
+
+def test_http2_remote_max_streams_update():
+    """
+    If the remote server updates the maximum concurrent streams value, we should
+    be adjusting how many streams we will allow.
+    """
+    origin = Origin(b"https", b"example.com", 443)
+    stream = MockStream(
+        [
+            hyperframe.frame.SettingsFrame(
+                settings={hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 1000}
+            ).serialize(),
+            hyperframe.frame.HeadersFrame(
+                stream_id=1,
+                data=hpack.Encoder().encode(
+                    [
+                        (b":status", b"200"),
+                        (b"content-type", b"plain/text"),
+                    ]
+                ),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(stream_id=1, data=b"Hello, world!").serialize(),
+            hyperframe.frame.SettingsFrame(
+                settings={hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 50}
+            ).serialize(),
+            hyperframe.frame.DataFrame(
+                stream_id=1, data=b"Hello, world...again!", flags=["END_STREAM"]
+            ).serialize(),
+        ]
+    )
+    with HTTP2Connection(origin=origin, stream=stream) as conn:
+        with conn.stream("GET", "https://example.com/") as response:
+            i = 0
+            for chunk in response.iter_stream():
+                if i == 0:
+                    assert chunk == b"Hello, world!"
+                    assert conn._h2_state.remote_settings.max_concurrent_streams == 1000
+                    assert conn._max_streams == min(
+                        conn._h2_state.remote_settings.max_concurrent_streams,
+                        conn._h2_state.local_settings.max_concurrent_streams,
+                    )
+                elif i == 1:
+                    assert chunk == b"Hello, world...again!"
+                    assert conn._h2_state.remote_settings.max_concurrent_streams == 50
+                    assert conn._max_streams == min(
+                        conn._h2_state.remote_settings.max_concurrent_streams,
+                        conn._h2_state.local_settings.max_concurrent_streams,
+                    )
+                i += 1

unasync.py

@@ -18,6 +18,7 @@
     ('aclose', 'close'),
     ('aclose_func', 'close_func'),
     ('aiterator', 'iterator'),
+    ('aiter_stream', 'iter_stream'),
     ('aread', 'read'),
     ('asynccontextmanager', 'contextmanager'),
     ('__aenter__', '__enter__'),