Race in sync http2 leads to stream waiting indefinitely (or until timeout) for window update frame when already received

[alvin-yuan]

I believe I've identified a race condition in httpcore/_sync/http2.py

It has to do with this comment in _receive_events:

httpcore/httpcore/_sync/http2.py

Line 354 in 31a4a56

# This conditional is a bit icky. We don't want to block reading if we've

From what I gather, it talks about how not self._events.get(stream_id) is the loop condition of _receive_stream_event and how it also needs to be checked within the read lock. The problem that I believe exists is that the loop condition for _wait_for_outgoing_flow (the other place that calls _receive_events) is not also checked within the read lock in some form.
The comment also goes on to describe how _wait_for_outgoing_flow wants to block here (on the _read_incoming_data call) until flow is available, but I believe there's a race where you can end up blocked on _read_incoming_data but you actually already have flow available again due to the order in which things are checked (i.e. we don't check again after acquiring the read lock).

Consequently, something like the following I think can happen (at least I'm under the impression that I ran into this):

• some stream A sends data until there is no remaining outgoing flow.
• some stream B acquires read lock and blocks on _read_incoming_data.
• stream A blocks on read lock in _wait_for_outgoing_flow's call to _receive_events.
• stream B processes incoming data including window update frame, releases lock.
• stream A acquires lock and immediately blocks on _read_incoming_data. Presumably the intention here is that it is waiting to hear a window update frame so it can send more data.
• In practice, that window update frame was already received and the loop condition of _wait_for_outgoing_flow is already satisfied, but the logic blocks anyway.
• If no other event happens to be received, stream A will timeout here with a ReadTimeout.
• If some other event happens to be received, stream A proceeds without issue, though having experienced an unnecessary delay in sending data.
• If no other event happens to be received and the timeout is sufficiently long, the other end may also terminate in some fashion due to unexpected behavior on our end (we've stopped streaming data for seemingly no reason).

For some additional context, my company is using httpcore (and httpx) to integrate with Alexa Voice Service. That involves what I'm guessing is a bit of an unconventional setup where we will stream (audio) data to amazon on one stream, but always have another stream open where we can receive events pushed to us from amazon's side. If I mess with httpcore._sync.http2.HTTP2Connection.CONFIG.logger to get it to show me its logs, I'm able to see the chain of events described above every once in a while. We're actually on a bit of an older version but from what I can tell, the code around this issue has not changed so I think it's still present on the current master.

I have not checked async or any other modules (we only happen to be using sync http2).

[tomchristie]

Okay, so our first task is reproducibility.

Two possible lines of attack here:

• We've got a fairly decent set of test cases for our HTTP/2 handling. Given that you've outlined this fairly clearly, perhaps we're able to replicate it at that level.
• A simple-as-possible example replication against a live URL. (Or an example of this running against a localhost server)

Thoughts?

[alvin-yuan]

Hey Tom,

Sorry, I've somewhat moved onto other tasks. If you're expecting a repro from me, I can maybe try to squeeze in some time in the next week or so to investigate. In terms of options, I'm not that familiar with the existing test framework, but taking a cursory glance suggests that it might be doable. The part that wasn't immediately clear to me is how easy it would be to have two or more streams on the same connection and control exactly when they might call _read_incoming_data and control the incoming data each stream ends up seeing for each call.

Regarding a live URL, I'm not aware of a great url to test with. I only really have familiarity with how the Alexa server behaves. A custom localhost server definitely should be able to work, but I'm not experienced with how to set one up quickly and finely control its behavior at this layer of abstraction.

[alvin-yuan]

Alright, this is as good of a repro as I could get right now. This is pretty consistent for me (and I also included some notes about what circumstances would avoid the problem). Running this should run into a ReadTimeout with a stack trace. Unfortunately, I couldn't get the issue to happen with two real streams and needed to simulate what one of the streams would do by directly calling into some private methods. But hopefully this suffices and is still realistic enough. Please excuse the ugliness of some of the code.

import httpcore._sync.http2
import httpx
import logging
import sys
import threading
import gc

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
log = logging.getLogger(__name__)

URL = "https://example.com"


def iter_content():
  yield b"A" # not sure why this first yield seems to be important for repro
  # Assuming window starts at ~65535
  # Assuming see remote settings change of window size to ~1048576
  # Need to exceed window size
  yield b"A" * 1548000

def make_get_request(client):
  request = client.build_request(
      method="GET",
      url=URL,
  )
  client.send(request)

def make_post_request(client):
  request = client.build_request(
      method="POST",
      url=URL,
      content=iter_content(),
      extensions={
          "trace": lambda event_name, info: log.info("TRACE POST %s: %s", event_name, info)
      },
  )
  client.send(request, stream=True)
  print("\nDONE WITH POST REQUEST\n")

def main():
  httpcore._sync.http2.HTTP2Connection.CONFIG.logger = log
  log.trace = lambda self, *vargs, **kwargs: log.info("TRACE-H2 %s %s", vargs, kwargs)

  client = httpx.Client(
      base_url=URL,
      http1=False,
      http2=True,
  )
  make_get_request(client) # to create HTTP2Connection object
  http2_connection = \
      [o for o in gc.get_objects() if isinstance(o, httpcore._sync.http2.HTTP2Connection)][0]

  class LockWrapper:
    def __init__(self, lock) -> None:
        self._lock = lock

    def __enter__(self) -> "Lock":
        print("\nPRE LOCK ATTEMPT ACQUIRE\n")
        self._lock.__enter__()
        return self._lock

    def __exit__(self, *args, **kwargs) -> None:
        self._lock.__exit__(*args, **kwargs)
        print("\nRELEASE LOCK\n")

  http2_connection._read_lock = LockWrapper(http2_connection._read_lock)


  class FakeRequest:
    def __init__(self) -> None:
      self.extensions = {}


  print("\nDONE SETTING UP\n")

  thread = threading.Thread(target=make_post_request, args=(client,))
  thread.start()
  # Simulate some other stream reading incoming data. We need to get the post request to block
  # on the read lock, then have the "other streams" process all of the window update frame events
  # and then not receive any subsequent events before the read timeout. Getting this to happen with
  # real stream requests is difficult due to needing to see a very specific order of execution and
  # no excess incoming events.
  with http2_connection._read_lock:
    print("\nHOLDING READ LOCK\n")
    # Read the window update frames, one for the entire connection (stream_id 0) and one for the
    # post request stream.
    http2_connection._read_incoming_data(FakeRequest())
    # Commenting one of these results in success
    http2_connection._read_incoming_data(FakeRequest())
    print("\nRELEASE READ LOCK\n")

  # Uncommenting the line below results in success, presumably because it results in receiving
  # more events which will unblock the post request's _read_incoming_data call, even though that
  # event is not relevant to the post request.

  # make_get_request(client)

  thread.join()

if __name__ == "__main__":
  main()
  print("\nDone")