programming - python async api

¿Cómo crear y ejecutar correctamente tareas simultáneas utilizando el módulo de asyncio de Python?

Question

Estoy tratando de comprender e implementar correctamente dos objetos de Task ejecutan simultáneamente usando el relativamente nuevo módulo de asyncio Python 3.

En pocas palabras, asyncio parece diseñado para manejar procesos asincrónicos y Task ejecución simultánea de Task en un bucle de eventos. Promueve el uso de await (aplicado en funciones asíncronas) como una forma libre de devolución de llamada para esperar y usar un resultado, sin bloquear el bucle de eventos. (Los futuros y las devoluciones de llamada siguen siendo una alternativa viable).

También proporciona la clase asyncio.Task() , una subclase especializada de Future diseñada para envolver corutinas. Preferentemente invocado mediante el método asyncio.ensure_future() . El uso previsto de las tareas de asincio es permitir que las tareas que se ejecutan independientemente se ejecuten ''concurrentemente'' con otras tareas dentro del mismo bucle de eventos. Tengo entendido que las Tasks están conectadas al bucle de eventos que luego sigue impulsando automáticamente la rutina entre las declaraciones de await .

Me gusta la idea de poder usar Tareas concurrentes sin necesidad de usar una de las clases Executor , pero no he encontrado mucha explicación sobre la implementación.

Así es como lo estoy haciendo actualmente:

import asyncio

print(''running async test'')

async def say_boo():
    i = 0
    while True:
        await asyncio.sleep(0)
        print(''...boo {0}''.format(i))
        i += 1

async def say_baa():
    i = 0
    while True:
        await asyncio.sleep(0)
        print(''...baa {0}''.format(i))
        i += 1

# wrap in Task object
# -> automatically attaches to event loop and executes
boo = asyncio.ensure_future(say_boo())
baa = asyncio.ensure_future(say_baa())

loop = asyncio.get_event_loop()
loop.run_forever()

En el caso de intentar ejecutar simultáneamente dos Tareas en bucle, he notado que, a menos que la Tarea tenga una expresión interna de await , se atascará en el bucle while, bloqueando efectivamente otras tareas para que no se ejecuten (al igual que un bucle while normal). Sin embargo, tan pronto como las Tareas tienen que (a) esperar, parecen ejecutarse simultáneamente sin ningún problema.

Por lo tanto, las declaraciones de await parecen proporcionar al bucle de eventos un punto de apoyo para alternar entre las tareas, dando el efecto de concurrencia.

Ejemplo de salida con await interna:

running async test
...boo 0
...baa 0
...boo 1
...baa 1
...boo 2
...baa 2

Ejemplo de salida sin await interna:

...boo 0
...boo 1
...boo 2
...boo 3
...boo 4

Preguntas

¿Esta implementación pasa por un ejemplo ''adecuado'' de tareas de bucle concurrente en asyncio ?

¿Es correcto que la única forma en que esto funciona es que una Task proporcione un punto de bloqueo ( await expresión) para que el bucle de eventos haga malabarismos con múltiples tareas?

Answer 1

No necesariamente necesita un yield from x para dar control sobre el bucle de eventos.

En su ejemplo, creo que la forma correcta sería hacer un yield None o, de manera equivalente, un yield simple, en lugar de un yield from asyncio.sleep(0.001) :

import asyncio

@asyncio.coroutine
def say_boo():
  i = 0
  while True:
    yield None
    print("...boo {0}".format(i))
    i += 1

@asyncio.coroutine
def say_baa():
  i = 0
  while True:
    yield
    print("...baa {0}".format(i))
    i += 1

boo_task = asyncio.async(say_boo())
baa_task = asyncio.async(say_baa())

loop = asyncio.get_event_loop()
loop.run_forever()

Las corutinas son simplemente viejos generadores de Python. Internamente, el bucle de eventos asyncio mantiene un registro de estos generadores y llama a gen.send() en cada uno de ellos en un bucle interminable. Cada vez que gen.send() , la llamada a gen.send() completa y el ciclo puede continuar. (Lo estoy simplificando; eche un vistazo a https://hg.python.org/cpython/file/3.4/Lib/asyncio/tasks.py#l265 para ver el código real)

Dicho esto, todavía iría a la ruta run_in_executor si necesita hacer un cálculo intensivo de la CPU sin compartir datos.

Answer 2

Sí, cualquier rutina que se esté ejecutando dentro de su bucle de eventos bloqueará la ejecución de otras rutinas y tareas, a menos que

1. Llama a otra corutina usando el yield from o await (si usa Python 3.5+).
2. Devoluciones.

Esto se debe a que el asyncio tiene un solo subproceso; La única forma de ejecutar el bucle de eventos es que ninguna otra rutina se ejecute activamente. El uso del yield from / waitit suspende temporalmente la rutina, lo que le da al bucle de eventos la oportunidad de funcionar.

Su código de ejemplo está bien, pero en muchos casos, probablemente no querrá código de ejecución prolongada que no esté ejecutando E / S asíncrona dentro del bucle de eventos para comenzar. En esos casos, a menudo tiene más sentido usar BaseEventLoop.run_in_executor para ejecutar el código en un hilo o proceso en segundo plano. ProcessPoolExecutor sería la mejor opción si su tarea está vinculada a la CPU, ThreadPoolExecutor se usaría si necesita hacer alguna E / S que no sea asyncio asincio.

Sus dos bucles, por ejemplo, están completamente vinculados a la CPU y no comparten ningún estado, por lo que el mejor rendimiento provendría del uso de ProcessPoolExecutor para ejecutar cada bucle en paralelo en las CPU:

import asyncio
from concurrent.futures import ProcessPoolExecutor

print(''running async test'')

def say_boo():
    i = 0
    while True:
        print(''...boo {0}''.format(i))
        i += 1


def say_baa():
    i = 0
    while True:
        print(''...baa {0}''.format(i))
        i += 1

if __name__ == "__main__":
    executor = ProcessPoolExecutor(2)
    loop = asyncio.get_event_loop()
    boo = asyncio.ensure_future(loop.run_in_executor(executor, say_boo))
    baa = asyncio.ensure_future(loop.run_in_executor(executor, say_baa))

    loop.run_forever()