One aiohttp ClientSession per thread?

Question

In the documentation for aiohttp.ClientSession I have read that you should, ideally, create one ClientSession per application. I suspect that that should read something like "create one ClientSession per thread" (I'm in python 3x).

import aiohttp
import asyncio

async def get_it_async(url: str, session):
    async with session.get(url) as resp:
        print(resp.status)
        print(len(await resp.text()))

def run_it(fn, *args, **kwargs):
    # New thread - get the loop.
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    assert not loop.is_running()
    return loop.run_until_complete(loop.create_task(fn(*args, **kwargs)))

        
from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(2)

client = aiohttp.ClientSession()
await asyncio.wrap_future(pool.submit(run_it, get_it_async, 'httyp://nytimes.com', client))

Gives me:

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
 in async-def-wrapper()

~\appdata\Local\Programs\Python\Python37\lib\concurrent\futures	hread.py in run(self)
     55 
     56         try:
---> 57             result = self.fn(*self.args, **self.kwargs)
     58         except BaseException as exc:
     59             self.future.set_exception(exc)

 in run_it(fn, *args, **kwargs)
     15 
     16 
---> 17 from concurrent.futures import ThreadPoolExecutor
     18 pool = ThreadPoolExecutor(2)
     19 

~\appdata\Local\Programs\Python\Python37\lib\asyncio\base_events.py in run_until_complete(self, future)
    585             raise RuntimeError('Event loop stopped before Future completed.')
    586 
--> 587         return future.result()
    588 
    589     def stop(self):

 in get_it_async(url, session)
      6         print(resp.status)
      7         print(len(await resp.text()))
----> 8 
      9 def run_it(fn, *args, **kwargs):
     10     # New thread - get the loop.

i:\gwatts\code\calratio2019\calms\.venv\lib\site-packages\aiohttp\client.py in __aenter__(self)
   1010 
   1011     async def __aenter__(self) -> _RetType:
-> 1012         self._resp = await self._coro
   1013         return self._resp
   1014 

i:\gwatts\code\calratio2019\calms\.venv\lib\site-packages\aiohttp\client.py in _request(self, method, str_or_url, params, data, json, cookies, headers, skip_auto_headers, auth, allow_redirects, max_redirects, compress, chunked, expect100, raise_for_status, read_until_eof, proxy, proxy_auth, timeout, verify_ssl, fingerprint, ssl_context, ssl, proxy_headers, trace_request_ctx)
    424         timer = tm.timer()
    425         try:
--> 426             with timer:
    427                 while True:
    428                     url, auth_from_url = strip_auth_from_url(url)

i:\gwatts\code\calratio2019\calms\.venv\lib\site-packages\aiohttp\helpers.py in __enter__(self)
    577 
    578         if task is None:
--> 579             raise RuntimeError('Timeout context manager should be used '
    580                                'inside a task')
    581 

RuntimeError: Timeout context manager should be used inside a task

Is there a bug in my code, or is my assumption correct?

user_1600898 · Answer

I suspect that that should read something like "create one ClientSession per thread" (I'm in python 3x).

That interpretation is incorrect for several reasons. First, the recommendation is meant exactly as written - one ClientSession is typically used per application, where an application is a program that performs a number of independent (but loosely related) downloads. For example, a web crawler would be an application, and so would be a bittorrent client, etc. Multiple applications can run inside a single process, in which case it makes sense to create multiple sessions which are completely independent from each other.

The second problem is that asyncio is single-threaded, so it doesn't make sense to speak of "one session per thread". One of the selling points of asyncio is that you can achieve scalable parallelism that doesn't depend on creating a huge number of OS threads. Your code that creates a thread to run a whole new event just to download something is technically correct, but unnecessary, especially since your top-level code appears to be running inside asyncio to begin with! It's also potentially complex and error-prone because if you ever want to share data between downloads, you will need to deal with inter-thread synchronization issues.

The way to achieve parallelism in asyncio is to use a single event loop in your process, and do everything inside it. For example, taking your get_it_async function as a starting point, you might do this to download three URLs in parallel:

async def main():
    async with aiohttp.ClientSession() as session:
        await asyncio.gather(
            get_it_async('https://nytimes.com', session),
            get_it_async('https://yahoo.com', session),
            get_it_async('https://cnn.com', session),
        )

# or, await main() if you call main from another async function
asyncio.run(main())

Functions like asyncio.wrap_future (and more typically its cousin loop.run_in_executor) should only be used when invoking CPU-bound or legacy blocking code from within an asyncio application. If you have async code and invoke it from async, just await it or use asyncio.gather() and similar to make it run in parallel.

user_13782669 · Answer

As I know, ClientSession() and all nested objects save loop object (e.g. self._loop) while instantiated. And when you try to use it in another thread (and therefore another active event loop), ClientSession internal code cannot find current task for own saved loop and raises this exception.

I think for every new thread (and for every new loop) new ClientSession should be instantiated.

One aiohttp ClientSession per thread?

2 Answers