subplot python

¿Cómo puedo ejecutar periódicamente una función con asyncio?

Question

Estoy migrando de tornado a asyncio , y no puedo encontrar el equivalente asyncio de PeriodicCallback de tornado . (Un PeriodicCallback toma dos argumentos: la función a ejecutar y la cantidad de milisegundos entre llamadas).

• ¿Hay tal equivalente en asyncio ?
• Si no, ¿cuál sería la forma más limpia de implementar esto sin correr el riesgo de obtener un RecursionError después de un tiempo?

Answer 1

Basado en. Respuesta de Jesse Jiryu Davis (con @Torkel Bjørnson-Langen y comentarios @ReWrite) esta es una mejora que evita la deriva.

import time
import asyncio

@asyncio.coroutine
def periodic(period):
    def g_tick():
        t = time.time()
        count = 0
        while True:
            count += 1
            yield max(t + count * period - time.time(), 0)
    g = g_tick()

    while True:
        print(''periodic'', time.time())
        yield from asyncio.sleep(next(g))

loop = asyncio.get_event_loop()
task = loop.create_task(periodic(1))
loop.call_later(5, task.cancel)

try:
    loop.run_until_complete(task)
except asyncio.CancelledError:
    pass

Answer 2

Cuando sientas que algo debería suceder "en el fondo" de tu programa asyncio.Task , asyncio.Task podría ser una buena manera de hacerlo. Puede leer esta publicación para ver cómo trabajar con las tareas.

Aquí hay una posible implementación de la clase que ejecuta alguna función periódicamente:

import asyncio
from contextlib import suppress


class Periodic:
    def __init__(self, func, time):
        self.func = func
        self.time = time
        self.is_started = False
        self._task = None

    async def start(self):
        if not self.is_started:
            self.is_started = True
            # Start task to call func periodically:
            self._task = asyncio.ensure_future(self._run())

    async def stop(self):
        if self.is_started:
            self.is_started = False
            # Stop task and await it stopped:
            self._task.cancel()
            with suppress(asyncio.CancelledError):
                await self._task

    async def _run(self):
        while True:
            await asyncio.sleep(self.time)
            self.func()

Vamos a probarlo:

async def main():
    p = Periodic(lambda: print(''test''), 1)
    try:
        print(''Start'')
        await p.start()
        await asyncio.sleep(3.1)

        print(''Stop'')
        await p.stop()
        await asyncio.sleep(3.1)

        print(''Start'')
        await p.start()
        await asyncio.sleep(3.1)
    finally:
        await p.stop()  # we should stop task finally


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

Salida:

Start
test
test
test

Stop

Start
test
test
test

[Finished in 9.5s]

Como ve al start , simplemente comenzamos una tarea que llama a algunas funciones y duerme un tiempo en bucle infinito. En la stop , simplemente cancelamos esa tarea. Tenga en cuenta que esa tarea debe detenerse en el momento en que finaliza el programa.

Una cosa más importante es que su devolución de llamada no debería tomar mucho tiempo para ejecutarse (o congelará su ciclo de evento). Si planeas llamar a algún func larga ejecución, posiblemente necesitarías ejecutarlo en ejecutor .

Answer 3

No hay soporte integrado para llamadas periódicas, no.

Simplemente crea tu propio bucle de programador que duerme y ejecuta cualquier tarea programada:

import math, time

async def scheduler():
    while True:
        # sleep until the next whole second
        now = time.time()
        await asyncio.sleep(math.ceil(now) - now)

        # execute any scheduled tasks
        await for task in scheduled_tasks(time.time()):
            await task()

El iterador scheduled_tasks() debe producir tareas que estén listas para ejecutarse en el momento dado. Tenga en cuenta que producir el cronograma y poner en marcha todas las tareas en teoría podría llevar más de 1 segundo; la idea aquí es que el programador rinde todas las tareas que deberían haber comenzado desde la última comprobación.

Answer 4

Para las versiones de Python por debajo de 3.5:

import asyncio

@asyncio.coroutine
def periodic():
    while True:
        print(''periodic'')
        yield from asyncio.sleep(1)

def stop():
    task.cancel()

loop = asyncio.get_event_loop()
loop.call_later(5, stop)
task = loop.create_task(periodic())

try:
    loop.run_until_complete(task)
except asyncio.CancelledError:
    pass

Para Python 3.5 y superior:

import asyncio

async def periodic():
    while True:
        print(''periodic'')
        await asyncio.sleep(1)

def stop():
    task.cancel()

loop = asyncio.get_event_loop()
loop.call_later(5, stop)
task = loop.create_task(periodic())

try:
    loop.run_until_complete(task)
except asyncio.CancelledError:
    pass