Python async / await "выстрелил и забыл"

115

Иногда необходимо выполнить некритическую асинхронную операцию, но я не хочу ждать ее завершения. В реализации сопрограммы Tornado вы можете «запустить и забыть» асинхронную функцию, просто пропустив yieldключевое слово.

Я пытался понять, как «выстрелить и забыть» с новым синтаксисом async/, awaitвыпущенным в Python 3.5. Например, упрощенный фрагмент кода:

async def async_foo():
    print("Do some stuff asynchronously here...")

def bar():
    async_foo()  # fire and forget "async_foo()"

bar()

Но происходит то, что он bar()никогда не выполняется, и вместо этого мы получаем предупреждение во время выполнения:

RuntimeWarning: coroutine 'async_foo' was never awaited
  async_foo()  # fire and forget "async_foo()"
Майк Н
источник
Связанный? stackoverflow.com/q/32808893/1639625 Фактически, я думаю, что это дубликат, но я не хочу мгновенно забивать его. Может кто-нибудь подтвердить?
tobias_k
3
@tobias_k, не думаю, что это дубликат. Ответ по ссылке слишком широк, чтобы отвечать на этот вопрос.
Михаил Герасимов
2
(1) ваш «основной» процесс продолжает работать вечно? Или (2) вы хотите позволить вашему процессу умереть, но позволить забытым задачам продолжать свою работу? Или (3) вы предпочитаете, чтобы основной процесс ждал забытых задач непосредственно перед завершением?
Жюльен Палар,

Ответы:

171

UPD:

Замените asyncio.ensure_futureна asyncio.create_taskвезде, если вы используете Python> = 3.7. Это более новый и приятный способ создания задачи .


asyncio.Task «выстрелил и забыл»

Согласно документации python, asyncio.Taskможно запустить некоторую сопрограмму для выполнения «в фоновом режиме» . Задача, созданная asyncio.ensure_future функцией , не будет блокировать выполнение (поэтому функция вернется немедленно!). Это похоже на способ «выстрелить и забыть», как вы просили.

import asyncio


async def async_foo():
    print("async_foo started")
    await asyncio.sleep(1)
    print("async_foo done")


async def main():
    asyncio.ensure_future(async_foo())  # fire and forget async_foo()

    # btw, you can also create tasks inside non-async funcs

    print('Do some actions 1')
    await asyncio.sleep(1)
    print('Do some actions 2')
    await asyncio.sleep(1)
    print('Do some actions 3')


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

Вывод:

Do some actions 1
async_foo started
Do some actions 2
async_foo done
Do some actions 3

Что делать, если задачи выполняются после завершения цикла событий?

Обратите внимание, что asyncio ожидает, что задача будет завершена в момент завершения цикла событий. Итак, если вы перейдете main()на:

async def main():
    asyncio.ensure_future(async_foo())  # fire and forget

    print('Do some actions 1')
    await asyncio.sleep(0.1)
    print('Do some actions 2')

Вы получите это предупреждение после завершения программы:

Task was destroyed but it is pending!
task: <Task pending coro=<async_foo() running at [...]

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

async def main():
    asyncio.ensure_future(async_foo())  # fire and forget

    print('Do some actions 1')
    await asyncio.sleep(0.1)
    print('Do some actions 2')


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

    # Let's also finish all running tasks:
    pending = asyncio.Task.all_tasks()
    loop.run_until_complete(asyncio.gather(*pending))

Убивайте задачи вместо того, чтобы ждать их

Иногда вы не хотите ждать выполнения задач (например, некоторые задачи могут быть созданы для вечного выполнения). В этом случае вы можете просто отменить () их вместо того, чтобы ждать их:

import asyncio
from contextlib import suppress


async def echo_forever():
    while True:
        print("echo")
        await asyncio.sleep(1)


async def main():
    asyncio.ensure_future(echo_forever())  # fire and forget

    print('Do some actions 1')
    await asyncio.sleep(1)
    print('Do some actions 2')
    await asyncio.sleep(1)
    print('Do some actions 3')


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

    # Let's also cancel all running tasks:
    pending = asyncio.Task.all_tasks()
    for task in pending:
        task.cancel()
        # Now we should await task to execute it's cancellation.
        # Cancelled task raises asyncio.CancelledError that we can suppress:
        with suppress(asyncio.CancelledError):
            loop.run_until_complete(task)

Вывод:

Do some actions 1
echo
Do some actions 2
echo
Do some actions 3
echo
Михаил Герасимов
источник
Я скопировал и пропустил первый блок и просто запустил его на своем конце, и по какой-то причине я получил: строка 4 async def async_foo (): ^ Как будто есть синтаксическая ошибка с определением функции в строке 4: "async def async_foo ( ):" Я что-то упускаю?
Гил Аллен
3
@GilAllen этот синтаксис работает только в Python 3.5+. Python 3.4 требует старого синтаксиса (см. Docs.python.org/3.4/library/asyncio-task.html ). Python 3.3 и ниже вообще не поддерживает asyncio.
Михаил Герасимов
Как бы вы убили задачи в потоке?… ̣У меня есть поток, который создает некоторые задачи, и я хочу убить все ожидающие, когда поток умирает в своем stop()методе.
Sardathrion - против злоупотреблений SE
@Sardathrion Я не уверен, указывает ли задача где-то в потоке, в котором она была создана, но ничто не мешает вам отслеживать их вручную: например, просто добавьте все задачи, созданные в потоке, в список, а когда придет время, отмените их, как объяснили выше.
Михаил Герасимов
2
Обратите внимание, что «Task.all_tasks () устарел, начиная с Python 3.7, вместо этого используйте asyncio.all_tasks ()»
Алексис
12

Спасибо Сергею за лаконичный ответ. Вот оформленная версия того же самого.

import asyncio
import time

def fire_and_forget(f):
    def wrapped(*args, **kwargs):
        return asyncio.get_event_loop().run_in_executor(None, f, *args, *kwargs)

    return wrapped

@fire_and_forget
def foo():
    time.sleep(1)
    print("foo() completed")

print("Hello")
foo()
print("I didn't wait for foo()")

Производит

>>> Hello
>>> foo() started
>>> I didn't wait for foo()
>>> foo() completed

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

nehem
источник
Я испытал существенное замедление после того, как использовал этот подход, создавая ~ 5 небольших задач «запустил и забыл» в секунду. Не используйте это в продакшене для длительной задачи. Это съест ваш процессор и память!
пир
10

Это не совсем асинхронное выполнение, но, возможно, вам подойдет run_in_executor () .

def fire_and_forget(task, *args, **kwargs):
    loop = asyncio.get_event_loop()
    if callable(task):
        return loop.run_in_executor(None, task, *args, **kwargs)
    else:    
        raise TypeError('Task must be a callable')

def foo():
    #asynchronous stuff here


fire_and_forget(foo)
Сергей Горностаев
источник
3
Хороший лаконичный ответ. Стоит отметить, что по executorумолчанию будет звонок concurrent.futures.ThreadPoolExecutor.submit(). Я упоминаю, потому что создание потоков не бесплатно; «выстрелил и забыл» 1000 раз в секунду, вероятно, сильно
Брэд Соломон
Ага. Я не прислушался к вашему предупреждению и испытал существенное замедление после использования этого подхода, создавая ~ 5 небольших задач «запустил и забыл» в секунду. Не используйте это в продакшене для длительной задачи. Это съест ваш процессор и память!
пир
3

По какой-то причине, если вы не можете использовать, asyncioто вот реализация с использованием простых потоков. Проверьте мои другие ответы и ответ Сергея тоже.

import threading

def fire_and_forget(f):
    def wrapped():
        threading.Thread(target=f).start()

    return wrapped

@fire_and_forget
def foo():
    time.sleep(1)
    print("foo() completed")

print("Hello")
foo()
print("I didn't wait for foo()")
nehem
источник
Если нам нужна только эта функция fire_and_forget и ничего больше из asyncio, было бы лучше использовать asyncio? Какие преимущества?
пир