Как добавить сопрограмму к запущенному циклу asyncio?

Ответ 1

Вы можете использовать create_task для планирования новых сопрограмм:

import asyncio

async def cor1():
    ...

async def cor2():
    ...

async def main(loop):
    await asyncio.sleep(0)
    t1 = loop.create_task(cor1())
    await cor2()
    await t1

loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
loop.close()

Ответ 2

Чтобы добавить функцию в уже запущенный цикл событий, вы можете использовать:

asyncio.ensure_future(my_coro())

В моем случае я использовал многопоточность (threading) вместе с asyncio и хотел добавить задачу в уже запущенный цикл событий. Для кого-либо еще в той же ситуации обязательно укажите контур события (поскольку он не существует внутри Thread). то есть:

В глобальной области:

event_loop = asyncio.get_event_loop()

Затем позже внутри Thread:

asyncio.ensure_future(my_coro(), loop=event_loop)

Ответ 3

Ваш вопрос очень близок к "Как добавить вызов функции к запущенной программе?"

Когда вам нужно добавить новый сопроцессор в цикл событий?

Посмотрим на некоторые примеры. Здесь программа, которая запускает цикл событий с двумя сопрограммами параллельно:

import asyncio
from random import randint


async def coro1():
    res = randint(0,3)
    await asyncio.sleep(res)
    print('coro1 finished with output {}'.format(res))
    return res

async def main():
    await asyncio.gather(
        coro1(),
        coro1()
    ) # here we have two coroutines running parallely

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Вывод:

coro1 finished with output 1
coro1 finished with output 2
[Finished in 2.2s]

Может быть, вам нужно добавить некоторые сопрограммы, которые будут принимать результаты coro1 и использовать его, как только он будет готов? В этом случае просто создайте сопрограмму, ожидающую coro1, и используйте ее возвращающее значение:

import asyncio
from random import randint


async def coro1():
    res = randint(0,3)
    await asyncio.sleep(res)
    print('coro1 finished with output {}'.format(res))
    return res

async def coro2():
    res = await coro1()
    res = res * res
    await asyncio.sleep(res)
    print('coro2 finished with output {}'.format(res))
    return res

async def main():
    await asyncio.gather(
        coro2(),
        coro2()
    ) # here we have two coroutines running parallely

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Вывод:

coro1 finished with output 1
coro2 finished with output 1
coro1 finished with output 3
coro2 finished with output 9
[Finished in 12.2s]

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

Ответ 4

Ни один из ответов здесь, похоже, точно не отвечает на вопрос. Можно добавить задачи в цикл запущенного события, выполнив "родительскую" задачу для вас. Я не уверен, какой самый pythonic способ убедиться, что родительский кончик не закончится, пока все дети не закончили (предполагая, что поведение вы хотите), но это действительно работает.

import asyncio
import random


async def add_event(n):
    print('starting ' + str(n))
    await asyncio.sleep(n)
    print('ending ' + str(n))
    return n


async def main(loop):

    added_tasks = []

    delays = [x for x in range(5)]

    # shuffle to simulate unknown run times
    random.shuffle(delays)

    for n in delays:
        print('adding ' + str(n))
        task = loop.create_task(add_event(n))
        added_tasks.append(task)
        await asyncio.sleep(0)

    print('done adding tasks')

    # make a list of tasks that (maybe) haven't completed
    running_tasks = added_tasks[::]

    # wait until we see that all tasks have completed
    while running_tasks:
        running_tasks = [x for x in running_tasks if not x.done()]
        await asyncio.sleep(0)

    print('done running tasks')

    # extract the results from the tasks and return them
    results = [x.result() for x in added_tasks]
    return results


loop = asyncio.get_event_loop()
results = loop.run_until_complete(main(loop))
loop.close()
print(results)