Update AsyncSemaphore to accept initial_value parameter

[charxhit]

I am currently working on an implementation that allows httpx to expose lower-level connection details in the TLS handshake and initial HTTP/2 settings so that they can be configured by users right from the top-level API. I have most of it down, however, I ran into an issue when changing the MAX_CONCURRENT_STREAMS to different values (or omitting it from the SETTINGS frame altogether, as RFC does not mandate its inclusion) that has to do with how httpcore interacts with this setting and the AsyncSemaphore it creates to manage and limit maximum streams.

Currently, httpcore provides a default value of 100 for MAXIMUM_CONCURRENT_STREAMS inside method httpcore._async.http2.AsyncHTTP2Connection._send_connection_init. If you monkey patch it to exclude that setting from being passed, so it isn't included in the initial SETTINGS frame (as clients like Firefox do), then all requests will hang indefinitely.
This is because, in the absence of a user-defined value of this setting, the h2_state.local_settings will return a value of 2**32 when asked for the max_concurrent_streams to denote that the setting is unbounded. This value is passed to the AsyncSemaphore class (from inside AsyncHTTP2Connection.handle_async_request) which uses it to denote both, the max_value and the initial_value for the semaphore. Since the initial value of max_streams is 1, this results in 2**32 - 1 calls to the semaphore's acquire() function to synchronize the state of the semaphore with the streams available. This obviously takes quite a while to happen.

The fix here is simple, allow AsyncSemaphore to accept param initial_value, and pass this param with the value of self.max_streams when creating the semaphore, so that no calls to acquire are required at all. Given this, I really don't understand the reasoning behind initializing the AsyncSemaphore with the same initial and max values. Even with the default value of 100, this logic unnecessarily requires 99 extra calls to acquire() in a frequently accessed method. Am I missing something here or was this a simple oversight? I would be happy to submit a PR to address this.

Here is the reproducible example:

import httpcore
import httpx
import trio

async def patch(self, request):
    """
    Same as the send_connection_init unction accept it omits the MAX_CONCURRENT_STREAMS setting
    """

    initial_values = {
        h2.settings.SettingCodes.ENABLE_PUSH: 0,
        # We omit the MAX_CONCURRENT_STREAMS, which should be perfectly legal to do
        # h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 100,
        h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 65536}

    self._h2_state.local_settings = h2.settings.Settings(
        client=True,
        initial_values=initial_values,
    )

    self._h2_state.initiate_connection()
    self._h2_state.increment_flow_control_window(2**24)

    await self._write_outgoing_data(request)

httpcore._async.http2.AsyncHTTP2Connection._send_connection_init = patch

async def main():
    async with httpx.AsyncClient(http2=True) as client:
        r = await client.get("https://example.com")  # Will never complete (or takes way too long at the very least)
        print('recvd!')

trio.run(main)

[charxhit]

This is also observable with the synchronous API (httpcore/_sync/http2.py:94-102), which creates a semaphore using self._max_streams_semaphore = Semaphore(local_settings_max_streams) and then proceeds to call acquire() in a loop till the count reaches the value of self.max_streams. If that is the case, then why not create the semaphore with the initial value in the first place:
Semaphore(self.max_streams). By doing this, we can skip the redundant calls to acquire()

[tomchristie]

Thanks for your scrutiny here. That seems to me like a valid refactoring / efficiency improvement(?)

Given this, I really don't understand the reasoning behind initializing the AsyncSemaphore with the same initial and max values.

The context history that we have available for this is #652 - if that helps.

Am I missing something here or was this a simple oversight?

I'm not sure. I'd suggest a draft PR demonstrating the refactoring here.

[dunkmann00]

I agree with the complaint that the way the semaphore is setup is unnecessary (I made that PR #652). I think I misunderstood how the semaphore works. I thought the initial value was something that you couldn't increase (i.e. couldn't release it more than that). This is wrong, so this code could be improved to be more efficient.

[dunkmann00]

However, after looking at the actual spec, I'm not sure what is the "correct" way to init the setting given how httpcore works. According to the spec, initially there should be no limit. So I don't know what you would initialize the semaphore with in that case.

And further still, the max-concurrent-streams setting is not a "lower of the two" thing:

. . .clients specify the maximum number of concurrent streams the server can initiate, and servers specify the maximum number of concurrent streams the client can initiate.

I'm not sure if httpcore handles streams initiated by the server? But if it does, or wants to at some point, that would mean you would have to track both client initiated streams and server initiated streams separately (I guess each with their own semaphore perhaps?) and make sure each limit is followed.

I don't know if it would be better for me to put this in a new discussion, but if you think I should @tomchristie, I can do that.

[HMaker]

@charxhit you found solution to that? that issue still happens with latest httpcore. I am also trying to customize H2Connection by changing the initial connection settings. It's stuck in a big loop there

httpcore/httpcore/_sync/http2.py

Lines 115 to 125 in a173552

	# 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()

local_settings_max_streams is 4294967297 because MAX_CONCURRENT_STREAMS was supressed.

[HMaker]

A subclass of Settings changing the default MAX_CONCURRENT_STREAMS seems to be enough

class MyH2Settings(Settings):
    
    @property
    def max_concurrent_streams(self):
        return self.get(SettingCodes.MAX_CONCURRENT_STREAMS, 1)

Need to monkey patch the H2Connection constructor to use that custom Settings subclass.

[HMaker]

@tomchristie is it OK to set the default client-side value of MAX_CONCURRENT_STREAMS to 1? I think that value will be changed later by the server.

[charxhit]

Hey, after digging through this again, it seems like I decided to monkey patch the httpcore._synchronization.AsyncSemaphore class, specifically the setup function to use an initial value of 1 when creating the semaphore.