With http2, responses aren't returned until previous requests have returned #3278
Unanswered
nmlorg
asked this question in
Potential Issue
Replies: 2 comments 5 replies
-
time.sleep(1)
delayed_fetch(6) # a: 1 - 7
time.sleep(1)
delayed_fetch(1) # b: 2 - 3
time.sleep(2)
delayed_fetch(6) # c: 4 - 10
time.sleep(1)
delayed_fetch(1) # d: 5 - 6
time.sleep(3)
delayed_fetch(1) # e: 8 - 9It looks the thread for request |
Beta Was this translation helpful? Give feedback.
0 replies
-
|
I think it's a race in I managed to get a test run to complete according to the expected timeline only after I added additional logging. I then added even more logging and managed to get it to flip between the two behaviors intermittently: with Trace('_receive_events _read_lock outside', logger, request, {'stream_id': stream_id}):
with self._read_lock:
with Trace('_receive_events _read_lock inside', logger, request, {'stream_id': stream_id}):
if self._connection_terminated is not None: |
Beta Was this translation helpful? Give feedback.
5 replies
-
|
It feels like this is what def __init__():
self._read_lock = threading.Lock()
self._read_condition = threading.Condition()
def _receive_stream_event(stream_id):
while not self._events.get(stream_id):
self._receive_events(stream_id)
def _receive_events(stream_id):
with self._read_condition:
if self._events.get(stream_id): # Another thread received data for me while I was blocked waiting for _read_condition's lock.
return
if self._read_lock.acquire(blocking=False): # I am now responsible for reading from the connection (and waking other threads up).
self._read_condition.release() # Let other threads get through to the wait below.
event = self._read_incoming_data() # Block until data is available.
self._read_condition.acquire() # Block any other threads from checking self._events or trying to acquire _read_lock until I dispose of this event.
self._read_condition.notify_all() # Schedule all threads in wait (below) to wake up as soon as I release _read_condition.
self._read_lock.release() # Allow another thread to take over as reader once I release _read_condition.
self._events[event.stream_id].append(event)
else: # Another thread is already responsible for reading from the connection (and waking me up).
#self._read_condition.wait_for(lambda: self._events.get(stream_id)) # Wake up iff data is waiting for me.
self._read_condition.wait() # Wake up even if no data is waiting for me so I can become the next reader if the current one is finished.The goal would be to make |
Beta Was this translation helpful? Give feedback.
-
|
It's possible that choosing to call self._read_condition.notify_all() # Schedule all threads in wait (below) to wake up as soon as I release _read_condition.becoming: if event.stream_id == stream_id:
self._read_condition.notify() # Schedule just one random thread in wait (below) to wake up (to become the new reader) as soon as I release _read_condition.
else:
self._read_condition.notify_all() # Schedule all threads in wait (below) to wake up (so the thread waiting for this event can return it) as soon as I release _read_condition. |
Beta Was this translation helpful? Give feedback.
-
for _ in range(5):
time.sleep(2)
delayed_fetch(9) |
Beta Was this translation helpful? Give feedback.
-
|
I just noticed For what it's worth, this does appear to affect for _ in range(5):
await asyncio.sleep(2)
delayed_fetch(9)test.py: import logging
import threading
import time
import httpcore
logging.basicConfig(format='%(relativeCreated)5i <%(threadName)s> %(filename)s:%(lineno)s] %(message)s',
level=logging.DEBUG)
logging.getLogger('hpack').setLevel(logging.INFO)
pool = httpcore.ConnectionPool(http2=True)
logging.info('pool=%s', pool)
def delayed_fetch(delay):
url = f'https://httpbin.org/delay/{delay}'
def worker():
logging.info('Starting fetch of %s.', url)
start = time.time()
response = pool.request('GET', url)
end = time.time()
logging.info('Finished fetch of %s (%s) in %.03fs.', url, response, end - start)
threading.Thread(target=worker, name=f"Thread {chr(ord('A') + threading.active_count() - 1) * 3}").start()
for _ in range(5):
time.sleep(2)
delayed_fetch(9)asynctest.py: import asyncio
import logging
import time
import httpcore
logging.basicConfig(format='%(relativeCreated)5i <%(threadName)s> %(filename)s:%(lineno)s] %(message)s',
level=logging.DEBUG)
logging.getLogger('hpack').setLevel(logging.INFO)
pool = httpcore.AsyncConnectionPool(http2=True)
logging.info('pool=%s', pool)
async def main():
async with asyncio.TaskGroup() as tg:
def delayed_fetch(delay):
url = f'https://httpbin.org/delay/{delay}'
async def worker():
logging.info('Starting fetch of %s.', url)
start = time.time()
response = await pool.request('GET', url)
end = time.time()
logging.info('Finished fetch of %s (%s) in %.03fs.', url, response, end - start)
tg.create_task(worker())
for _ in range(5):
await asyncio.sleep(2)
delayed_fetch(9)
asyncio.run(main()) |
Beta Was this translation helpful? Give feedback.
-
|
My first approach was to use a fixed single I am a little outside of my comfort zone (especially with the async stuff), but I can try to build a proof of concept of the per-request |
Beta Was this translation helpful? Give feedback.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
Hi! I think this is showing that, when http2 is enabled, when multiple threads send requests [to the same host?] through a single
Client, the thread that first used theClientis receiving all responses and holding them until its request's response is received. (When http2 is disabled, each request gets its own TCP session, and requests are processed immediately.)I have a mix of long polls and mutating RPCs to a single host, so what I'm seeing is that RPCs that are triggered by timers (instead of because a long poll just finished) are getting backed up until the current long poll ends. (And I presume multiple long polls will also back each other up.)
Beta Was this translation helpful? Give feedback.
All reactions