Как используется генератор в библиотеке asyncio?

Роль генераторов в asyncio

Генераторы — это фундамент асинхронного программирования в Python. asyncio построена на механизме генераторов и корутин для реализации cooperative multitasking без OS threads.

Историческая справка

До async/await (Python 3.4)

Корутины писались как генераторы с @asyncio.coroutine:

import asyncio

# Старый стиль (Python 3.4-3.6)
@asyncio.coroutine
def fetch_data():
    # yield from это yield для генератора
    response = yield from asyncio.sleep(1)
    return response

# Это был генератор, вызывающий другие генераторы

Система работала так:

• Функция с yield from это генератор
• Event loop запускал генератор пока тот не выполнялся
• При yield from asyncio.sleep(1) генератор паузировался
• Event loop переключался на другие корутины
• Когда sleep завершалось, loop возобновлял генератор

Как asyncio использует генераторы сейчас

1. Под капотом async/await

Современный async/await синтаксис — это синтаксический сахар над генераторами:

# Это
async def fetch_data():
    await asyncio.sleep(1)
    return "data"

# Компилируется примерно вот в это (внутри):
def fetch_data():
    yield from asyncio.sleep(1)  # Это генератор
    return "data"

Проверим это в CPython:

import asyncio
import inspect

async def my_coroutine():
    await asyncio.sleep(1)

# Это на самом деле генератор
print(inspect.iscoroutinefunction(my_coroutine))  # True
print(inspect.isgeneratorfunction(my_coroutine))  # False (но похож)

# Но это объект который похож на генератор
coro = my_coroutine()
print(type(coro))  # <class coroutine>
print(inspect.iscoroutine(coro))  # True

# Внутри у него есть send() и throw() как у генератора
print(hasattr(coro, "send"))  # True
print(hasattr(coro, "throw"))  # True

2. Event Loop работает с генераторами

Event loop это просто умный диспетчер генераторов:

# Упрощённая демонстрация как работает asyncio
class SimpleEventLoop:
    def __init__(self):
        self.tasks = []  # Очередь генераторов
        self.ready = []  # Готовые к запуску
    
    def create_task(self, coro):
        self.tasks.append(coro)
    
    def run_until_complete(self, coro):
        self.tasks.append(coro)
        
        while self.tasks or self.ready:
            # Берём генератор из очереди
            task = self.tasks.pop(0) if self.tasks else self.ready.pop(0)
            
            try:
                # Запускаем генератор на один шаг
                # send(None) = вызов next(generator)
                awaitable = task.send(None)
                
                # Если генератор выдал что-то (await ...)
                # это значит он хочет ждать
                if isinstance(awaitable, SimpleAwaitable):
                    # Регистрируем callback когда awaitable завершится
                    awaitable.add_done_callback(
                        lambda: self.ready.append(task)
                    )
            
            except StopIteration as e:
                # Генератор закончился - это return значение
                print(f"Task completed with result: {e.value}")

class SimpleAwaitable:
    def __init__(self):
        self.callbacks = []
    
    def add_done_callback(self, fn):
        self.callbacks.append(fn)

# Использование
async def example():
    print("Start")
    await asyncio.sleep(0.1)  # Это await выполняет send()
    print("End")

loop = SimpleEventLoop()
loop.run_until_complete(example())

3. Tasks и Futures — обёртки над генераторами

import asyncio

async def my_task():
    await asyncio.sleep(1)
    return 42

# Task это генератор обёрнутый в специальный класс
task = asyncio.create_task(my_task())
print(type(task))  # <class asyncio.tasks.Task>

# Task наследуется от Future
print(isinstance(task, asyncio.Future))  # True

# Внутри Task есть _coro (корутина = генератор)
print(hasattr(task, "_coro"))  # True

# Task имеет callbacks которые вызываются когда генератор завершится
print(hasattr(task, "_callbacks"))  # True

Практические примеры использования генераторов

1. Создание своей awaitable (на основе генератора)

import asyncio
from typing import Generator

class CustomAwaitable:
    def __init__(self, delay: float):
        self.delay = delay
    
    def __await__(self) -> Generator:
        """
        __await__ должен вернуть генератор
        Это то что на самом деле yield-ит в asyncio
        """
        # Это генератор
        yield asyncio.sleep(self.delay)
        # После того как sleep завершится
        return "Custom awaitable finished"

async def main():
    result = await CustomAwaitable(1.0)
    print(result)  # Custom awaitable finished

asyncio.run(main())

2. Асинхронные генераторы

async for работает с генераторами которые async:

import asyncio

# Асинхронный генератор (содержит await)
async def async_generator():
    for i in range(5):
        await asyncio.sleep(0.1)
        yield i  # Это yield в async контексте

async def main():
    # async for работает с генератором
    async for value in async_generator():
        print(value)  # 0, 1, 2, 3, 4

asyncio.run(main())

3. Контекстные менеджеры (async with)

class AsyncResource:
    async def __aenter__(self):
        print("Acquiring resource")
        await asyncio.sleep(0.1)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Releasing resource")
        await asyncio.sleep(0.1)

async def main():
    async with AsyncResource() as resource:
        print("Using resource")
        await asyncio.sleep(0.1)

# Это синтаксический сахар над генератором в __aenter__ и __aexit__
asyncio.run(main())

4. Цепочка корутин через генераторы

async def step1():
    await asyncio.sleep(0.1)
    return "Step 1 done"

async def step2(prev_result):
    await asyncio.sleep(0.1)
    return f"{prev_result} -> Step 2 done"

async def step3(prev_result):
    await asyncio.sleep(0.1)
    return f"{prev_result} -> Step 3 done"

async def pipeline():
    # Каждый await это send() в генератор
    result = await step1()  # Генератор паузируется
    result = await step2(result)  # Генератор паузируется
    result = await step3(result)  # Генератор паузируется
    return result

async def main():
    result = await pipeline()
    print(result)
    # Step 1 done -> Step 2 done -> Step 3 done

asyncio.run(main())

Внутреннее устройство

Машина состояний генератора в asyncio

# Генератор имеет состояния
def sample_generator():
    print("State 1: Start")
    value = yield "pause point 1"
    print(f"State 2: Received {value}")
    yield "pause point 2"
    print("State 3: End")

gen = sample_generator()
print(next(gen))  # State 1: Start, выводит: pause point 1
print(gen.send("hello"))  # State 2: Received hello, выводит: pause point 2

# asyncio делает то же самое но с await вместо yield

Как await работает с генератором

import asyncio

async def outer():
    result = await inner()  # Это вызывает send() на inner
    return result

async def inner():
    await asyncio.sleep(1)  # Тут генератор "паузируется"
    return 42

# Примерно эквивалентно:
def outer_generator():
    result = yield from inner_generator()  # yield from = await
    return result

def inner_generator():
    yield asyncio.sleep(1)  # yield = pause point
    return 42

Оптимизация с генераторами

# Проблема: создание слишком много Tasks
async def slow_approach():
    # Каждый create_task это новая обёртка генератора
    tasks = [asyncio.create_task(fetch_url(url)) for url in urls]
    return await asyncio.gather(*tasks)

# Решение: использовать gather напрямую без Tasks
async def fast_approach():
    # gather сам управляет генераторами эффективнее
    return await asyncio.gather(
        *(fetch_url(url) for url in urls)
    )

Ключевые понимания

1. async/await = генераторы с синтаксическим сахаром
2. Event loop = диспетчер генераторов (запускает send() на каждом)
3. await = yield from для корутин
4. Cooperative multitasking = генератор добровольно паузируется
5. No threads = генераторы переключаются в одном потоке

Без генераторов asyncio не существовала бы.