Является ли нормальной практикой вызывать асинхронную функцию внутри цикла и как собрать все результаты?

Question

[Вадим]

async def my_func(value: object):
    some code
    return x


loop = asyncio.get_event_loop()

tasks = []
while True:
    task = my_func(some_object)
    tasks.append(asyncio.gather(task))
    if y == z:
        break

................
other code
................
results = loop.run_until_complete(tasks)
loop.close()

Является ли данный подход нормальным и как мне собрать/распечатать результаты всех вызовов my_func - так как я указал не работает

Comments

[Алан Гибизов]

Повторно указываю на недопустимость нарушения Регламента ресурса, а именно п.3.4
Если вы будете продолжать в таком духе, вопросы будут удаляться модератором, а аккаунт может быть заблокирован временно или навсегда за неоднократные нарушения.
Это последнее китайское предупреждение.

Answer 1

[Jack444]

Если нужен ассинхронный цикл то делается это через генератор

async def my_func(value: object):
    some code
    return x

async def my_func2():
    while True:
        y = await my_func(some_object)
        if y == z:
            break
        yield y

async def my_func3():
    async for i in my_func2():
        print(i)

Или ещё как вариант собрать таски и пойти циклом по gather.
for i in await gather(*tasks):

Comments

[Вадим]

кажется генератор самый старый вариант, я сделал так. Вроде работает. Dummy code в функциях

"""async testing module"""
import asyncio
import datetime


async def put_message_on_s3(message_date_: datetime, n_exec: int):
    """async function"""
    x = (12 - n_exec) / 2 + 0.1
    await asyncio.sleep(float(x))
    print(
        "message_date {} and number of execution = {}".format(
            message_date_.strftime("%B %-d, %Y"), str(n_exec)
        )
    )
    return "execution number: " + str(n_exec)


loop = asyncio.get_event_loop()

tasks = []
groups = []
groups_tuple = tuple()


async def main_func():
    """main async func"""
    ind = 0
    interim_results = []
    while True:
        mydate = datetime.datetime.now()
        task = asyncio.ensure_future(put_message_on_s3(mydate, ind))
        tasks.append(task)
        ind += 1
        print("\nNEXT MESSAGE\n")
        await asyncio.sleep(1)
        if ind >= 10:
            break

    for a_task in tasks:
        interim_result = await a_task
        interim_results.append(interim_result)
        # print(interim_result)

    return interim_results


loop = asyncio.get_event_loop()
if loop.is_closed():
    loop = asyncio.new_event_loop()

completion = loop.run_until_complete(main_func())
print("put_message_on_s3()= {}".format(completion))
loop.close()

[Jack444]

Вадим, Сойдет, Асинхронный генератор не самый старый а более низкоуровневый, сейчас функция gather его заменяет а вот ensure_future устаревает и в ближайших версиях будет удален, сейчас надо использовать asyncio.create_task(). Написанный код поидеи синхронный чтобы он стал асинхронным вместо

for a_task in tasks:
    interim_result = await a_task
    interim_results.append(interim_result)

напиши
return await asyncio.gather(*tasks)
В таком случае на порядки быстрее скрипт отработает, а в текущем случае код всеравно зависает на этой строке await asyncio.sleep(float(x))

[Вадим]

Jack444, а почему написанный код синхронный... например на печать то выводится "NEXT MESSAGE", то "message_date {} and number of execution = {}" причем в случайном порядке

Answer 2

[Пасечник Кузьмич]

Ваш код работает синхронно, каждая итерация блокируется до выполнения функции. Функцию gather надо вызывать с коллекцией сопрограмм, чтобы они выполнялись конкурентно.

Comments

[Вадим]

Можете подправить код чтобы работал асинхронно или хотя бы объяснить в чем дело?

[Пасечник Кузьмич]

Вадим, подправить не могу, приведённый вами код бессмысленный, непонятно от чего отталкиваться. Я уже объяснил, асинхронность имеет смысл, когда несколько асинхронных функций кооперативно делят операции ввода/вывода, а у вас запускается одна асинхронная функция, её выполнение поручается синхронной функции, после её завершения создаётся вторая асинхронная функция, и так далее.

[Вадим]

хорошо, асинхронность не работает когда асинхронная функция вызывается из синхронной?

[Vindicar]

Вадим, просто почитай как работает gather(). Куча вопросов отпадёт.

[Вадим]

Vindicar, Пасечник Кузьмич - вот так... dummy code

"""async testing module"""
import asyncio
import datetime


async def put_message_on_s3(message_date_: datetime, n_exec: int):
    """async function"""
    x = (12 - n_exec) / 2 + 0.1
    await asyncio.sleep(float(x))
    print(
        "message_date {} and number of execution = {}".format(
            message_date_.strftime("%B %-d, %Y"), str(n_exec)
        )
    )
    return "execution number: " + str(n_exec)


loop = asyncio.get_event_loop()

tasks = []
groups = []
groups_tuple = tuple()


async def main_func():
    """main async func"""
    ind = 0
    interim_results = []
    while True:
        mydate = datetime.datetime.now()
        task = asyncio.ensure_future(put_message_on_s3(mydate, ind))
        tasks.append(task)
        ind += 1
        print("\nNEXT MESSAGE\n")
        await asyncio.sleep(1)
        if ind >= 10:
            break

    for a_task in tasks:
        interim_result = await a_task
        interim_results.append(interim_result)
        # print(interim_result)

    return interim_results


loop = asyncio.get_event_loop()
if loop.is_closed():
    loop = asyncio.new_event_loop()

completion = loop.run_until_complete(main_func())
print("put_message_on_s3()= {}".format(completion))
loop.close()

Answer 3

[Владимир Олохтонов]

Стандартный паттерн выглядит вот так:

tasks = [my_func(v) for v in values]
results = await asyncio.gather(*tasks)

Comments

[Вадим]

можно без gather, через loop и future

[Владимир Олохтонов]

Вадим, можно, но не нужно.

По мере освоения области стало понятно, что старый путь с create_task таит в себе множество подводных камней.

В Python 3.11 чётко обозначено движение в направлении structured concurrency: https://docs.python.org/3.11/library/asyncio-task.... (ознакомиться с мнением Гвидо ван Россума можно тут)

Почитайте вот эту статью для понимания особенностей такого подхода: https://vorpus.org/blog/notes-on-structured-concur...

[Вадим]

Владимир Олохтонов,

на самом деле столько разных tutorials пл asyncio и не поймешь какой подход использовать. А какие камни с create_task?

Статья достаточно большая, попробую осилить. Спасибо вам большое... кстати, как вам этот код? Наполненность фунций не имеет значения !

"""async testing module"""
import asyncio
import datetime


async def put_message_on_s3(message_date_: datetime, n_exec: int):
    """async function"""
    x = (12 - n_exec) / 2 + 0.1
    await asyncio.sleep(float(x))
    print(
        "message_date {} and number of execution = {}".format(
            message_date_.strftime("%B %-d, %Y"), str(n_exec)
        )
    )
    return "execution number: " + str(n_exec)


loop = asyncio.get_event_loop()

tasks = []
groups = []
groups_tuple = tuple()


async def main_func():
    """main async func"""
    ind = 0
    interim_results = []
    while True:
        mydate = datetime.datetime.now()
        task = asyncio.ensure_future(put_message_on_s3(mydate, ind))
        tasks.append(task)
        ind += 1
        print("\nNEXT MESSAGE\n")
        await asyncio.sleep(1)
        if ind >= 10:
            break

    for a_task in tasks:
        interim_result = await a_task
        interim_results.append(interim_result)
        # print(interim_result)

    return interim_results


loop = asyncio.get_event_loop()
if loop.is_closed():
    loop = asyncio.new_event_loop()

completion = loop.run_until_complete(main_func())
print("put_message_on_s3()= {}".format(completion))
loop.close()

[Владимир Олохтонов]

Код сложнее чем нужен для выполняемой задачи, но вы дожидаетесь результата всех запущенных корутин вручную, так что с точки зрения структуры ок, в прод я бы такое не пропустил конечно :)

Про подводные камни: с практической точки зрения главная проблема в том, что в случае исключения вы не узнаете об этом до завершения программы если корутину не дожидаться, вторичные проблемы связаны с корректным завершением всего дерева запущенных корутин и тем что они могут держать ссылки на протухшие объекты.

[Вадим]

Я вас замучил вопросами... вы дожидаетесь результата всех запущенных корутин вручную - а как лучше их завершать? Через gather?

[Владимир Олохтонов]

Вадим, до 3.11 через gather или wait если нужен таймаут, начиная с 3.11 через TaskGroup.