asyncio + aiohttp ClientConnectorError The semaphore timeout

ChunMa716 0
Viewed 39

I'm scraping URLS that contain JSON, and my code works flawlessly if I dare say. But a problem always pops up between the 20s-21s range with

"aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host xtreme-gacha.com:80 ssl:False [The semaphore timeout period has expired]"

Any suggestions?

ps: The code is reproducible

import asyncio
import sys
from timeit import default_timer
from aiohttp import ClientSession
import json
import aiohttp
from aiolimiter import AsyncLimiter








headers = {
  'accept': 'application/json, text/javascript, */*; q=0.01',
  'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36',
}


def fetch_async(urls):
    start_time = default_timer()

    loop = asyncio.get_event_loop() 
    future = asyncio.ensure_future(fetch_all(urls)) 
    loop.run_until_complete(future) 

    tot_elapsed = default_timer() - start_time
    print('Total time taken : ' , str(tot_elapsed))

async def fetch_all(urls):
    
    tasks = []
    fetch.start_time = dict() 
    async with sem:
        async with ClientSession(timeout=timeout, headers=headers) as session:
            for url in urls:
                task = asyncio.ensure_future(fetch(sem, url, session))
                tasks.append(task) 
            _ = await asyncio.gather(*tasks) 

async def fetch(sem, url, session):
    fetch.start_time[url] = default_timer()
    async with sem:
        async with session.get(url, ssl= False) as response:
            raw = await response.text() # here
            text = raw.encode().decode('utf-8-sig')
            jsonn=  json.loads(text)
            

            elapsed = default_timer() - fetch.start_time[url]
            print(url ,  ' took ',   str(elapsed) , str(sys.getsizeof(jsonn)))




if __name__ == '__main__':
    
    timeout = aiohttp.ClientTimeout(total=5*60, connect=None,
                      sock_connect=None, sock_read=None)
    sem =  AsyncLimiter(max_rate=10, time_period=1)

    urls = ['http://xtreme-gacha.com/bbs/accounts/fresh/40246F'] * 200




    results = fetch_async(urls)

Edit for @Artiom Kozyrev soulation This is the output after the 20s-21s of Succesfully sending the requests

Traceback (most recent call last):
  File "C:\bg\benv\lib\site-packages\aiohttp\connector.py", line 986, in _wrap_create_connection
    return await self._loop.create_connection(*args, **kwargs)  # type: ignore[return-value]  # noqa
  File "C:\Python310\lib\asyncio\base_events.py", line 1064, in create_connection
    raise exceptions[0]
  File "C:\Python310\lib\asyncio\base_events.py", line 1049, in create_connection
    sock = await self._connect_sock(
  File "C:\Python310\lib\asyncio\base_events.py", line 960, in _connect_sock
    await self.sock_connect(sock, address)
  File "C:\Python310\lib\asyncio\proactor_events.py", line 705, in sock_connect
    return await self._proactor.connect(sock, address)
  File "C:\Python310\lib\asyncio\windows_events.py", line 817, in _poll
    value = callback(transferred, key, ov)
  File "C:\Python310\lib\asyncio\windows_events.py", line 604, in finish_connect
    ov.getresult()
OSError: [WinError 121] The semaphore timeout period has expired

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "c:\bg\test3.py", line 58, in <module>
    asyncio.get_event_loop().run_until_complete(main_entry_point())
  File "C:\Python310\lib\asyncio\base_events.py", line 646, in run_until_complete
    return future.result()
  File "c:\bg\test3.py", line 53, in main_entry_point
    await asyncio.gather(*[worker(q, session) for _ in range(WORKER_NUM)])
  File "c:\bg\test3.py", line 38, in worker
    start, end, result = await fetch(session, url, HEADERS)
  File "C:\bg\benv\lib\site-packages\bucketratelimiter\bucket_rate_limiters\asyncio_bucket.py", line 69, in wrapper
    return await self.wrap_operation(f, *args, **kwargs)
  File "C:\bg\benv\lib\site-packages\bucketratelimiter\bucket_rate_limiters\asyncio_bucket.py", line 45, in wrap_operation
    res = await func(*args, **kwargs)
  File "c:\bg\test3.py", line 27, in fetch
    result = await session.get(url, headers=headers)
  File "C:\bg\benv\lib\site-packages\aiohttp\client.py", line 535, in _request
    conn = await self._connector.connect(
  File "C:\bg\benv\lib\site-packages\aiohttp\connector.py", line 542, in connect
    proto = await self._create_connection(req, traces, timeout)
  File "C:\bg\benv\lib\site-packages\aiohttp\connector.py", line 907, in _create_connection
    _, proto = await self._create_direct_connection(req, traces, timeout)
  File "C:\bg\benv\lib\site-packages\aiohttp\connector.py", line 1206, in _create_direct_connection
    raise last_exc
  File "C:\bg\benv\lib\site-packages\aiohttp\connector.py", line 1175, in _create_direct_connection
    transp, proto = await self._wrap_create_connection(
  File "C:\bg\benv\lib\site-packages\aiohttp\connector.py", line 992, in _wrap_create_connection
    raise client_error(req.connection_key, exc) from exc
aiohttp.client_exceptions.ClientConnectorError: Cannot connect to host xtreme-gacha.com:80 ssl:default [The semaphore timeout period has expired]
python-asyncio python
1 Answers

You can solve your problem with the help of new BucketRateLimiter library. I am author of the library and will be very grateful if you'll be using it.

Please find working solution below:

import asyncio
from datetime import datetime
import time

from aiohttp import ClientSession
from bucketratelimiter import AsyncioBucketTimeRateLimiter


TASKS_TO_COMPLETE = ['http://xtreme-gacha.com/bbs/accounts/fresh/40246F'] * 200

WORKER_NUM = 30

HEADERS = {
  'accept': 'application/json, text/javascript, */*; q=0.01',
  'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                'Chrome/105.0.0.0 Safari/537.36',
}


LIMITER = AsyncioBucketTimeRateLimiter(max_size=10, recovery_time=1)


@LIMITER
async def fetch(session: ClientSession, url, headers):
    format_time = "%H:%M:%S"
    start = datetime.utcnow().strftime(format_time)
    result = await session.get(url, headers=headers)
    end = datetime.utcnow().strftime(format_time)
    return start, end, result


async def worker(q: asyncio.Queue, session: ClientSession) -> None:
    """Workers which do some stuff."""
    while True:
        url = await q.get()
        if url is None:
            break
        start, end, result = await fetch(session, url, HEADERS)
        print(f"Result: {result} | {start} - {end}")


async def main_entry_point() -> None:
    """Main entry point of our asyncio app."""
    q = asyncio.Queue()
    for url in TASKS_TO_COMPLETE:
        await q.put(url)  # send tasks to queue

    for _ in range(WORKER_NUM):
        await q.put(None)  # use poison pill to stop workers

    async with LIMITER:
        async with ClientSession() as session:
            await asyncio.gather(*[worker(q, session) for _ in range(WORKER_NUM)])


if __name__ == '__main__':
    start_t = time.monotonic()
    asyncio.get_event_loop().run_until_complete(main_entry_point())
    print(f"Time passed: {time.monotonic() - start_t}")
Artiom Kozyrev0
Related