Как цикл событий узнает что корутина закончила работу в asyncio?

Как цикл событий узнает о завершении корутины в asyncio

Это один из самых важных механизмов в asyncio. Цикл событий отслеживает состояние корутины через систему сигналов (callbacks) и Future объекты.

Механизм отслеживания

1. Задача (Task) как обертка над корутиной

Когда ты вызываешь asyncio.create_task(), создается объект Task, который:

• Оборачивает корутину
• Отслеживает её состояние
• Уведомляет цикл событий о завершении

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return 'done'

async def main():
    # Task - это Future, обертка над корутиной
    task = asyncio.create_task(my_coroutine())
    
    # Task имеет методы:
    print(task.done())  # False - еще не завершена
    result = await task
    print(task.done())  # True - завершена
    print(result)       # 'done'

asyncio.run(main())

2. Состояние Future/Task

Vнутри задачи есть внутреннее состояние:

• _state: 'PENDING' (ожидание) или 'FINISHED' (завершена)
• _callbacks: список колбэков для вызова при завершении

import asyncio
import inspect

async def check_task_state():
    async def worker():
        await asyncio.sleep(0.1)
        return 42
    
    task = asyncio.create_task(worker())
    
    # Внутреннее состояние
    print(f'Состояние: {task._state}')  # PENDING
    
    await task
    
    print(f'Состояние: {task._state}')  # FINISHED
    print(f'Результат: {task._coro}')   # Информация о корутине

asyncio.run(check_task_state())

Как происходит уведомление

Шаг 1: Регистрация callback-функций

Когда задача создана, callback-функции могут быть зарегистрированы через add_done_callback():

async def main():
    async def worker():
        await asyncio.sleep(1)
        return 'result'
    
    task = asyncio.create_task(worker())
    
    # Регистрируем callback - вызовется при завершении
    def on_completion(t):
        print(f'Task завершена! Результат: {t.result()}')
    
    task.add_done_callback(on_completion)
    
    await task

asyncio.run(main())

Шаг 2: Сигнал от await/yield from

Корутина использует yield для передачи управления циклу событий. При завершении await выражения цикл событий получает сигнал:

async def detailed_flow():
    print('1. Начало корутины')
    
    # Здесь управление передается циклу событий
    # Цикл получает Future от asyncio.sleep
    await asyncio.sleep(1)
    
    print('2. Корутина возобновилась - цикл знает, что sleep завершился')
    return 'done'

# На уровне CPython это работает через generator protocol

Шаг 3: Цикл событий проверяет готовность

Цикл событий постоянно:

1. Проверяет какие Future/задачи готовы (используя selectors)
2. Запускает их callback-функции
3. Возобновляет корутины через .send() или вызывает result()

# Упрощенная модель работы цикла событий
class SimpleEventLoop:
    def __init__(self):
        self.ready = []  # Готовые к выполнению задачи
        self.waiting = {}  # Ожидающие задачи
    
    def run_until_complete(self, coro):
        task = asyncio.Task(coro)
        while not task.done():
            # 1. Проверяем готовые задачи
            if task in self.ready:
                # 2. Возобновляем корутину
                try:
                    next_future = task._coro.send(None)
                except StopIteration as e:
                    # 3. Корутина завершилась!
                    task._state = 'FINISHED'
                    task._result = e.value
                    # 4. Вызываем callback-функции
                    self._call_callbacks(task)
                    break

Детальный механизм завершения

Через Future.set_result()

import asyncio

async def example():
    future = asyncio.Future()
    
    # Где-то в коде
    future.set_result(42)  # Сигнал завершения!
    
    # Теперь await вернет результат
    result = await future
    print(result)  # 42

asyncio.run(example())

Через исключение (set_exception)

async def error_example():
    future = asyncio.Future()
    
    # Завершение с ошибкой
    future.set_exception(ValueError('Ошибка!'))
    
    try:
        await future
    except ValueError as e:
        print(f'Поймали: {e}')

asyncio.run(error_example())

Практический пример: мониторинг завершения

import asyncio
import time

async def worker(task_id):
    print(f'Task {task_id}: начало')
    await asyncio.sleep(task_id * 0.1)
    print(f'Task {task_id}: завершена')
    return task_id * 100

async def monitor_tasks():
    tasks = [asyncio.create_task(worker(i)) for i in range(1, 4)]
    
    # Способ 1: Ждем все задачи
    results = await asyncio.gather(*tasks)
    print(f'Результаты: {results}')
    
    # Способ 2: Ждем завершение в порядке готовности
    tasks = [asyncio.create_task(worker(i)) for i in range(1, 4)]
    for future in asyncio.as_completed(tasks):
        result = await future
        print(f'Готов результат: {result}')
    
    # Способ 3: Проверяем вручную
    task = asyncio.create_task(worker(5))
    while not task.done():
        print('Задача еще выполняется...')
        await asyncio.sleep(0.1)
    print(f'Задача завершена: {task.result()}')

asyncio.run(monitor_tasks())

На уровне CPython

Цикл событий использует selector (epoll на Linux, kqueue на macOS):

import selectors

# Цикл событий использует selector для отслеживания:
# 1. I/O событий (файлы, сокеты)
# 2. Таймеров (asyncio.sleep)
# 3. Callback-функций

selector = selectors.DefaultSelector()
# Регистрирует файловые дескрипторы
selector.register(fd, selectors.EVENT_READ, callback=on_ready)

# В цикле:
for key, mask in selector.select(timeout=0.01):
    key.callback(key.fileobj)  # Вызывает callback

Выводы

1. Task обертка - корутина оборачивается в Task (подкласс Future)
2. Callback-функции - регистрируются через add_done_callback()
3. Future сигналы - set_result() или set_exception() уведомляют цикл
4. Selector отслеживает - I/O и таймеры через epoll/kqueue
5. Проверка состояния - .done() и await возвращают управление
6. Цикл событий постоянно проверяет готовность и вызывает callback-функции