Depuración con asyncio

Hay varias características útiles de depuración integradas en asyncio.

Primero, el bucle de eventos usa logging para emitir mensajes de estado mientras corre. Algunos de estos están disponibles si el registro está habilitado en una aplicación. Otros se pueden habilitar indicando al bucle que emita más mensajes de depuración. Llama a set_debug() pasando un valor booleano indicando si se debe habilitar o no la depuración.

Debido a que las aplicaciones creadas en asyncio son altamente sensibles a co-rutinas codiciosas que no ceden el control, hay apoyo para detección de devoluciones de llamada lentas integradas en el bucle de eventos. Enciéndelo para habilitar la depuración y controlar la definición de «lento» configurando la propiedad slow_callback_duration del bucle al número de segundos después de los cuales debe emitirse una advertencia.

Finalmente, si una aplicación que usa asyncio termina sin limpiar algunas de las co-rutinas u otros recursos, lo que puede significar que hay un error lógico que impide que parte del código de la aplicación se ejecute. Habilitar las advertencias ResourceWarning causa que estos casos sean reportados cuando termine el programa.

asyncio_debug.py

import argparse
import asyncio
import logging
import sys
import time
import warnings

parser = argparse.ArgumentParser('debugging asyncio')
parser.add_argument(
    '-v',
    dest='verbose',
    default=False,
    action='store_true',
)
args = parser.parse_args()

logging.basicConfig(
    level=logging.DEBUG,
    format='%(levelname)7s: %(message)s',
    stream=sys.stderr,
)
LOG = logging.getLogger('')


async def inner():
    LOG.info('inner starting')
    # Use a blocking sleep to simulate
    # doing work inside the function.
    time.sleep(0.1)
    LOG.info('inner completed')


async def outer(loop):
    LOG.info('outer starting')
    await asyncio.ensure_future(loop.create_task(inner()))
    LOG.info('outer completed')


event_loop = asyncio.get_event_loop()
if args.verbose:
    LOG.info('enabling debugging')

    # Enable debugging
    event_loop.set_debug(True)

    # Make the threshold for "slow" tasks very very small for
    # illustration. The default is 0.1, or 100 milliseconds.
    event_loop.slow_callback_duration = 0.001

    # Report all mistakes managing asynchronous resources.
    warnings.simplefilter('always', ResourceWarning)

LOG.info('entering event loop')
event_loop.run_until_complete(outer(event_loop))

Cuando se ejecuta sin la depuración habilitada, todo se ve bien con esta aplicación.

$ python3 asyncio_debug.py

  DEBUG: Using selector: KqueueSelector
   INFO: entering event loop
   INFO: outer starting
   INFO: inner starting
   INFO: inner completed
   INFO: outer completed

Activar la depuración expone algunos de los problemas que tiene, incluido el hecho de que, aunque inner() termine, lleva más tiempo hacerlo que el slow_callback_duration que se ha establecido y que el bucle de eventos no se cierra correctamente cuando el programa termina.

$ python3 asyncio_debug.py -v

  DEBUG: Using selector: KqueueSelector
   INFO: enabling debugging
   INFO: entering event loop
   INFO: outer starting
   INFO: inner starting
   INFO: inner completed
WARNING: Executing <Task finished coro=<inner() done, defined at
asyncio_debug.py:33> result=None created at asyncio_debug.py:43>
took 0.103 seconds
   INFO: outer completed