Когда подходящее время для вызова loop.close ()?

Distilled to its simplest, my use case is a bunch of blocking "old school" calls, which I wrap in the run_in_executor() and an outer coroutine; if any of those calls goes wrong, I want to stop progress, cancel the ones still outstanding

Это не может работать так, как предполагалось, потому что run_in_executor отправляет функцию в пул потоков, а потоки ОС не могут быть отменены в Python (или на других языках, которые их раскрывают). Отмена будущего, возвращенного run_in_executor, будет пытаться отменить базовый concurrent.futures.Future, но это будет иметь эффект только в том случае, если функция блокировки еще не запущена, например потому что пул потоков занят. Как только он начинает выполняться, его нельзя безопасно отменить. Поддержка безопасной и надежной отмены - одно из преимуществ использования asyncio по сравнению с потоками.

Если вы имеете дело с синхронным кодом, будь то устаревший блокирующий вызов или более продолжительный код, связанный с процессором, вы должны запустить его с run_in_executor и включить способ его прерывания. Например, код может иногда проверять флаг stop_requested и завершать работу, если это правда, возможно, вызывая исключение. Затем вы можете «отменить» эти задачи, установив соответствующий флаг или флаги.

how should I modify the code above in a way that with the call to close() included does not raise?

Насколько я могу судить, в настоящее время нет возможности сделать это без модификации blocking и кода верхнего уровня. run_in_executor будет настаивать на информировании цикла событий о результате, и это не удается, когда цикл событий закрывается. Отмена asyncio future не помогает, потому что проверка отмены выполняется в потоке цикла событий, а ошибка возникает до этого, когда call_soon_threadsafe вызывается рабочим потоком. (Можно было бы переместить проверку в рабочий поток, но следует тщательно проанализировать, не приводит ли это к состоянию гонки между вызовом cancel() и фактической проверкой.)

why does it raise at all? what more does the loop want from the coros/tasks: they either exited; raised; or were canceled: isn't this enough to keep it happy?

Он хочет, чтобы функции блокировки, переданные в run_in_executor (буквально называемые blocking в вопросе), которые уже были запущены, завершили работу до закрытия цикла событий. Вы отменили асинхронное будущее, но лежащее в основе параллельное будущее все еще хочет «вызвать домой», обнаружив, что цикл замкнут.

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

• Не пытайтесь отменить отложенные фьючерсы. Их отмена на первый взгляд кажется правильной, но это не дает вам возможности использовать wait() для этих фьючерсов, поскольку asyncio будет считать их завершенными.
• Вместо этого отправьте вашим фоновым задачам событие для конкретного приложения, информирующее их о том, что они должны быть прерваны.
• Позвоните в loop.run_until_complete(asyncio.wait(pending)) перед loop.close().

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

what actually happens if I don't call close() - in this trivial case, I presume it's largely redundant; but what might the consequences be in a "real" production code?

Так как типичный цикл обработки событий работает, пока работает приложение, не должно возникнуть проблем, если вы не вызовете close() в самом конце программы. Операционная система в любом случае очистит ресурсы при выходе из программы.

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

Пока проблема апстрима не будет исправлен, другой способ обойти проблему - заменить использование run_in_executor специальной версией без недостатка. Хотя запуск собственного run_in_executor поначалу звучит как плохая идея, на самом деле это лишь небольшой связующий элемент между concurrent.futures и будущим asyncio.

Простая версия run_in_executor может быть полностью реализована с использованием общедоступного API этих двух классов:

def run_in_executor(executor, fn, *args):
    """Submit FN to EXECUTOR and return an asyncio future."""
    loop = asyncio.get_event_loop()
    if args:
        fn = functools.partial(fn, *args)
    work_future = executor.submit(fn)
    aio_future = loop.create_future()
    aio_cancelled = False

    def work_done(_f):
        if not aio_cancelled:
            loop.call_soon_threadsafe(set_result)

    def check_cancel(_f):
        nonlocal aio_cancelled
        if aio_future.cancelled():
            work_future.cancel()
            aio_cancelled = True

    def set_result():
        if work_future.cancelled():
            aio_future.cancel()
        elif work_future.exception() is not None:
            aio_future.set_exception(work_future.exception())
        else:
            aio_future.set_result(work_future.result())

    work_future.add_done_callback(work_done)
    aio_future.add_done_callback(check_cancel)

    return aio_future

Когда loop.run_in_executor(blocking) заменяется на run_in_executor(executor, blocking), где executor является ThreadPoolExecutor, созданным в main(), код работает без других модификаций.

Конечно, в этом варианте синхронные функции будут продолжать выполняться в другом потоке до завершения, несмотря на то, что они были отменены, но это неизбежно без их модификации для поддержки явного прерывания.