AsyncClient does not support being used within multiple event loops #2058

Activity

1. lovelydinosaur commented on Feb 1, 2022

   

   lovelydinosaur

   on Feb 1, 2022

   Member

   More actions

   Um... yes certainly not an expected usage, or something that I'd expect to work.

   It's not so much about "thread safety" (which I'd consider to be to do with locking during concurrency), as that it doesn't support a single client instance being used across multiple different event loops.

   I guess you might want to take a look at if the behaviour here is any different with:

   • Using aiohttp instead of httpx.
   • When using trio instead of httpx.

   But generally I wouldn't expect it to work, and I wouldn't expect us to support it.

   but there are scenarios where both are required

   Perhaps you can explain more why you're bumping into this?
2. lovelydinosaur

   closed this as completedon Feb 1, 2022
3. lovelydinosaur

   reopened this on Feb 1, 2022
4. lovelydinosaur

   changed the title [-]AsyncClient is not thread safe (RuntimeError: Event loop is closed)[/-] [+]AsyncClient does not support being used within multiple event loops[/+] on Feb 1, 2022
5. lovelydinosaur

   closed this as completedon Feb 4, 2022
6. bede commented on May 31, 2022

   

   bede

   on May 31, 2022

   More actions

   Like the OP I am receiving ssl.SSLWantReadError: The operation did not complete (read) (_ssl.c:2548)

   The following code raises this intermittently, and having read this issue I wondered if there is problem with the pattern I am using. download_single_async() is basically just awaiting a client.get() call

       async with httpx.AsyncClient(transport=transport, limits=limits, timeout=10) as client:
           guids_types_urls = {}
           for guid in guids:
               for file_type in file_types:
                   guids_types_urls[(guid, file_type)] = f"{endpoint}/{guid}/{file_type}"
           tasks = [
               download_single_async(
                   client,
                   guid,
                   file_type,
                   url,
                   headers,
                   out_dir,
                   guids_names[guid] if guids_names else None,
               )
               for (guid, file_type), url in guids_types_urls.items()
           ]
           return [
               await f
               for f in tqdm(
                   asyncio.as_completed(tasks),
                   desc=f"Downloading {len(tasks)} files for {len(guids)} samples",
                   bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt}",
                   total=len(tasks),
               )
           ]
7. provinzkraut commented on Sep 28, 2022

   

   provinzkraut

   on Sep 28, 2022

   Last edited by provinzkraut

   More actions

   Um... yes certainly not an expected usage, or something that I'd expect to work.

   I'm running into the same issue and I'd argue that it's actually a fairly common usage. Maybe the original example wasn't that great but I think I can illustrate the issue a bit more clearly:

   import asyncio
   import httpx
   
   client = httpx.AsyncClient()
   
   loop = asyncio.new_event_loop()
   loop.run_until_complete(client.get("https://example.org"))
   loop.close()
   
   loop = asyncio.new_event_loop()
   loop.run_until_complete(client.get("https://example.org"))
   loop.close()

   Having two running event loops simultaneously might not be common, but creating and closing event loops most certainly is, and it causes the same issue.

   It's not so much about "thread safety" (which I'd consider to be to do with locking during concurrency), as that it doesn't support a single client instance being used across multiple different event loops.

   I'd say that this is not thread safe, as you cannot pass an AsyncClient instance into another thread and make requests there without either causing a RuntimeError there or in the main thread. Not being able to be used across multiple event loops is what causes AsyncClient to not be thread safe, as that is required when using threads.

   import asyncio
   import httpx
   from concurrent.futures import ThreadPoolExecutor, wait
   
   def request_in_thread():
       loop = asyncio.new_event_loop()
       loop.run_until_complete(http_client.get("https://example.org"))
       loop.close()
   
   
   with ThreadPoolExecutor() as executor:
       future = executor.submit(request_in_thread)  # Now the client is unusable in the main thread
       wait([future])
   
   
   loop = asyncio.new_event_loop()
   loop.run_until_complete(http_client.get("https://example.org"))
   loop.close()

   The only way around this currently is to either manually reset the pool (which is hacky) or to create a new client instance for each request (which is not recommended).

   Possible fix

   If a connection in the pool "sticks" to an event loop, don't try to reuse that connection if the event loop it's associated with is closed.

   Alternatives

   If discarding connections with closed associated event loops is out of scope or simply not a desirable feature (I'm not sure what the drawbacks of that would be though), it would be nice to at least have an option to either reset the pool manually, or to opt out of it (ie have a non-pool transport).

   Workarounds

   Set keepalive_expiry=0 by instantiating AsyncClient like this:

   import httpx
   client = httpx.AsyncClient(limits=httpx.Limits(keepalive_expiry=0))

   More hacky workaround that still makes use of connections pools within the same event loop:

   from httpx._transports.default import AsyncHTTPTransport
   from httpx import AsyncClient
   
   class LoopSafeAsyncHTTPTransport(AsyncHTTPTransport):
       def __init__(self, *args, **kwargs):
           super().__init__(*args, **kwargs)
           self._active_loop = None
   
       async def handle_async_request(self, request):
           current_loop = asyncio.get_event_loop()
           if self._active_loop and current_loop is not self._active_loop:
               self._pool._pool = []  # this should / could be replace with a proper re-initialisation
           self._active_loop = current_loop
           return await super().handle_async_request(request)
   
   client = AsyncClient(transport=LoopSafeAsyncHTTPTransport())

   The drawback here is that connection pooling won't be used for any request, but you still get the overhead of creating and managing the pool.
8. Zac-HD commented on Feb 23, 2024

   

   Zac-HD

   on Feb 23, 2024

   More actions

   FYI this is leading Stripe to disable connection pooling, because of user friction where unittest creates a separate loop for each test.

   If creating a separate pool per-event-loop when running on asyncio is feasible, I think that would allow connection pooling to stay default-enabled and lead to an overall better user experience.

   👍React with 👍3Reacted by Guillaume Gelin, Richard Marmorstein and Nick Boultbee