Отладка с помощью asyncio

В asyncio встроено несколько полезных функций отладки.

Во-первых, цикл событий использует ведение журнала для отправки сообщений о состоянии во время работы. Некоторые из них доступны, если в приложении включено ведение журнала. Другие можно включить, указав циклу выдавать больше отладочных сообщений. Вызовите set_debug () , передав логическое значение, указывающее, следует ли включать отладку.

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

Наконец, если приложение, использующее asyncio , завершает работу без очистки некоторых сопрограмм или других ресурсов, это может означать, что есть логическая ошибка, препятствующая запуску некоторого кода приложения. Включение предупреждений ResourceWarning приводит к тому, что об этих случаях сообщается при выходе из программы.

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',
    defaultFalse,
    action'store_true',
)
args  parser.parse_args()

logging.basicConfig(
    levellogging.DEBUG,
    format'%(levelname)7s: %(message)s',
    streamsys.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))

При запуске без включенной отладки все выглядит нормально с этим приложением.

$ python3 asyncio_debug.py

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

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

$ python3 asyncio_debug.py -v

  DEBUG: Using selector: KqueueSelector
   INFO: enabling debugging
   INFO: entering event loop
   INFO: outer starting
WARNING: Executing 
running at asyncio_debug.py:33>
object at 0x106e0d288>()] created at asyncio_debug.py:43>
.../lib/python3.7/asyncio/base_events.py:158] created at
.../lib/python3.7/asyncio/base_events.py:552> took 0.001 seconds
   INFO: inner starting
   INFO: inner completed
WARNING: Executing >
took 0.101 seconds
   INFO: outer completed