Ограничие asyncio task по памяти?

Question

[kaveri571]

Есть дамп данных размером 4 гб, дамп читаю построчно и добавляю задачи в asyncio task. Создается около 2х миллионов задач в памяти, что естественно памяти не хватает. Как ограничить asyncio task к примеру до 10 000, чтобы после выполнения их, создались новые task из дампа, то есть разбить чтение дампа на части. Задача стоит огромный дамп (который сначала подредактировать) положить в базу данных.

async def main(path, engine):
    tasks = []
    with jsonlines.open(path) as reader:
        for inx, obj in enumerate(reader, start=1):
            task = asyncio.create_task(load_today_db(engine=engine, obj=obj))
            tasks.append(task)
        await asyncio.gather(*tasks)

Answer 1

[Алексей Черемисин]

Задача совсем несложная, используйте очереди! https://docs.python.org/3/library/asyncio-queue.html

Comments

[kaveri571]

Подробнее можно?! Не совсем понял как можно применить

[Алексей Черемисин]

Делаете очередь с нужным объемом, в нее пуляете, пулялка висит, пока из очереди писалка не выгребет нужного количества. Так пулялка и писалка вместе и живут. Пулялка пишет в очередь и тормозится, пока пулялка не разгрузит очередь...
Там кстати пример есть...

[kaveri571]

Набросал, все равно по максимуму идет нагрузка. В этой теме плохо разбираюсь, поэтому все спрашиваю

def read_json(path):
    with jsonlines.open(path) as reader:
        for inx, obj in enumerate(reader, start=1):
            yield obj

async def main(path, engine):
    N = 100
    tasks = asyncio.Queue()

    for item in read_json(path=path):
        tasks.put_nowait(load_today_db(engine=engine, obj=item))

    async def worker():
        while not tasks.empty():
            await tasks.get_nowait()

    await asyncio.gather(*[worker() for _ in range(N)])

[Алексей Черемисин]

Вот так примерно

import asyncio

async def consumer(queue):
    loop = True
    while(loop or (not queue.empty())):
        try:
            item = await queue.get()
            print(f'- Item {item}')
        except asyncio.CancelledError:
            loop = False
            print('-- Interrupt task')
        

async def producer(queue):
    for item in range(1,1001):
        print(f'+ Item {item}')
        await queue.put(item)

async def main():
    queue = asyncio.Queue(maxsize=100)
    p = asyncio.create_task(producer(queue))
    c = asyncio.create_task(consumer(queue))
    
    await p
    c.cancel()
    await c

asyncio.run(main())

[Алексей Черемисин]

Вот так даже интереснее :)

import asyncio
import random

async def consumer(queue):
    loop = True
    while(loop or (not queue.empty())):
        try:
            item = await queue.get()
            print(f'- Item {item}')
            await asyncio.sleep(random.randint(1,100)*.00001)
        except asyncio.CancelledError:
            loop = False
            print('-- Interrupt task')
        

async def producer(queue):
    for item in range(1,1001):
        print(f'+ Item {item}')
        await queue.put(item)
        await asyncio.sleep(random.randint(1,100)*.00001)

async def main():
    queue = asyncio.Queue(maxsize=100)
    p = asyncio.create_task(producer(queue))
    c = asyncio.create_task(consumer(queue))
    
    await p
    c.cancel()
    await c

asyncio.run(main())

.......
+ Item 991
+ Item 992
- Item 990
+ Item 993
+ Item 994
- Item 991
- Item 992
+ Item 995
- Item 993
+ Item 996
+ Item 997
- Item 994
+ Item 998
- Item 995
+ Item 999
- Item 996
+ Item 1000
- Item 997
-- Interrupt task
- Item 998
- Item 999
- Item 1000

[d'Ivan]

Алексей Черемисин , я бы уточнил, что "пуляка" в данном контексте происходит от слова pull (тянуть, дергать).
То есть потребитель сообщений из очереди.

[kaveri571]

Алексей Черемисин, Использовал ваш пример, работает, но вопрос появился новый, когда except срабатывает, получается мы отбрасывает эту задачу, как ее повторно отработать?! При таком подходе, когда данные в бд ложишь, происходит лимит соединений, как можно этот момент учитывать? использовать pgbouncer?

[Алексей Черемисин]

kaveri571,

когда except срабатывает, получается мы отбрасывает эту задачу, как ее повторно отработать?!

Все просто, Exception у нас срабатывает в следующих случаях:
- либо в методе .get, и здесь мы просто получаем Exception, но ничего не достаем из очереди и просто уходим на следующий цикл, где достанем остаток и продолжим работать
- либо на таймауте, это в последнем примере, но перед этим мы типа уже достали из очереди и что-то положили в базу :) И да, таймаут нам не нужен, а просто заменяется на обращение к базе данных, оберните это в отдельную функцию, и там уже обрабатывайте ошибки самой базы.

[Алексей Черемисин]

Роман Мирр, пулялка, не от аглицкого "pull", а от русского запулить, выпустить пулю, и больше соответствует аглицкому push :) Мне немного некомфортны consumer и produser, всегда в них путаюсь :) Как с право и лево :)

[Алексей Черемисин]

kaveri571, У нас же всего два потока, как может здесь заканчиваться лимит соединений к базе?! У вас похоже где-то ошибка. Я в всоих примерах не порождаю кучи потоков - из всего два, один читает и запихивает в очередь, второй - из этой очереди читает.