Почему при больших данных виснет asyncio?

Question

[Никита Каменев]

async def run(r):
    tasks = []
    sem = asyncio.Semaphore(1000)

    async with ClientSession() as session:
        for url in r:
            task = asyncio.ensure_future(bound_fetch(sem, url, session))
            tasks.append(task)

        responses = await asyncio.gather(*tasks)

with open('0.txt') as f:
    urls = f.read().splitlines()

que = []
for url in urls:
    que.append(url)

    if len(que) == 5000:
        loop = asyncio.get_event_loop()
        future = asyncio.ensure_future(run(que))
        loop.run_until_complete(future)
        que = []

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run(que))
loop.run_until_complete(future)

Я подаю в потоки массив по 5000 элементов, если убрать эту часть и подать весь массив из 1млн, то виснет практически моментально. Если подавать слайсами по 5к начинает виснуть примерно через пару минут.

Где ошибка?

Comments

[Сергей Горностаев]

Вот бы ещё видеть, что делает bound_fetch и зачем ему семафор.

[Никита Каменев]

Сергей Горностаев,

async def bound_fetch(sem, _url, session, wappalyzer, col):
    # Getter function with semaphore.
    async with sem:
        try:
            page = await WebPage.new_from_url_async(url='http://' + _url, verify=False, aiohttp_client_session=session, timeout=7, headers={'User-Agent':'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.131 Safari/537.36'})
        except:
            return {}

        tech = wappalyzer.analyze_with_categories(page)
        print(tech)

[Никита Каменев]

Сергей Горностаев,

It says “too many open files”, and probably refers to number of open sockets. Why does it call them files? Sockets are just file descriptors, operating systems limit number of open sockets allowed. How many files are too many? I checked with python resource module and it seems like it’s around 1024. How can we bypass this? Primitive way is just increasing limit of open files. But this is probably not the good way to go. Much better way is just adding some synchronization in your client limiting number of concurrent requests it can process. I’m going to do this by adding asyncio.Semaphore() with max tasks of 1000.

Answer 1

[Vindicar]

Ты грузишь весь миллион адресов в память, дважды.
Первый раз, когда делаешь f.read(), потом в рамках .splitlines() создаётся копия (разбитая по кусочкам-строкам).
Ну и да, миллион индивидуальных тасков - это тоже дохрена. asyncio ведь надо проверить, может ли тот или иной таск продолжить работу.

Я бы сделал фиксированного размера пул тасков-воркеров , и заставил каждого воркера в цикле делать f.readline() самостоятельно, чтобы получить url для загрузки. И весь список в памяти хранить не надо, и контроль над количеством тасков получше.

Comments

[Никита Каменев]

Я бы сделал фиксированного размера пул тасков-воркеров

у меня же он фиксирован - 5к тасков

[Vindicar]

Никита Каменев, ну фиксированного размера = тасков 50/100. Я вот без понятия, как asyncio ведёт себя на таком количестве тасков, да ещё и с gather впридачу.
Либо можно оставить те же 5к, но не пересоздавать их - а заставить работать циклически.