asyncio и асинхронное программирование в Python: конкурентность, корутины, таски, async/await, event loop

def simple_generator():
    print("Начало")
    yield 1
    print("Продолжение")
    yield 2
    print("Конец")
    
gen = simple_generator()
next(gen)  # Выведет "Начало", вернет 1
next(gen)  # Выведет "Продолжение", вернет 2

# Упрощенная схема event loop
ready_tasks = []
waiting_tasks = []
 
while True:
    # Проверяем I/O события (сокеты, таймеры)
    check_io_events()
    
    # Берем готовую задачу
    if ready_tasks:
        task = ready_tasks.pop(0)
        try:
            task.step()  # Выполняем до следующего await
        except StopIteration:
            # Корутина завершилась
            pass
    
    # Если нечего делать, ждем I/O
    if not ready_tasks:
        wait_for_io_events()

import asyncio
 
async def fetch_data():
    print("Начинаем загрузку")
    await asyncio.sleep(2)
    print("Загрузка завершена")
    return {"data": "результат"}
 
# Просто вызов не выполняет функцию
coro = fetch_data()  # Получаем объект корутины
print(type(coro))  # <class 'coroutine'>
 
# Запуск через asyncio.run()
result = asyncio.run(fetch_data())

async def task_one():
    print("Task 1: старт")
    await asyncio.sleep(1)
    print("Task 1: финиш")
 
async def task_two():
    print("Task 2: старт")
    await asyncio.sleep(0.5)
    print("Task 2: финиш")
 
async def main():
    await asyncio.gather(task_one(), task_two())
 
asyncio.run(main())

async def background_task():
    for i in range(5):
        print(f"Фоновая задача: {i}")
        await asyncio.sleep(0.5)
 
async def main():
    # Создаем Task - корутина начнет выполняться сразу
    task = asyncio.create_task(background_task())
    
    # Делаем другую работу
    await asyncio.sleep(1)
    print("Основная работа выполнена")
    
    # Ждем завершения фоновой задачи
    await task
 
asyncio.run(main())

async def slow_operation():
    await asyncio.sleep(10)
    return "готово"
 
async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Операция заняла слишком много времени")
 
asyncio.run(main())

async def fetch_url(url, delay):
    await asyncio.sleep(delay)
    return f"Данные с {url}"
 
async def main():
    urls = [
        ("site1.com", 2),
        ("site2.com", 0.5),
        ("site3.com", 1)
    ]
    
    tasks = [fetch_url(url, delay) for url, delay in urls]
    
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"Получено: {result}")
 
asyncio.run(main())

async def failing_task():
    await asyncio.sleep(0.5)
    raise ValueError("Что-то пошло не так")
 
async def success_task():
    await asyncio.sleep(0.3)
    return "успех"
 
async def main():
    results = await asyncio.gather(
        failing_task(),
        success_task(),
        return_exceptions=True
    )
    
    for result in results:
        if isinstance(result, Exception):
            print(f"Ошибка: {result}")
        else:
            print(f"Результат: {result}")
 
asyncio.run(main())

async def simple_coro():
    return "результат"
 
coro = simple_coro()
print(type(coro))  # <class 'coroutine'>
print(dir(coro))   # методы: send, throw, close, __await__

async def inner():
    print("Inner: начало")
    await asyncio.sleep(0)  # Точка переключения
    print("Inner: завершение")
    return 42
 
async def outer():
    print("Outer: до await")
    result = await inner()
    print(f"Outer: получили {result}")
    return result
 
# Под капотом await превращается примерно в это:
# iter_obj = inner().__await__()
# value = None
# while True:
#     try:
#         value = iter_obj.send(value)
#     except StopIteration as e:
#         result = e.value
#         break

import dis
 
async def example():
    await asyncio.sleep(1)
    
dis.dis(example)
# Вывод покажет байткод с GET_AWAITABLE и YIELD_FROM

async def broken():
    await 42  # TypeError: object int can't be used in 'await' expression

class MyAwaitable:
    def __await__(self):
        # Возвращаем генератор, который один раз yield'ит и завершается
        yield  # Отдаем управление event loop
        return "кастомный результат"
 
async def use_custom():
    result = await MyAwaitable()
    print(result)  # "кастомный результат"
 
asyncio.run(use_custom())

async def sequential():
    result1 = await slow_operation()  # Ждем завершения
    result2 = await another_operation()  # Только потом запускаем это
    return result1 + result2
 
# Параллельная версия:
async def parallel():
    task1 = asyncio.create_task(slow_operation())
    task2 = asyncio.create_task(another_operation())
    result1 = await task1
    result2 = await task2
    return result1 + result2

# Упрощенная иллюстрация того, как работает event loop
class SimpleEventLoop:
    def __init__(self):
        self.ready_queue = []
        self.waiting_tasks = {}  # task: (event_type, event_data)
        self.selector = selectors.DefaultSelector()
    
    def run_until_complete(self, coro):
        task = Task(coro)
        self.ready_queue.append(task)
        
        while self.ready_queue or self.waiting_tasks:
            # Выполняем готовые задачи
            while self.ready_queue:
                task = self.ready_queue.pop(0)
                try:
                    # Продвигаем корутину до следующего await
                    task.step()
                except StopIteration as e:
                    # Задача завершилась
                    return e.value
            
            # Ждем I/O событий
            if self.waiting_tasks:
                events = self.selector.select(timeout=0.1)
                for key, mask in events:
                    task = key.data
                    self.ready_queue.append(task)
                    del self.waiting_tasks[task]

import asyncio
import time
 
async def greedy_task():
    # Плохой пример - блокирует event loop
    print("Начинаю долгие вычисления")
    start = time.time()
    result = sum(i[B]2 for i in range(10_000_000))
    print(f"Вычисления заняли {time.time() - start:.2f}с")
    return result
 
async def polite_task():
    # Хороший пример - периодически отдает управление
    print("Начинаю вежливые вычисления")
    start = time.time()
    result = 0
    for i in range(10_000_000):
        result += i[/B]2
        if i % 100_000 == 0:
            await asyncio.sleep(0)  # Передаем управление
    print(f"Вычисления заняли {time.time() - start:.2f}с")
    return result

async def background_worker():
    try:
        while True:
            await asyncio.sleep(1)
            print("Работаю в фоне")
    except asyncio.CancelledError:
        print("Получен сигнал остановки")
        # Очистка ресурсов
        raise
 
async def main():
    worker = asyncio.create_task(background_worker())
    await asyncio.sleep(5)
    worker.cancel()  # Явно отменяем фоновую задачу
    try:
        await worker
    except asyncio.CancelledError:
        print("Фоновая задача остановлена")

import asyncio
 
async def fetch_data():
    print("Загрузка началась")
    await asyncio.sleep(1)
    return "данные"
 
# Корутина создана, но не запущена
coro = fetch_data()
print(f"Тип: {type(coro)}")  # <class 'coroutine'>
[H2]Ничего не выведется - код не выполняется[/H2]
 
# Запуск через await
asyncio.run(coro)  # Теперь выполнится

async def background_work():
    for i in range(3):
        print(f"Фоновая работа: шаг {i}")
        await asyncio.sleep(0.5)
    return "завершено"
 
async def main():
    # Создаем Task - корутина запускается сразу
    task = asyncio.create_task(background_work())
    print("Task создан, продолжаем работу")
    
    # Делаем что-то еще, пока task работает
    await asyncio.sleep(1)
    print("Прошла секунда")
    
    # Теперь дожидаемся результата
    result = await task
    print(f"Результат: {result}")
 
asyncio.run(main())

async def demo_task_state():
    task = asyncio.create_task(asyncio.sleep(2))
    
    print(f"Сразу после создания: done={task.done()}")
    await asyncio.sleep(1)
    print(f"Через секунду: done={task.done()}")
    
    await task  # Дожидаемся завершения
    print(f"После завершения: done={task.done()}")
 
asyncio.run(demo_task_state())

async def cancellable_task():
    try:
        for i in range(10):
            print(f"Итерация {i}")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Задача отменена, очищаю ресурсы")
        raise  # Важно пробросить исключение дальше
 
async def main():
    task = asyncio.create_task(cancellable_task())
    await asyncio.sleep(2.5)  # Даем поработать
    task.cancel()  # Отменяем
    
    try:
        await task
    except asyncio.CancelledError:
        print("Отмена подтверждена")
 
asyncio.run(main())

def task_finished(task):
    if task.cancelled():
        print("Задача была отменена")
    elif task.exception():
        print(f"Задача упала с ошибкой: {task.exception()}")
    else:
        print(f"Задача вернула: {task.result()}")
 
async def main():
    task = asyncio.create_task(asyncio.sleep(1))
    task.add_done_callback(task_finished)
    await task
 
asyncio.run(main())

async def fetch(n):
    await asyncio.sleep(n * 0.1)
    return f"результат {n}"
 
async def main():
    # gather - простой способ
    results = await asyncio.gather(
        fetch(3), fetch(1), fetch(2)
    )
    print(results)  # ['результат 3', 'результат 1', 'результат 2']
    
    # wait - с контролем
    tasks = [asyncio.create_task(fetch(i)) for i in range(5)]
    done, pending = await asyncio.wait(
        tasks, 
        return_when=asyncio.FIRST_COMPLETED
    )
    print(f"Завершено: {len(done)}, в процессе: {len(pending)}")
    
    # Отменяем незавершенные
    for task in pending:
        task.cancel()
 
asyncio.run(main())

async def failing_task():
    await asyncio.sleep(0.1)
    raise ValueError("упс")
 
async def main():
    task = asyncio.create_task(failing_task())
    await asyncio.sleep(1)  # Task упал, но мы не заметили
    # Предупреждение появится только при завершении event loop
 
asyncio.run(main())

# Концептуальная схема работы планировщика
class Scheduler:
    def __init__(self):
        self.ready = deque()  # Готовые к выполнению
        self.sleeping = []    # Ожидают таймера
        self.waiting = {}     # Ожидают I/O
        
    def schedule(self, coro):
        """Добавить корутину в ready queue"""
        self.ready.append(coro)
    
    def run(self):
        while self.ready or self.sleeping or self.waiting:
            # Проверяем истекшие таймеры
            now = time.time()
            for timer in self.sleeping[:]:
                if timer.deadline <= now:
                    self.ready.append(timer.coro)
                    self.sleeping.remove(timer)
            
            # Проверяем готовые I/O события
            ready_io = check_io_events(timeout=0)
            for fd in ready_io:
                coro = self.waiting.pop(fd)
                self.ready.append(coro)
            
            # Выполняем одну готовую корутину
            if self.ready:
                coro = self.ready.popleft()
                result = coro.send(None)
                
                if isinstance(result, SleepRequest):
                    self.sleeping.append(Timer(result.duration, coro))
                elif isinstance(result, IORequest):
                    self.waiting[result.fd] = coro

# Реальный код из asyncio (упрощенно)
def _run_once(self):
    # Вычисляем timeout для select
    timeout = None
    if self._ready:
        timeout = 0  # Есть готовые задачи, не ждем
    elif self._scheduled:
        # Ждем до ближайшего таймера
        timeout = self._scheduled[0]._when - time.monotonic()
    
    # Ждем I/O события
    event_list = self._selector.select(timeout)
    
    # Обрабатываем готовые события
    for key, mask in event_list:
        callback, data = key.data
        callback(data)
    
    # Обрабатываем истекшие таймеры
    end_time = time.monotonic()
    while self._scheduled:
        handle = self._scheduled[0]
        if handle._when > end_time:
            break
        heapq.heappop(self._scheduled)
        handle._run()
    
    # Выполняем готовые callback'и
    ntodo = len(self._ready)
    for i in range(ntodo):
        handle = self._ready.popleft()
        handle._run()

async def fetch_data():
    await asyncio.sleep(1)
    return "данные"
 
async def wrong_way():
    result = fetch_data()  # Забыли await
    print(result)  # <coroutine object fetch_data at 0x...>
    
async def right_way():
    result = await fetch_data()  # Правильно
    print(result)  # "данные"

import requests  # Синхронная библиотека
import asyncio
 
async def bad_fetch(url):
    # Блокирует весь event loop на время запроса
    response = requests.get(url)
    return response.text
 
async def good_fetch(url):
    # Асинхронная версия
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def nested_call():
    await asyncio.sleep(1)
    
async def broken():
    # Ошибка: loop уже работает
    asyncio.run(nested_call())
    
async def correct():
    # Правильно: просто await
    await nested_call()

async def background_task():
    await asyncio.sleep(1)
    raise ValueError("что-то сломалось")
 
async def silent_failure():
    task = asyncio.create_task(background_task())
    await asyncio.sleep(2)
    # task упал, но мы не узнали
 
async def handled_failure():
    task = asyncio.create_task(background_task())
    try:
        await task
    except ValueError as e:
        print(f"Поймали ошибку: {e}")

async def cleanup_needed():
    connection = await db.connect()
    try:
        while True:
            await asyncio.sleep(1)
            # Работаем с соединением
    except asyncio.CancelledError:
        await connection.close()  # Обязательно закрываем
        raise

# Плохо
async def multiple_requests():
    for url in urls:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                data = await response.text()
 
# Хорошо
async def efficient_requests():
    async with aiohttp.ClientSession() as session:
        for url in urls:
            async with session.get(url) as response:
                data = await response.text()

# Синхронный код - идентифицируем I/O операции
import requests
import psycopg2
 
def fetch_user_data(user_id):
    # I/O операция 1: HTTP запрос
    response = requests.get(f"https://api.example.com/users/{user_id}")
    user_data = response.json()
    
    # I/O операция 2: запрос к базе
    conn = psycopg2.connect("dbname=test")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,))
    orders = cursor.fetchall()
    
    return {"user": user_data, "orders": orders}

# Асинхронная версия - заменили библиотеки
import aiohttp
import asyncpg
 
async def fetch_user_data(user_id):
    # Асинхронный HTTP запрос
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as response:
            user_data = await response.json()
    
    # Асинхронный запрос к базе
    conn = await asyncpg.connect("postgresql://localhost/test")
    orders = await conn.fetch("SELECT * FROM orders WHERE user_id = $1", user_id)
    await conn.close()
    
    return {"user": user_data, "orders": orders}

async def fetch_user_data_parallel(user_id):
    # Запускаем операции параллельно
    async def get_api_data():
        async with aiohttp.ClientSession() as session:
            async with session.get(f"https://api.example.com/users/{user_id}") as response:
                return await response.json()
    
    async def get_db_data():
        conn = await asyncpg.connect("postgresql://localhost/test")
        orders = await conn.fetch("SELECT * FROM orders WHERE user_id = $1", user_id)
        await conn.close()
        return orders
    
    # Выполняем одновременно
    user_data, orders = await asyncio.gather(
        get_api_data(),
        get_db_data()
    )
    
    return {"user": user_data, "orders": orders}

# Синхронный цикл
def fetch_multiple_users(user_ids):
    results = []
    for user_id in user_ids:
        data = fetch_user_data(user_id)  # Блокирует на каждой итерации
        results.append(data)
    return results
 
# Асинхронный вариант
async def fetch_multiple_users(user_ids):
    # Создаем список корутин
    tasks = [fetch_user_data(user_id) for user_id in user_ids]
    # Выполняем все параллельно
    results = await asyncio.gather(*tasks)
    return results

async def safe_database_operation():
    conn = await asyncpg.connect("postgresql://localhost/test")
    try:
        result = await conn.fetch("SELECT * FROM data")
        return result
    finally:
        await conn.close()  # Гарантированно закрываем
 
# Или с контекстным менеджером
async def safer_operation():
    async with asyncpg.create_pool("postgresql://localhost/test") as pool:
        async with pool.acquire() as conn:
            result = await conn.fetch("SELECT * FROM data")
            return result

async def robust_fetch(user_ids):
    tasks = [fetch_user_data(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Разделяем успешные и неудачные результаты
    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]
    
    return successes, failures

async def fetch_with_timeout(user_id):
    try:
        result = await asyncio.wait_for(
            fetch_user_data(user_id),
            timeout=5.0  # Максимум 5 секунд
        )
        return result
    except asyncio.TimeoutError:
        return None  # Или обработать по-другому

# Синхронная точка входа
def main():
    user_ids = [1, 2, 3, 4, 5]
    results = fetch_multiple_users(user_ids)
    process_results(results)
 
if __name__ == "__main__":
    main()
 
# Асинхронная точка входа
async def main():
    user_ids = [1, 2, 3, 4, 5]
    results = await fetch_multiple_users(user_ids)
    await process_results(results)  # Если process_results тоже асинхронная
 
if __name__ == "__main__":
    asyncio.run(main())

import concurrent.futures
 
async def process_heavy_data(data):
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        # Вычисления идут в отдельном процессе
        result = await loop.run_in_executor(
            pool, 
            cpu_intensive_function, 
            data
        )
    return result