Просмотр и расширение Trio с помощью trio.lowlevel

Глобальная статистика

trio.lowlevel.current_statistics() → RunStatistics

Возвращает объект, содержащий отладочную информацию на уровне цикла обработки событий:

class trio.lowlevel.RunStatistics

Объект, содержащий отладочную информацию на уровне цикла обработки событий.

В настоящее время определены следующие поля:

• tasks_living (целое число): Количество задач, которые были созданы и ещё не завершены.

• tasks_runnable (целое число): Количество задач, которые в настоящее время находятся в очереди выполнения (по сравнению с заблокированными, ожидающими чего-то).

• seconds_to_next_deadline (вещественное число): Время до следующей запланированной даты истечения срока действия области отмены. Может быть отрицательным, если срок действия истек, но мы ещё не обработали отмены. Может быть inf, если нет запланированных сроков действия.

• run_sync_soon_queue_size (целое число): Количество необработанных обратных вызовов, поставленных в очередь с помощью trio.lowlevel.TrioToken.run_sync_soon().

• io_statistics (объект): Некоторые статистические данные из I/O-бэкенда Trio. У него всегда есть атрибут backend, который является строкой, называющей используемый бэкенд ввода-вывода операционной системы; другие атрибуты варьируются в зависимости от бэкенда.

Проверка наличия Trio

Если вам нужно взаимодействовать с активным запуском Trio — например, вам нужно узнать current_time() или current_task() — Trio нужно иметь определённое состояние, иначе вы получите RuntimeError("must be called from async context"). Это требует, чтобы вы либо:

• были косвенно внутри (и в том же потоке, что и) вызова trio.run(), для получения информации на уровне запуска, такой как current_time() или current_clock(); или

• были косвенно внутри задачи Trio, для получения информации на уровне задачи, такой как current_task() или current_effective_deadline().

Внутренне это состояние предоставляется переменными потока, отслеживающими текущий запуск и текущую задачу. Иногда полезно заранее знать, потерпит ли неудачу вызов, или иметь динамичную информацию для защиты от запуска чего-либо внутри или вне Trio. Для этого вызовите trio.lowlevel.in_trio_run() или trio.lowlevel.in_trio_task(), которые дадут ответы в соответствии со следующей таблицей.

trio.lowlevel.in_trio_run() → bool

Проверяет, находимся ли мы в запуске Trio. Возвращает True только в том случае, если current_time() будет успешным.

См. также обсуждение различных способов обнаружения Trio.

trio.lowlevel.in_trio_task() → bool

Проверяет, находимся ли мы в задаче Trio. Возвращает True только в том случае, если current_task() будет успешным.

См. также обсуждение различных способов обнаружения Trio.

Текущий таймер

trio.lowlevel.current_clock() → Clock

Возвращает текущий Clock.

API инструментов

API инструментов предоставляет стандартный способ добавления пользовательской инструментики в цикл выполнения. Хотите построить гистограмму задержек планирования, записать трассировку стека любого задания, которое блокирует цикл выполнения более чем на 50 мс, или измерить процент времени работы процесса, потраченного на ожидание ввода-вывода? Это то место.

Общая идея заключается в том, что в любой момент времени trio.run() поддерживает набор «инструментов», которые представляют собой объекты, реализующие интерфейс trio.abc.Instrument. Когда происходит интересное событие, он перебирает эти инструменты и уведомляет их, вызывая соответствующий метод. Учебник содержит простой пример использования этого для отслеживания.

Поскольку это подключение к Trio на довольно низком уровне, нужно быть осторожным. Обратные вызовы выполняются синхронно, и во многих случаях, если они завершатся ошибкой, нет никакого разумного способа распространить эту ошибку (например, мы можем быть глубоко в механизме обработки исключений...). Поэтому наша текущая стратегия обработки исключений, поднятых инструментами, заключается в (а) записи исключения в журнал "trio.abc.Instrument", который по умолчанию выводит трассировку стека в стандартный поток ошибок, и (б) отключении нарушающего инструмент.

Вы можете зарегистрировать начальный список инструментов, передав их в trio.run(). add_instrument() и remove_instrument() позволяют добавлять и удалять инструменты во время выполнения.

trio.lowlevel.add_instrument(instrument: Instrument) → None

Начать инструментацию текущего цикла выполнения заданным инструментом.

Параметры:

instrument (trio.abc.Instrument) – Инструмент, который нужно активировать.

Если instrument уже активен, ничего не делает.

trio.lowlevel.remove_instrument(instrument: Instrument) → None

Остановка инструментации текущего цикла выполнения заданным инструментом.

Параметры:

instrument (trio.abc.Instrument) – Инструмент, который нужно деактивировать.

Возбуждает:

KeyError – если инструмент не активен. Это может произойти либо потому, что вы его никогда не добавляли, либо потому, что вы его добавили, а затем он поднял необработанное исключение и был автоматически деактивирован.

И вот интерфейс для реализации, если вы хотите создать свой собственный Instrument:

class trio.abc.Instrument

Интерфейс для инструментации цикла выполнения.

Инструменты не обязаны наследоваться от этого абстрактного базового класса, и все эти методы являются необязательными. Этот класс служит в основном для документации.

after_io_wait(timeout: float) → None

Вызывается после обработки ожидающего ввода-вывода.

Параметры:

timeout (float) – Количество секунд, на которое мы были готовы ждать. Это время может или не может истечь, в зависимости от того, был ли готов какой-либо ввод-вывод.

after_run() → None

Вызывается непосредственно перед возвращением trio.run().

after_task_step(task: Task) → None

Вызывается при возвращении в основной цикл выполнения после того, как задание сгенерировало выход.

Параметры:

task (trio.lowlevel.Task) – Задание, которое только что выполнилось.

before_io_wait(timeout: float) → None

Вызывается перед блокированием для ожидания готовности ввода-вывода.

Параметры:

timeout (float) – Количество секунд, на которое мы готовы ждать.

before_run() → None

Вызывается в начале trio.run().

before_task_step(task: Task) → None

Вызывается непосредственно перед возобновлением выполнения заданного задания.

Параметры:

task (trio.lowlevel.Task) – Задание, которое скоро будет выполнено.

task_exited(task: Task) → None

Вызывается, когда заданное задание завершается.

Параметры:

task (trio.lowlevel.Task) – Завершенное задание.

task_scheduled(task: Task) → None

Вызывается, когда заданное задание становится выполнимым.

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

Параметры:

task (trio.lowlevel.Task) – Задание, которое стало выполнимым.

task_spawned(task: Task) → None

Вызывается при создании заданного задания.

Параметры:

task (trio.lowlevel.Task) – Новое задание.

В учебнике есть полный пример определения пользовательского инструмента для регистрации внутренних решений планирования Trio.

Универсально доступный API

Все среды предоставляют следующие функции:

await trio.lowlevel.wait_readable(obj)

Ожидает, пока ядро сообщит, что данный объект готов к чтению.

В системах Unix, obj должен быть целочисленным дескриптором файла или объектом с методом .fileno(), возвращающим целочисленный дескриптор файла. Можно передавать любой тип дескриптора файла, хотя точное поведение будет зависеть от ядра. Например, для файлов на диске это вряд ли будет полезно.

В системах Windows, obj должен быть целочисленным дескриптором SOCKET-ручки или объектом с методом .fileno(), возвращающим целочисленный дескриптор SOCKET-ручки. Дескрипторы файлов не поддерживаются, как и ручки, ссылающиеся на что-либо, кроме SOCKET.

Исключения:

• trio.BusyResourceError – если другая задача уже ожидает, пока данный сокет станет готовым к чтению.

• trio.ClosedResourceError – если другая задача вызывает notify_closing(), пока эта функция ещё работает.

await trio.lowlevel.wait_writable(obj)

Ожидает, пока ядро сообщит, что данный объект готов к записи.

См. wait_readable для определения obj.

Исключения:

• trio.BusyResourceError – если другая задача уже ожидает, пока данный сокет станет готовым к записи.

• trio.ClosedResourceError – если другая задача вызывает notify_closing(), пока эта функция ещё работает.

trio.lowlevel.notify_closing(obj)

Вызовите эту функцию перед закрытием дескриптора файла (в Unix) или сокета (в Windows). Это заставит любые вызовы wait_readable или wait_writable на данном объекте немедленно проснуться и выбросить ClosedResourceError.

Эта функция не закрывает объект – вам все равно нужно сделать это самостоятельно. Также следует быть осторожным, чтобы не запускать новые задачи, ожидающие объекта между вызовом этой функции и фактическим закрытием. Поэтому, чтобы корректно закрыть что-то, обычно нужно выполнить эти шаги в определённом порядке:

1. Явно пометить объект как закрытый, чтобы любые новые попытки использовать его прервались до начала.

2. Вызвать notify_closing, чтобы разбудить всех уже существующих пользователей.

3. Фактически закрыть объект.

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

API, специфичное для Unix

FdStream поддерживает обертывание Unix-файлов (таких как пайп или TTY) в виде потока.

Если у вас есть два разных дескриптора файлов для отправки и получения и вы хотите объединить их в единый двунаправленный поток Stream, используйте trio.StapledStream:

bidirectional_stream = trio.StapledStream(
    trio.lowlevel.FdStream(write_fd),
    trio.lowlevel.FdStream(read_fd)
)

class trio.lowlevel.FdStream(fd: int)

Базируется на Stream

Представляет поток, используя дескриптор файла пайпа, TTY и т.п.

fd должен ссылаться на файл, открытый для чтения и/или записи, поддерживающий асинхронный ввод-вывод (пайпы и TTY будут работать, файлы на диске — вероятно, нет). Возвращаемый поток берет на себя владение fd, поэтому закрытие потока также закроет fd. Как и в os.fdopen, вы не должны напрямую использовать fd после его обертывания в поток с помощью этой функции.

Для использования в качестве потока Trio открытый файл должен быть помещён в режим без блокировки. К сожалению, это влияет на все ввод-вывод, проходящие через подлежащий открытый файл, включая ввод-вывод, использующий другой дескриптор файла, чем тот, который был передан Trio. Если другие потоки или процессы используют дескрипторы файлов, связанные через os.dup или наследование через os.fork к тому, который использует Trio, они вряд ли будут готовы к внезапному внедрению семантики асинхронного ввода-вывода. Например, можно использовать FdStream(os.dup(sys.stdin.fileno())) для получения потока для чтения со стандартного ввода, но это безопасно только при серьёзных оговорках: ваш стандартный ввод не должен быть общим ни с какими другими процессами, и вы не должны вызывать синхронные методы sys.stdin, пока поток, возвращённый FdStream, не будет закрыт. См. проблему #174 для обсуждения сложностей, связанных с ослаблением этого ограничения.

Параметры:

fd (int) – fd для обертывания.

Возвращаемое значение:

Новый объект FdStream.

API, специфичное для Kqueue

TODO: эти функции реализованы, но на данный момент представляют собой больше набросок, чем что-то реальное. См. #26.

trio.lowlevel.current_kqueue()

await trio.lowlevel.wait_kevent(ident, filter, abort_func)

with trio.lowlevel.monitor_kevent(ident, filter) as queue

API Windows

await trio.lowlevel.WaitForSingleObject(handle)

Асинхронный и отменяемый вариант WaitForSingleObject. Только для Windows.

Параметры:

handle – Движок Win32, как целое число Python.

Исключения:

OSError – Если движок недействителен, например, если он уже закрыт.

TODO: эти функции реализованы, но пока являются скорее набросками, чем чем-то реальным. См. #26 и #52.

trio.lowlevel.register_with_iocp(handle)

await trio.lowlevel.wait_overlapped(handle, lpOverlapped)

await trio.lowlevel.write_overlapped(handle, data)

await trio.lowlevel.readinto_overlapped(handle, data)

trio.lowlevel.current_iocp()

with trio.lowlevel.monitor_completion_key() as queue

Абстракция очереди ожидания

class trio.lowlevel.ParkingLot

Справедливая очередь ожидания с возможностью отмены и повторного помещения в очередь.

Этот класс обобщает сложные части реализации очереди ожидания. Он полезен для реализации синхронизирующих примитивов более высокого уровня, таких как очереди и блокировки.

В дополнение к методам ниже, вы можете использовать len(parking_lot) для получения количества приостановленных задач и if parking_lot: ... для проверки наличия приостановленных задач.

: list[Task] broken_by

break_lot(task: Task | None = None) → None

Разбить эту парковку, с задачей task отмеченной как задача, которая это сделала.

Это приводит к тому, что все приостановленные задачи генерируют ошибку, а любые будущие попытки парковки также вызовут ошибку. Unpark и repark становятся пустыми операциями, так как парковка пуста.

Возникающая ошибка содержит ссылку на задачу, переданную в качестве параметра. Задача также сохраняется в парковке в атрибуте broken_by.

await park() → None

Приостановить текущую задачу до тех пор, пока она не будет разбужена вызовом unpark() или unpark_all().

Исключения:

BrokenResourceError – если попытка парковки выполняется в сломанной парковке или парковка ломается, прежде чем мы дойдём до разбуживания.

repark(new_lot: ParkingLot, *, count: int | float = 1) → None

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

Это удаляет из одной парковки count задачи и переупорядочивает их в другой, сохраняя порядок. Например:

async def parker(lot):
    print("sleeping")
    await lot.park()
    print("woken")

async def main():
    lot1 = trio.lowlevel.ParkingLot()
    lot2 = trio.lowlevel.ParkingLot()
    async with trio.open_nursery() as nursery:
        nursery.start_soon(parker, lot1)
        await trio.testing.wait_all_tasks_blocked()
        assert len(lot1) == 1
        assert len(lot2) == 0
        lot1.repark(lot2)
        assert len(lot1) == 0
        assert len(lot2) == 1
        # This wakes up the task that was originally parked in lot1
        lot2.unpark()

Если приостановленных задач меньше, чем count, тогда перемещаются доступные задачи и возвращается успешный результат.

Параметры:

• new_lot (ParkingLot) – парковка, в которую нужно переместить задачи.

• count (int|math.inf) – количество задач для перемещения.

repark_all(new_lot: ParkingLot) → None

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

См. repark() для деталей.

statistics() → ParkingLotStatistics

Возвращает объект, содержащий отладочную информацию.

В настоящее время определены следующие поля:

• tasks_waiting: Количество задач, заблокированных в методе park() этой парковки.

unpark(*, count: int | float = 1) → list[Task]

Разбудить одну или несколько задач.

Это разбудит count задачи, заблокированные в park(). Если приостановленных задач меньше, чем count, то разбуживаются доступные задачи, и затем возвращается успешный результат.

Параметры:

count (int | math.inf) – количество задач для разбуживания.

unpark_all() → list[Task]

Разбудить все приостановленные задачи.

class trio.lowlevel.ParkingLotStatistics(tasks_waiting: int)

Объект, содержащий отладочную информацию для ParkingLot.

В настоящее время определены следующие поля:

• tasks_waiting (int): Количество задач, заблокированных в методе trio.lowlevel.ParkingLot.park() этой парковки.

trio.lowlevel.add_parking_lot_breaker(task: Task, lot: ParkingLot) → None

Регистрирует задачу как разрушитель парковки. См. ParkingLot.break_lot().

Исключения:

trio.BrokenResourceError – если задача уже завершена.

trio.lowlevel.remove_parking_lot_breaker(task: Task, lot: ParkingLot) → None

Удаляет регистрацию задачи как разрушителя парковки. См. ParkingLot.break_lot()

Функции контрольных точек низкого уровня

await trio.lowlevel.checkpoint() → None

Чистая контрольная точка.

Проверяет отмену и позволяет планировать другие задачи без блокировки.

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

Эквивалентно await trio.sleep(0) (которое реализуется вызовом checkpoint().)

Следующие две функции используются вместе для создания контрольной точки:

await trio.lowlevel.checkpoint_if_cancelled() → None

Вызовите контрольную точку, если контекст вызывающего элемента был отменён.

Эквивалентно (но потенциально более эффективному):

if trio.current_effective_deadline() == -inf:
    await trio.lowlevel.checkpoint()

Это либо нет операция, либо она позволяет планировать другие задачи, а затем генерирует исключение trio.Cancelled.

Обычно используется вместе с cancel_shielded_checkpoint().

await trio.lowlevel.cancel_shielded_checkpoint() → None

Введите точку планирования, но не точку отмены.

Это не контрольная точка, но это половина контрольной точки, а в сочетании с checkpoint_if_cancelled() она может образовать полную контрольную точку.

Эквивалентно (но потенциально более эффективному):

with trio.CancelScope(shield=True):
    await trio.lowlevel.checkpoint()

Они часто используются в случаях, когда у вас есть операция, которая может или не может заблокировать выполнение, и вы хотите реализовать стандартную семантику контрольных точек Trio. Пример:

async def operation_that_maybe_blocks():
    await checkpoint_if_cancelled()
    try:
        ret = attempt_operation()
    except BlockingIOError:
        # need to block and then retry, which we do below
        pass
    else:
        # operation succeeded, finish the checkpoint then return
        await cancel_shielded_checkpoint()
        return ret
    while True:
        await wait_for_operation_to_be_ready()
        try:
            return attempt_operation()
        except BlockingIOError:
            pass

Эта логика немного запутанная, но выполняет всё следующее:

• Каждый успешный путь выполнения проходит через контрольную точку (предполагая, что wait_for_operation_to_be_ready является безусловной контрольной точкой)

• Наши семантика отмены говорят, что Cancelled должно быть возбуждено только в том случае, если операция не произошла. Использование cancel_shielded_checkpoint() на ветви выхода пораньше достигает этой цели.

• В пути, где мы всё-таки блокируемся, мы не проходим через какие-либо точки планирования до этого, что избегает некоторых ненужных операций.

• Избегает неявного объединения BlockingIOError с любыми ошибками, генерируемыми attempt_operation или wait_for_operation_to_be_ready, сохраняя цикл while True: за пределами блока except BlockingIOError:.

Эти функции также могут быть полезны в других ситуациях. Например, когда trio.to_thread.run_sync() планирует какую-то работу для выполнения в потоке-работнике, она блокируется до завершения работы (поэтому это точка планирования), но по умолчанию она не допускает отмены. Таким образом, чтобы убедиться, что вызов всегда действует как контрольная точка, она вызывает checkpoint_if_cancelled() перед запуском потока.

Низкоуровневое блокирование

await trio.lowlevel.wait_task_rescheduled(abort_func: Callable[[Callable[[], NoReturn]], Abort]) → Any

Уложить текущую задачу в сон с поддержкой отмены.

Это низкоуровневый API для блокирования в Trio. Каждый раз, когда задача Task блокируется, она делает это, вызывая эту функцию (обычно косвенно через какой-либо API более высокого уровня).

Это сложный интерфейс без защитных ограждений. Если вы можете использовать ParkingLot или встроенные функции ожидания ввода-вывода, то вам следует это сделать.

В общем случае, перед вызовом этой функции, вы подготавливаете «кого-то», кто вызовет reschedule() для текущей задачи в какой-то момент позже.

Затем вы вызываете wait_task_rescheduled(), передавая abort_func, «обработчик отмены».

(Терминология: в Trio «отмена» — это процесс попытки прервать заблокированную задачу для передачи отмены.)

Есть два возможных сценария дальнейших действий:

1. «Кто-то» вызывает reschedule() для текущей задачи, и wait_task_rescheduled() возвращает или генерирует любое значение или ошибку, которые были переданы в reschedule().

2. Контекст вызова переходит в состояние отмены (например, из-за истечения таймаута). В этом случае вызывается abort_func. Его интерфейс выглядит так:

   def abort_func(raise_cancel):
       ...
       return trio.lowlevel.Abort.SUCCEEDED  # or FAILED

   Он должен попытаться очистить любые состояния, связанные с этим вызовом, и, в частности, организовать, чтобы reschedule() не вызывался позже. Если (и только если!) это удастся, он должен вернуть Abort.SUCCEEDED, в этом случае задача будет автоматически перепланирована с соответствующей ошибкой Cancelled.

   В противном случае, он должен вернуть Abort.FAILED. Это означает, что задача не может быть отменена в данный момент и всё ещё должна убедиться, что «кто-то» в конечном итоге вызовет reschedule().

   На этом этапе снова есть два варианта. Вы можете просто проигнорировать отмену: дождаться завершения операции, а затем перепланировать и продолжить как обычно. (Например, именно это делает trio.to_thread.run_sync(), если отмена отключена.) Другой вариант заключается в том, что abort_func успешно отменяет операцию, но по какой-то причине не может сообщить об этом сразу. (Пример: в Windows можно запросить отмену асинхронной («перекрытой») операции ввода-вывода, но этот запрос также асинхронный — вы узнаете об этом позже, отменилась операция или нет.) Для сообщения об отложенной отмене вам следует самостоятельно перепланировать задачу и вызвать raise_cancel обработчик, переданный в abort_func, для поднятия исключения Cancelled (или, возможно, KeyboardInterrupt) в эту задачу. Любой из описанных ниже подходов может подойти:

   # Option 1:
   # Catch the exception from raise_cancel and inject it into the task.
   # (This is what Trio does automatically for you if you return
   # Abort.SUCCEEDED.)
   trio.lowlevel.reschedule(task, outcome.capture(raise_cancel))
   
   # Option 2:
   # wait to be woken by "someone", and then decide whether to raise
   # the error from inside the task.
   outer_raise_cancel = None
   def abort(inner_raise_cancel):
       nonlocal outer_raise_cancel
       outer_raise_cancel = inner_raise_cancel
       TRY_TO_CANCEL_OPERATION()
       return trio.lowlevel.Abort.FAILED
   await wait_task_rescheduled(abort)
   if OPERATION_WAS_SUCCESSFULLY_CANCELLED:
       # raises the error
       outer_raise_cancel()

   В любом случае гарантируется, что мы вызываем abort_func не более одного раза на каждый вызов wait_task_rescheduled().

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

Предупреждение

Если ваш abort_func генерирует ошибку или возвращает любое значение, отличное от Abort.SUCCEEDED или Abort.FAILED, Trio потерпит полную неудачу. Будьте внимательны! Аналогично, вполне возможно заблокировать программу Trio, не перепланировав заблокированную задачу, или нанести ущерб, вызвав reschedule() слишком много раз. Помните, что мы говорили выше о том, что следует использовать API более высокого уровня, если это возможно?

class trio.lowlevel.Abort(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

enum.Enum используется в качестве возвращаемого значения от функций отмены.

См. wait_task_rescheduled() для получения подробной информации.

SUCCEEDED

FAILED

trio.lowlevel.reschedule(task: Task, next_send: Outcome[object] =

Перепланировать задачу с заданным outcome.Outcome.

См. wait_task_rescheduled() для подробных сведений.

Должен быть ровно один вызов reschedule() для каждого вызова wait_task_rescheduled(). (И при подсчете имейте в виду, что возвращение Abort.SUCCEEDED из обработчика отмены эквивалентно вызову reschedule() один раз.)

Параметры:

• task (trio.lowlevel.Task) – задача, подлежащая перепланировке. Должна быть заблокирована в вызове wait_task_rescheduled().

• next_send (outcome.Outcome) – значение (или ошибка), которое должно быть возвращено (или возбуждено) из wait_task_rescheduled().

Вот пример класса блокировки, реализованного с помощью wait_task_rescheduled() напрямую. Эта реализация имеет ряд недостатков, включая отсутствие справедливости, отмену O(n), отсутствие проверки ошибок, отсутствие вставки контрольной точки на пути без блокировок и т. д. Если вы действительно хотите реализовать свою собственную блокировку, изучите реализацию trio.Lock и используйте ParkingLot, который обрабатывает некоторые из этих проблем за вас. Но это служит иллюстрацией основного строения API wait_task_rescheduled():

class NotVeryGoodLock:
    def __init__(self):
        self._blocked_tasks = collections.deque()
        self._held = False

    async def acquire(self):
        # We might have to try several times to acquire the lock.
        while self._held:
            # Someone else has the lock, so we have to wait.
            task = trio.lowlevel.current_task()
            self._blocked_tasks.append(task)
            def abort_fn(_):
                self._blocked_tasks.remove(task)
                return trio.lowlevel.Abort.SUCCEEDED
            await trio.lowlevel.wait_task_rescheduled(abort_fn)
            # At this point the lock was released -- but someone else
            # might have swooped in and taken it again before we
            # woke up. So we loop around to check the 'while' condition
            # again.
        # if we reach this point, it means that the 'while' condition
        # has just failed, so we know no-one is holding the lock, and
        # we can take it.
        self._held = True

    def release(self):
        self._held = False
        if self._blocked_tasks:
            woken_task = self._blocked_tasks.popleft()
            trio.lowlevel.reschedule(woken_task)

Что такое «гостевой режим»?

Цикл событий действует как центральный координатор для управления всеми операциями ввода-вывода в вашей программе. Обычно это означает, что ваше приложение должно выбрать один цикл событий и использовать его для всего. Но что, если вам нравится Trio, но вам также нужно использовать фреймворк, например, Qt или PyGame, у которого есть свой собственный цикл событий? Тогда вам нужен способ запустить оба цикла событий одновременно.

Объединение циклов событий возможно, но стандартные подходы имеют существенные недостатки:

• Опрос: в этом случае вы используете цикл ожидания для ручного проверки ввода-вывода в обоих циклах событий много раз в секунду. Это добавляет задержку и тратит время процессора и электроэнергию.

• Подключаемые бэкенды ввода-вывода: в этом случае вы повторно реализуете один из API циклов событий поверх другого, поэтому в итоге получаете только один цикл событий. Это требует значительной работы для каждой пары циклов событий, которые вы хотите интегрировать, и различные бэкенды неизбежно ведут к несовместимому поведению, вынуждая пользователей программировать с учетом наименьшего общего знаменателя. И если два цикла событий предоставляют разные наборы функций, может быть даже невозможно реализовать один через другой.

• Запуск двух циклов событий в отдельных потоках: Это работает, но большинство API циклов событий не являются потокобезопасными, поэтому в этом подходе вам нужно тщательно отслеживать, какой код выполняется в каком цикле событий, и помнить об использовании явного межпоточного обмена сообщениями всякий раз, когда вы взаимодействуете с другим циклом — в противном случае вы рискуете возникновением трудноотслеживаемых гонок и повреждения данных.

Вот почему Trio предлагает четвертый вариант: гостевой режим. Гостевой режим позволяет выполнять trio.run поверх какого-либо другого «хостового» цикла событий, например Qt. Его преимущества:

• Эффективность: гостевой режим ориентирован на события, вместо использования цикла ожидания, поэтому он имеет низкую задержку и не тратит электроэнергию.

• Нет необходимости думать о потоках: ваш код Trio выполняется в том же потоке, что и хостовый цикл событий, поэтому вы можете свободно вызывать синхронные API Trio из хоста и вызывать синхронные API хоста из Trio. Например, если вы создаете приложение GUI с Qt в качестве хостового цикла, то создание кнопки отмены и подключение ее к trio.CancelScope так же просто, как написание:

  # Trio code can create Qt objects without any special ceremony...
  my_cancel_button = QPushButton("Cancel")
  # ...and Qt can call back to Trio just as easily
  my_cancel_button.clicked.connect(my_cancel_scope.cancel)

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

• Согласованное поведение: гостевой режим использует тот же код, что и обычный Trio: тот же планировщик, тот же код ввода-вывода, то же самое во всем. Таким образом, вы получаете полный набор функций, и все работает так, как ожидается.

• Простая интеграция и широкая совместимость: практически каждый цикл событий предлагает какую-то потокобезопасную операцию «планирования обратного вызова», и этого достаточно, чтобы использовать его как хостовый цикл.

Действительно? Как это возможно?

Примечание

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

Все циклы событий имеют одинаковую базовую структуру. Они циклически выполняют две операции:

1. Ожидание уведомления операционной системы о том, что произошло что-то интересное, например, данные, прибывающие на сокет, или истекло время ожидания. Это делается путем вызова специфичного для платформы системного вызова sleep_until_something_happens() — select, epoll, kqueue, GetQueuedCompletionEvents и т.д.

2. Выполнение всех пользовательских задач, относящихся к тому, что произошло, а затем возвращение к шагу 1.

Проблема здесь в шаге 1. Два разных цикла событий в одном потоке могут по очереди выполнять пользовательские задачи в шаге 2, но когда они простаивают и ничего не происходит, они не могут оба вызвать свою функцию sleep_until_something_happens() одновременно.

Стратегии «опрос» и «подключаемые бэкенды» решают эту проблему, модифицируя циклы, чтобы оба шага 1 могли выполняться одновременно в одном потоке. Поддержание всего в одном потоке отлично для шага 2, но правки шага 1 создают проблемы.

Стратегия «отдельные потоки» решает эту проблему, перемещая оба шага в отдельные потоки. Это позволяет выполнить шаг 1, но недостатком является то, что теперь пользовательские задачи в шаге 2 также выполняются в отдельных потоках, поэтому пользователи вынуждены заниматься межпоточной координацией.

Идея гостевого режима заключается в объединении лучших частей каждого подхода: мы перемещаем шаг 1 Trio в отдельный рабочий поток, сохраняя шаг 2 Trio в основном хостовом потоке. Таким образом, когда приложение простаивает, оба цикла событий выполняют свои sleep_until_something_happens() одновременно в своих потоках. Но когда приложение оживает и ваш код фактически выполняется, всё происходит в одном потоке. Сложности с потоками обрабатываются прозрачно внутри Trio.

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

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

Однако есть полезная оптимизация: нам необходим поток только тогда, когда наш sleep_until_something_happens() вызов фактически приостанавливается, то есть когда часть Trio вашей программы простаивает и не имеет ничего общего. Таким образом, прежде чем переключиться в рабочий поток, мы дважды проверяем, простаивает ли он, и если нет, то пропускаем рабочий поток и переходим непосредственно к шагу 2. Это означает, что ваше приложение платит дополнительную цену за переключение потоков только в моменты, когда в противном случае оно приостанавливалось, поэтому оно должно минимально влиять на общую производительность вашего приложения.

Общий накладные расходы будут зависеть от вашего хостового цикла, вашей платформы, вашего приложения и т. д. Но мы ожидаем, что в большинстве случаев приложения, работающие в гостевом режиме, будут только на 5-10% медленнее, чем тот же код, использующий trio.run. Если вы обнаружите, что это не так для вашего приложения, сообщите нам, и мы посмотрим, сможем ли мы это исправить!

Реализация гостевого режима для вашего любимого цикла событий

Давайте пройдемся по тому, что вам нужно сделать, чтобы интегрировать гостевой режим Trio с вашим любимым циклом событий. Рассматривайте этот раздел как список задач.

Начало работы: Первый шаг — заставить что-то базовое работать. Вот минимальный пример запуска Trio поверх asyncio, который вы можете использовать в качестве модели:

import asyncio
import trio

# A tiny Trio program
async def trio_main():
    for _ in range(5):
        print("Hello from Trio!")
        # This is inside Trio, so we have to use Trio APIs
        await trio.sleep(1)
    return "trio done!"

# The code to run it as a guest inside asyncio
async def asyncio_main():
    asyncio_loop = asyncio.get_running_loop()

    def run_sync_soon_threadsafe(fn):
        asyncio_loop.call_soon_threadsafe(fn)

    def done_callback(trio_main_outcome):
        print(f"Trio program ended with: {trio_main_outcome}")

    # This is where the magic happens:
    trio.lowlevel.start_guest_run(
        trio_main,
        run_sync_soon_threadsafe=run_sync_soon_threadsafe,
        done_callback=done_callback,
    )

    # Let the host loop run for a while to give trio_main time to
    # finish. (WARNING: This is a hack. See below for better
    # approaches.)
    #
    # This function is in asyncio, so we have to use asyncio APIs.
    await asyncio.sleep(10)

asyncio.run(asyncio_main())

Вы можете видеть, что мы используем API-интерфейсы, специфичные для asyncio, для запуска цикла, а затем вызываем trio.lowlevel.start_guest_run. Эта функция очень похожа на trio.run и принимает все те же аргументы. Но у нее есть два отличия:

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

И во-вторых, она требует двух дополнительных ключевых аргументов: run_sync_soon_threadsafe и done_callback.

Для run_sync_soon_threadsafe нам нужна функция, которая принимает синхронный обратный вызов и планирует его для выполнения в вашем цикле хоста. И эта функция должна быть «потокобезопасной» в том смысле, что вы можете безопасно вызывать её из любого потока. Поэтому вам нужно разобраться, как написать функцию, которая делает это, используя API вашего цикла хоста. В asyncio это легко, так как call_soon_threadsafe делает именно то, что нам нужно; для вашего цикла это может быть более или менее сложно.

Для done_callback вы передаете функцию, которую Trio автоматически вызовет при завершении выполнения Trio, так что вы знаете, что это завершено и что произошло. В этой базовой стартовой версии мы просто выводим результат; в следующем разделе мы обсудим лучшие альтернативы.

На этом этапе вы должны быть в состоянии запустить простую программу Trio внутри вашего цикла хоста. Теперь мы превратим этот прототип во что-то надежное.

Жизненный цикл циклов событий: Одна из самых сложных вещей в большинстве циклов событий — правильное завершение. А наличие двух циклов событий усложняет это еще больше!

Если возможно, мы рекомендуем следовать этой схеме:

• Запустите свой цикл хоста

• Немедленно вызовите start_guest_run для запуска Trio

• Когда Trio завершит работу и вызов done_callback, завершите цикл хоста

• Убедитесь, что ничего другого не завершает ваш цикл хоста

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

Вот как мы бы расширили наш пример asyncio, чтобы реализовать эту схему:

# Improved version, that shuts down properly after Trio finishes
async def asyncio_main():
    asyncio_loop = asyncio.get_running_loop()

    def run_sync_soon_threadsafe(fn):
        asyncio_loop.call_soon_threadsafe(fn)

    # Revised 'done' callback: set a Future
    done_fut = asyncio_loop.create_future()
    def done_callback(trio_main_outcome):
        done_fut.set_result(trio_main_outcome)

    trio.lowlevel.start_guest_run(
        trio_main,
        run_sync_soon_threadsafe=run_sync_soon_threadsafe,
        done_callback=done_callback,
    )

    # Wait for the guest run to finish
    trio_main_outcome = await done_fut
    # Pass through the return value or exception from the guest run
    return trio_main_outcome.unwrap()

А затем вы можете инкапсулировать всю эту механику в утилитарную функцию, которая предоставляет API-интерфейс, похожий на trio.run, но запускает оба цикла вместе:

def trio_run_with_asyncio(trio_main, *args, **trio_run_kwargs):
    async def asyncio_main():
        # same as above
        ...

    return asyncio.run(asyncio_main())

Технически, можно использовать и другие схемы. Но есть некоторые важные ограничения, которые нужно соблюдать:

• Вы должны позволить программе Trio завершиться. Многие циклы событий позволяют остановить цикл событий в любой момент, и все ожидающие обратные вызовы/задачи и т.д. просто… не выполняются. Trio использует более структурированную систему, где вы можете отменять вещи, но код всегда выполняется до конца, поэтому finally блоки выполняются, ресурсы очищаются и т.д. Если вы остановите свой цикл хоста слишком рано, прежде чем вызов done_callback не будет вызван, это прервет выполнение Trio посредине без возможности очистки. Это может оставить ваш код в несогласованном состоянии и определенно оставит внутренние компоненты Trio в несогласованном состоянии, что вызовет ошибки, если вы снова попытаетесь использовать Trio в этом потоке.

  Некоторым программам нужно иметь возможность завершаться в любой момент, например, в ответ на закрытие окна графического интерфейса или выбор пользователем пункта «Выход» в меню. В таких случаях мы рекомендуем обернуть всю вашу программу в trio.CancelScope и отменить её, когда вы захотите выйти.

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

• Если вы или ваш цикл хоста не зарегистрировали обработчик для signal.SIGINT до запуска Trio (это не обычно), Trio возьмет на себя доставку KeyboardInterrupt. И поскольку Trio не может определить, какой код хоста безопасно прервать, он будет только передавать KeyboardInterrupt в часть кода Trio. Это нормально, если ваша программа настроена на завершение, когда часть Trio завершается, потому что KeyboardInterrupt будет распространяться из Trio и затем вызовет завершение вашего цикла хоста, что именно вам и нужно.

Учитывая эти ограничения, мы считаем, что самый простой подход — всегда запускать и останавливать оба цикла вместе.

Управление сигналами: «Сигналы» — это низкоуровневая примитивная межпроцессная коммуникация. Когда вы нажимаете Ctrl+C для завершения программы, используется сигнал. Обработка сигналов в Python имеет много подвижных элементов. Одним из этих элементов является signal.set_wakeup_fd, который циклы событий используют для того, чтобы гарантировать, что они проснутся, когда придёт сигнал, чтобы на него отреагировать. (Если у вас когда-либо был цикл событий, который игнорировал вас, когда вы нажимали Ctrl+C, это, вероятно, было потому, что они не использовали signal.set_wakeup_fd правильно.)

Однако, только один цикл событий может использовать signal.set_wakeup_fd одновременно. В гостевом режиме это может привести к проблемам: Trio и цикл хоста могут начать борьбу за использование signal.set_wakeup_fd.

Некоторые циклы событий, например asyncio, не будут работать правильно, если они не выиграют эту борьбу. К счастью, Trio немного менее избирателен: пока кто-то гарантирует, что программа просыпается, когда приходит сигнал, она должна работать правильно. Поэтому, если ваш цикл хоста хочет использовать signal.set_wakeup_fd, отключите поддержку signal.set_wakeup_fd в Trio, и оба цикла будут работать правильно.

С другой стороны, если ваш цикл хоста не использует signal.set_wakeup_fd, то единственный способ сделать всё правильно — включить поддержку signal.set_wakeup_fd Trio.

По умолчанию Trio предполагает, что ваш цикл хоста не использует signal.set_wakeup_fd. Он пытается обнаружить, когда это приводит к конфликту с циклом хоста, и выводит предупреждение — но, к сожалению, к тому времени, как он его обнаруживает, ущерб уже нанесён. Поэтому, если вы получаете это предупреждение, вам следует отключить поддержку signal.set_wakeup_fd Trio, передав host_uses_signal_set_wakeup_fd=True в start_guest_run.

Если вы не видите никаких предупреждений с вашим начальным прототипом, вы, вероятно, в порядке. Но единственный способ быть уверенным — проверить исходный код вашего цикла хоста. Например, asyncio может или не может использовать signal.set_wakeup_fd в зависимости от версии Python и операционной системы.

Небольшая оптимизация: Наконец, рассмотрите небольшую оптимизацию. Некоторые циклы событий предлагают две версии своего API «вызвать эту функцию вскоре»: одну, которую можно использовать из любого потока, и одну, которая может быть использована только из потока цикла событий, причём последняя является более дешевой. Например, asyncio имеет как call_soon_threadsafe, так и call_soon.

Если у вас есть такой цикл, вы также можете передать ключевой аргумент run_sync_soon_not_threadsafe=... в start_guest_run, и Trio автоматически будет его использовать, когда это необходимо.

Если у вашего цикла нет такого разделения, не беспокойтесь об этом; run_sync_soon_not_threadsafe= является необязательным. (Если он не передан, Trio будет просто использовать вашу потокобезопасную версию во всех случаях.)

И всё! Если вы выполнили все эти шаги, вы должны получить чистую интегрированную гибридную схему цикла событий. Идите создавать крутые GUI/игры/что угодно!

Ограничения

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

Пользовательские таймеры могут использоваться в режиме гостя, но они влияют только на таймауты Trio, а не на таймауты цикла хоста. А таймер мгновенного перехода и связанные с ним trio.testing.wait_all_tasks_blocked технически могут использоваться в режиме гостя, но они будут учитывать только задачи Trio при решении, нужно ли перепрыгивать через таймер или все задачи заблокированы.

Справочник

trio.lowlevel.start_guest_run(async_fn: Callable[..., Awaitable[RetT]], *args: object, run_sync_soon_threadsafe: Callable[[Callable[[], object]], object], done_callback: Callable[[outcome.Outcome[RetT]], object], run_sync_soon_not_threadsafe: Callable[[Callable[[], object]], object] | None = None, host_uses_signal_set_wakeup_fd: bool = False, clock: Clock | None = None, instruments: Sequence[Instrument] = (), restrict_keyboard_interrupt_to_checkpoints: bool = False, strict_exception_groups: bool = True) → None

Запуск «гостевого» выполнения Trio поверх другого цикла событий «хоста».

Каждый цикл хоста может иметь только одно гостевое выполнение одновременно.

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

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

После успешного возвращения start_guest_run() гостевое выполнение будет достаточно настроенно для вызова функций Trio с синхронизацией, таких как current_time(), spawn_system_task() и current_trio_token(). Если во время этой ранней настройки гостевого выполнения произойдет TrioInternalError, оно будет поднято из start_guest_run(). Все остальные ошибки, включая все ошибки, поднятые функцией async_fn, будут доставлены в ваш done_callback в какой-то момент после успешного возвращения из start_guest_run().

Параметры:

• run_sync_soon_threadsafe –

  Произвольное вызываемое значение, которому будет передана функция в качестве единственного аргумента:

  def my_run_sync_soon_threadsafe(fn):
      ...

  Это вызываемое значение должно запланировать fn() для выполнения хостом в следующий проход по циклу. Должен поддерживать вызов из произвольных потоков.

• done_callback –

  Произвольное вызываемое значение:

  def my_done_callback(run_outcome):
      ...

  Когда выполнение Trio завершится, Trio вызовет этот обратный вызов, чтобы сообщить вам об этом. Аргументом является outcome.Outcome, сообщающий о том, что должно было быть возвращено или поднято trio.run. Эта функция может делать что угодно, но обычно вы захотите остановить цикл хоста, разразобраться с результатом и т. д.

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

• host_uses_signal_set_wakeup_fd (bool) – Передайте True, если ваш цикл хоста использует signal.set_wakeup_fd, и False в противном случае. Более подробную информацию см. в разделе Реализация режима гостя для вашего любимого цикла событий.

Для значения других аргументов см. trio.run.