Практика · Асинхронність — базово

Лекція: lecture.html · Тест: quiz.html · Домашнє завдання: homework.html

Тут ми зміряємо секундоміром усе, що в лекції було намальовано смугами.

Що зробимо:

1. переконаємось, що виклик корутинної функції нічого не виконує;
2. побачимо на власному екрані, чому asyncio.run() у зошиті падає, і як із цим жити;
3. зміряємо послідовний await проти asyncio.gather — і звіримо числа з формулою «послідовно = сума пауз, асинхронно = найдовша пауза»;
4. підкладемо в корутину time.sleep і подивимось, як виграш зникає до нуля;
5. перепишемо все на asyncio.TaskGroup і зловимо ExceptionGroup через except*;
6. перевіримо, що на обчисленнях асинхронність не дає нічого.

Мережі тут не буде. Кожне «завантаження» — це asyncio.sleep: та сама пауза, що й у справжнього запиту, тільки без залежності від інтернету. Для механіки циклу подій різниці немає жодної.

0 · Що нам знадобиться

asyncio — частина стандартної бібліотеки, ставити нічого не треба. asyncio.TaskGroup і синтаксис except* зʼявились у Python 3.11 — перевіримо, що версія достатньо свіжа, щоб уся практика працювала.

import asyncio
import sys
import threading
import time

print("Python:", sys.version.split()[0])
# TaskGroup і except* потрібні в розділах 6 і 7 — краще дізнатись про це зараз
print("asyncio.TaskGroup доступний:", hasattr(asyncio, "TaskGroup"))
print("версія достатня (3.11+):", sys.version_info >= (3, 11))

Python: 3.12.2
asyncio.TaskGroup доступний: True
версія достатня (3.11+): True

1 · Виклик корутинної функції нічого не виконує

Найперша розбіжність зі звичайними функціями з теми 14: дужки після імені не запускають тіло. Вони створюють обʼєкт-корутину — рецепт, який ще ніхто не почав готувати.

Нижче ми викличемо завантажити("продажі") і навмисно не поставимо await. Якщо тіло виконається — ми побачимо рядок «почав». Подивись, чи побачимо.

async def завантажити(назва, пауза=0.4):
    """Імітує звернення до сховища: чекає `пауза` секунд і повертає назву великими."""
    print(f"   почав: {назва}")
    # asyncio.sleep — це очікування, під час якого цикл подій вільний
    await asyncio.sleep(пауза)
    print(f"   готово: {назва}")
    return назва.upper()


print("викликаємо завантажити('продажі') без await:")
рецепт = завантажити("продажі")
print("   тип того, що повернулось:", type(рецепт).__name__)
print("   у виводі вище немає рядка «почав» — тіло не виконувалось")

# корутину, яку так і не запустили, треба закрити явно,
# інакше Python на прибиранні сміття напише RuntimeWarning
рецепт.close()
print("   рецепт закрито, попередження не буде")

викликаємо завантажити('продажі') без await:
   тип того, що повернулось: coroutine
   у виводі вище немає рядка «почав» — тіло не виконувалось
   рецепт закрито, попередження не буде

2 · Чому asyncio.run() у зошиті падає

asyncio.run() створює новий цикл подій. Але зошит сам працює всередині циклу подій — і двох циклів в одному потоці бути не може.

Спіймаємо цю помилку й прочитаємо її текст. Це не поломка зошита: у звичайному файлі .py той самий рядок відпрацював би нормально.

корутина = завантажити("склад")
try:
    asyncio.run(корутина)
    print("несподівано: asyncio.run спрацював")
except RuntimeError as помилка:
    print("тип помилки:", type(помилка).__name__)
    print("текст:      ", помилка)
finally:
    # корутина так і не запустилась — закриваємо її, щоб не було попередження
    корутина.close()

print()
print("У файлі .py цей самий рядок працює: там циклу подій ще немає.")

тип помилки: RuntimeError
текст:       asyncio.run() cannot be called from a running event loop

У файлі .py цей самий рядок працює: там циклу подій ще немає.

3 · У зошиті await пишеться прямо в клітинці

Оскільки цикл подій уже запущений, нам не треба його створювати — треба лише віддати йому корутину. Робиться це звичайним await на верхньому рівні клітинки. Це особливість Jupyter та IPython; у файлі .py так писати не можна.

початок = time.perf_counter()
звіт = await завантажити("продажі")          # ось так — без asyncio.run
секунди = time.perf_counter() - початок

print("повернулось:", звіт)
print(f"тривало:     {секунди:.2f} c")

assert звіт == "ПРОДАЖІ", "корутина мала повернути назву великими літерами"
assert секунди >= 0.4, "пауза 0,4 c не могла минути швидше за 0,4 c"
print("✅ await у клітинці працює, пауза витримана")

   почав: продажі

   готово: продажі
повернулось: ПРОДАЖІ
тривало:     0.40 c
✅ await у клітинці працює, пауза витримана

4 · Як усе-таки запустити корутину «як у скрипті»

Інколи треба саме поведінка asyncio.run() — наприклад, щоб перевірити код, який поїде у файл. Вихід простий: дати корутині власний потік із власним циклом подій. Це не спосіб щось прискорити, це спосіб не конфліктувати з циклом зошита.

Функція нижче знадобиться нам далі не раз, тож напишемо її один раз і по-людськи.

def запустити(корутина):
    """Виконує корутину у власному циклі подій в окремому потоці.

    Потрібно тільки в Jupyter: у зошиті вже крутиться свій цикл, і asyncio.run()
    сюди не пускають. У звичайному .py-файлі замість цього пишуть asyncio.run().
    """
    скринька = {}

    def працівник():
        цикл = asyncio.new_event_loop()
        try:
            скринька["результат"] = цикл.run_until_complete(корутина)
        finally:
            цикл.close()

    потік = threading.Thread(target=працівник)
    потік.start()
    потік.join()          # чекаємо, поки потік добіжить до кінця
    return скринька["результат"]


print("перевіряємо запустити() на тій самій корутині:")
результат = запустити(завантажити("склад"))
print("повернулось:", результат)

assert результат == "СКЛАД", "запустити() має повертати те саме, що й await"
print("✅ той самий результат, тільки через власний цикл подій")

перевіряємо запустити() на тій самій корутині:
   почав: склад

   готово: склад
повернулось: СКЛАД
✅ той самий результат, тільки через власний цикл подій

5 · Послідовно проти gather: беремо секундомір

Тепер головний вимір усієї теми. Три «звіти», кожен чекає 0,4 секунди.

Спершу — послідовно, три await підряд. У лекції це верхня смуга першої схеми: очікування вишикувані одне за одним.

ЗВІТИ = ["продажі", "склад", "повернення"]
ПАУЗА = 0.4

async def послідовно():
    """Три await підряд: кожен чекає завершення попереднього."""
    зібрані = []
    for назва in ЗВІТИ:
        зібрані.append(await завантажити(назва, ПАУЗА))
    return зібрані


початок = time.perf_counter()
результат_послідовно = await послідовно()
час_послідовно = time.perf_counter() - початок

print()
print("результат:", результат_послідовно)
print(f"тривало:   {час_послідовно:.2f} c")
print(f"формула з лекції — сума пауз: {len(ЗВІТИ) * ПАУЗА:.2f} c")

   почав: продажі

   готово: продажі
   почав: склад

   готово: склад
   почав: повернення

   готово: повернення

результат: ['ПРОДАЖІ', 'СКЛАД', 'ПОВЕРНЕННЯ']
тривало:   1.20 c
формула з лекції — сума пауз: 1.20 c

Тепер те саме через asyncio.gather. Різниця в коді — один рядок, різниця в часі має бути втричі. Стеж за виводом: усі три «почав» надрукуються підряд, до першого «готово».

async def разом():
    """gather віддає всі три корутини циклу одразу, а потім чекає всіх."""
    return await asyncio.gather(
        завантажити(ЗВІТИ[0], ПАУЗА),
        завантажити(ЗВІТИ[1], ПАУЗА),
        завантажити(ЗВІТИ[2], ПАУЗА),
    )


початок = time.perf_counter()
результат_разом = await разом()
час_разом = time.perf_counter() - початок

print()
print("результат:", результат_разом)
print(f"тривало:   {час_разом:.2f} c")
print(f"формула з лекції — найдовша пауза: {ПАУЗА:.2f} c")
print(f"швидше в:  {час_послідовно / час_разом:.1f} рази")

   почав: продажі
   почав: склад
   почав: повернення

   готово: продажі
   готово: склад
   готово: повернення

результат: ['ПРОДАЖІ', 'СКЛАД', 'ПОВЕРНЕННЯ']
тривало:   0.40 c
формула з лекції — найдовша пауза: 0.40 c
швидше в:  3.0 рази

Звіряємо вимір із формулою

Лекція обіцяла дві речі: послідовно вийде сума пауз, асинхронно — найдовша з них. Перевіримо це assert-ами. Допуск беремо щедрий: запуск корутини й перемикання теж коштують часу, просто мікросекунди проти сотень мілісекунд.

очікуване_послідовно = len(ЗВІТИ) * ПАУЗА     # 3 × 0,4 = 1,2 c
очікуване_разом = ПАУЗА                        # найдовша пауза = 0,4 c

print(f"послідовно: модель {очікуване_послідовно:.2f} c, вимір {час_послідовно:.2f} c")
print(f"разом:      модель {очікуване_разом:.2f} c, вимір {час_разом:.2f} c")

assert очікуване_послідовно <= час_послідовно < очікуване_послідовно + 0.5, \
    "послідовний варіант має тривати приблизно суму пауз"
assert очікуване_разом <= час_разом < очікуване_разом + 0.5, \
    "gather має тривати приблизно найдовшу паузу, а не суму"
assert час_разом < час_послідовно / 2, "gather мав дати щонайменше дворазовий виграш"
print("✅ вимір збігся з формулою з лекції")

послідовно: модель 1.20 c, вимір 1.20 c
разом:      модель 0.40 c, вимір 0.40 c
✅ вимір збігся з формулою з лекції

Порядок результатів gather не залежить від порядку завершення

Це важлива дрібниця: gather повертає список у порядку аргументів. Тому розпакування працює передбачувано, навіть якщо третій звіт відповів першим.

Доведемо: дамо трьом задачам різні паузи, щоб вони гарантовано закінчились у зворотному порядку.

async def різні_паузи():
    return await asyncio.gather(
        завантажити("довгий", 0.30),    # закінчиться останнім
        завантажити("середній", 0.20),
        завантажити("швидкий", 0.10),   # закінчиться першим
    )


порядок = await різні_паузи()
print()
print("список від gather:", порядок)

assert порядок == ["ДОВГИЙ", "СЕРЕДНІЙ", "ШВИДКИЙ"], \
    "gather мав зберегти порядок аргументів, а не порядок завершення"
print("✅ порядок у списку — той, у якому ми передали корутини")

   почав: довгий
   почав: середній
   почав: швидкий
   готово: швидкий

   готово: середній

   готово: довгий

список від gather: ['ДОВГИЙ', 'СЕРЕДНІЙ', 'ШВИДКИЙ']
✅ порядок у списку — той, у якому ми передали корутини

6 · Пастка: time.sleep усередині корутини

А тепер зламаємо все одним рядком. Візьмемо ту саму корутину й замінимо await asyncio.sleep(пауза) на time.sleep(пауза).

Синтаксис лишається асинхронним: async def, await asyncio.gather, усе на місці. Але time.sleep спиняє потік, а потік один — і в ньому крутиться цикл подій.

Дивись на вивід: цього разу «почав» і «готово» йтимуть парами, бо друга задача навіть не стартує, поки перша не відпустить потік.

async def завантажити_блокуюче(назва, пауза=0.4):
    """Та сама корутина, але з time.sleep — синхронною паузою."""
    print(f"   почав: {назва}")
    time.sleep(пауза)          # ← ось цей рядок і є помилка
    print(f"   готово: {назва}")
    return назва.upper()


async def разом_але_блокуюче():
    return await asyncio.gather(
        завантажити_блокуюче(ЗВІТИ[0], ПАУЗА),
        завантажити_блокуюче(ЗВІТИ[1], ПАУЗА),
        завантажити_блокуюче(ЗВІТИ[2], ПАУЗА),
    )


початок = time.perf_counter()
результат_блокуючий = await разом_але_блокуюче()
час_блокуючий = time.perf_counter() - початок

print()
print(f"gather з asyncio.sleep: {час_разом:.2f} c")
print(f"gather з time.sleep:    {час_блокуючий:.2f} c")
print(f"виграш від async:       ×{час_послідовно / час_блокуючий:.1f}")

   почав: продажі

   готово: продажі
   почав: склад

   готово: склад
   почав: повернення

   готово: повернення

gather з asyncio.sleep: 0.40 c
gather з time.sleep:    1.20 c
виграш від async:       ×1.0

# результат той самий — розійшовся тільки час
assert результат_блокуючий == результат_разом, "дані мали лишитись ті самі"
# час зійшовся не з найдовшою паузою, а з їхньою сумою
assert час_блокуючий > 2 * час_разом, \
    "time.sleep мав звести виграш від gather нанівець"
assert час_блокуючий >= len(ЗВІТИ) * ПАУЗА, \
    "заблокований цикл змушує паузи лягти одна за одною"

print(f"три паузи по {ПАУЗА} c лягли одна за одною: {час_блокуючий:.2f} c")
print("✅ один синхронний виклик — і асинхронність зникла, хоч код лишився асинхронним")

три паузи по 0.4 c лягли одна за одною: 1.20 c
✅ один синхронний виклик — і асинхронність зникла, хоч код лишився асинхронним

Як полагодити, коли синхронний виклик прибрати не можна

Буває, що потрібна бібліотека просто не має асинхронної версії. Тоді блокуючий виклик віддають окремому потоку через asyncio.to_thread — і чекають його звичайним await, не спиняючи цикл.

async def завантажити_через_потік(назва, пауза=0.4):
    """Блокуючу паузу виносимо в окремий потік — цикл подій лишається вільним."""
    print(f"   почав: {назва}")
    await asyncio.to_thread(time.sleep, пауза)   # ← той самий time.sleep, але збоку
    print(f"   готово: {назва}")
    return назва.upper()


async def разом_через_потоки():
    return await asyncio.gather(
        завантажити_через_потік(ЗВІТИ[0], ПАУЗА),
        завантажити_через_потік(ЗВІТИ[1], ПАУЗА),
        завантажити_через_потік(ЗВІТИ[2], ПАУЗА),
    )


початок = time.perf_counter()
результат_потоки = await разом_через_потоки()
час_потоки = time.perf_counter() - початок

print()
print(f"той самий time.sleep, але через to_thread: {час_потоки:.2f} c")
print(f"для порівняння, напряму в корутині:        {час_блокуючий:.2f} c")

assert час_потоки < час_блокуючий / 2, "to_thread мав повернути накладання очікувань"
print("✅ виграш повернувся: цикл подій більше ніхто не тримає")

   почав: продажі
   почав: склад
   почав: повернення

   готово: продажі
   готово: склад
   готово: повернення

той самий time.sleep, але через to_thread: 0.40 c
для порівняння, напряму в корутині:        1.20 c
✅ виграш повернувся: цикл подій більше ніхто не тримає

7 · asyncio.TaskGroup — сучасний спосіб

З Python 3.11 те саме пишуть через async with asyncio.TaskGroup(). Задачі створюються всередині блоку, а вихід із блоку сам чекає на всіх — забути await неможливо. Результат кожної задачі беремо методом .result() після блоку.

async def разом_групою():
    """TaskGroup: створили задачі в блоці, на виході з блоку дочекались усіх."""
    async with asyncio.TaskGroup() as група:
        задача_а = група.create_task(завантажити(ЗВІТИ[0], ПАУЗА))
        задача_б = група.create_task(завантажити(ЗВІТИ[1], ПАУЗА))
        задача_в = група.create_task(завантажити(ЗВІТИ[2], ПАУЗА))
    # сюди потрапляємо лише тоді, коли всі три завершились
    return [задача_а.result(), задача_б.result(), задача_в.result()]


початок = time.perf_counter()
результат_групою = await разом_групою()
час_групою = time.perf_counter() - початок

print()
print("результат:", результат_групою)
print(f"тривало:   {час_групою:.2f} c   (gather давав {час_разом:.2f} c)")

assert результат_групою == результат_разом, "TaskGroup мав дати той самий список"
assert час_групою < час_послідовно / 2, "TaskGroup накладає очікування так само, як gather"
print("✅ той самий результат і той самий час — різниця тільки в записі")

   почав: продажі
   почав: склад
   почав: повернення

   готово: продажі
   готово: склад
   готово: повернення

результат: ['ПРОДАЖІ', 'СКЛАД', 'ПОВЕРНЕННЯ']
тривало:   0.41 c   (gather давав 0.40 c)
✅ той самий результат і той самий час — різниця тільки в записі

8 · Коли одна задача в групі падає

Домовимось: звіт із назвою "зламаний" кидає ValueError. Подивимось, як на це реагують два інструменти.

TaskGroup скасовує решту задач і випускає назовні ExceptionGroup — навіть якщо впала одна. Ловлять таку групу через except*.

Щоб скасування було видно, а не тільки заявлено, зробимо так: «зламаний» звіт падає через 0,1 c, а два здорових мали б завершитись аж через 0,5 c і дописати себе у список дійшли_до_кінця. Якщо їх справді скасували — список лишиться порожнім.

дійшли_до_кінця = []

async def завантажити_ризиковано(назва, пауза=0.4):
    """Те саме завантаження, але «зламаний» звіт кидає помилку."""
    await asyncio.sleep(пауза)
    if назва == "зламаний":
        raise ValueError(f"немає звіту «{назва}»")
    # цей рядок виконається, тільки якщо задачу не скасували раніше
    дійшли_до_кінця.append(назва)
    return назва.upper()


спіймані = []

async def група_із_поламаною():
    async with asyncio.TaskGroup() as група:
        група.create_task(завантажити_ризиковано("продажі", 0.5))
        група.create_task(завантажити_ризиковано("зламаний", 0.1))
        група.create_task(завантажити_ризиковано("склад", 0.5))


початок = time.perf_counter()
try:
    await група_із_поламаною()
    print("група відпрацювала без помилок")
except* ValueError as пачка:
    # у except* приходить не сам виняток, а група винятків
    print("тип того, що вилетіло:", type(пачка).__name__)
    for окрема in пачка.exceptions:
        спіймані.append(str(окрема))
        print("   не вдалося:", окрема)
час_групи = time.perf_counter() - початок

print(f"група протривала: {час_групи:.2f} c   (здорові задачі просили 0,50 c)")
print("задач дійшло до кінця:", дійшли_до_кінця)

assert len(спіймані) == 1, "мала впасти рівно одна задача"
assert "зламаний" in спіймані[0], "у тексті помилки має бути назва звіту"
assert дійшли_до_кінця == [], "решту задач мали скасувати, а не дочекатись"
assert час_групи < 0.4, "група мала завершитись одразу після падіння, а не через 0,5 c"
print("✅ ExceptionGroup спіймано через except*, і решту задач справді скасовано")

тип того, що вилетіло: ExceptionGroup
   не вдалося: немає звіту «зламаний»
група протривала: 0.10 c   (здорові задачі просили 0,50 c)
задач дійшло до кінця: []
✅ ExceptionGroup спіймано через except*, і решту задач справді скасовано

gather поводиться інакше. За замовчуванням перший виняток летить у місце виклику. А з return_exceptions=True нічого не летить: на місці невдалої задачі в списку лежить сам обʼєкт винятку, і ми самі вирішуємо, що з ним робити.

Це доречно тоді, коли часткова відповідь краща за жодну: два звіти з трьох — це все ще два звіти.

змішані = await asyncio.gather(
    завантажити_ризиковано("продажі", 0.2),
    завантажити_ризиковано("зламаний", 0.2),
    завантажити_ризиковано("склад", 0.2),
    return_exceptions=True,
)

вдалі = []
невдалі = []
for елемент in змішані:
    # виняток тут — звичайне значення в списку, тому перевіряємо тип
    if isinstance(елемент, Exception):
        невдалі.append(елемент)
    else:
        вдалі.append(елемент)

print("що прийшло списком:")
for елемент in змішані:
    print("  ", repr(елемент))
print()
print("вдалих:", вдалі)
print("невдалих:", [str(e) for e in невдалі])

assert вдалі == ["ПРОДАЖІ", "СКЛАД"], "два звіти мали дійти цілими"
assert len(невдалі) == 1 and isinstance(невдалі[0], ValueError), \
    "третій мав повернутись винятком, а не впасти вгору"
print("✅ часткова відповідь зібрана, програма не впала")

що прийшло списком:
   'ПРОДАЖІ'
   ValueError('немає звіту «зламаний»')
   'СКЛАД'

вдалих: ['ПРОДАЖІ', 'СКЛАД']
невдалих: ['немає звіту «зламаний»']
✅ часткова відповідь зібрана, програма не впала

9 · На обчисленнях асинхронність не дає нічого

Останній вимір — про головну межу з лекції. Замінимо очікування на справжню роботу процесора: додавання квадратів у циклі. Ніяких await усередині немає, і бути не може: цикл рахує й керування нікому не віддає.

Якби асинхронність «прискорювала програми», ми побачили б виграш. Подивимось.

Один нюанс виміру: обчислення, на відміну від очікувань, залежать від того, чим іще зайнята машина. Тому кожен варіант проженемо тричі й візьмемо найменший час — це найчистіший із трьох прогонів.

МЕЖА = 800_000

def сума_квадратів(межа):
    """Чиста робота процесора: жодного очікування, жодного місця для await."""
    підсумок = 0
    for число in range(межа):
        підсумок += число * число
    return підсумок


async def порахувати(межа):
    """Корутина, всередині якої немає жодного await — вона просто рахує."""
    return сума_квадратів(межа)


async def обчислити_послідовно():
    підсумки = []
    for _ in range(3):
        підсумки.append(await порахувати(МЕЖА))
    return підсумки


async def обчислити_разом():
    return await asyncio.gather(порахувати(МЕЖА), порахувати(МЕЖА), порахувати(МЕЖА))


async def найкращий_час(що_робити, спроб=3):
    """Повторює вимір кілька разів і повертає НАЙМЕНШИЙ час.

    Обчислення, на відміну від очікувань, дуже чутливі до того, чим зайнята
    машина. Найменший із кількох вимірів — найчистіший: він відповідає прогону,
    у якому нам найменше заважали.
    """
    найменший = None
    результат = None
    for _ in range(спроб):
        початок = time.perf_counter()
        результат = await що_робити()
        витрачено = time.perf_counter() - початок
        if найменший is None or витрачено < найменший:
            найменший = витрачено
    return найменший, результат


час_обчислень_послідовно, обчислення_послідовно = await найкращий_час(обчислити_послідовно)
час_обчислень_разом, обчислення_разом = await найкращий_час(обчислити_разом)

print(f"три обчислення послідовно:   {час_обчислень_послідовно:.2f} c")
print(f"три обчислення через gather: {час_обчислень_разом:.2f} c")
print(f"«виграш»: ×{час_обчислень_послідовно / час_обчислень_разом:.2f}")

три обчислення послідовно:   0.35 c
три обчислення через gather: 0.46 c
«виграш»: ×0.77

# результат, звісно, однаковий — питання лише в часі
assert обчислення_разом == обчислення_послідовно, "числа мали збігтись"

прискорення = час_обчислень_послідовно / час_обчислень_разом
# на очікуваннях gather давав ×3; тут маємо отримати приблизно ×1
assert прискорення < 1.6, \
    "на обчисленнях gather не має давати помітного прискорення"
assert час_разом < час_послідовно / 2, \
    "для порівняння: на очікуваннях виграш був щонайменше дворазовим"

print(f"на очікуваннях gather дав ×{час_послідовно / час_разом:.1f}")
print(f"на обчисленнях gather дав ×{прискорення:.2f}")
print("✅ асинхронність прибирає простій, а не додає рук: рахувати швидше вона не вміє")

на очікуваннях gather дав ×3.0
на обчисленнях gather дав ×0.77
✅ асинхронність прибирає простій, а не додає рук: рахувати швидше вона не вміє

Підсумок вимірів

Усе, що в лекції було намальовано смугами, ми щойно отримали секундоміром:

Що робили	Час	Висновок
три await підряд	≈ сума пауз	очікування вишикувані одне за одним
asyncio.gather	≈ найдовша пауза	очікування накладаються
asyncio.TaskGroup	те саме	сучасний запис того самого
time.sleep у корутині	≈ сума пауз	цикл заблоковано, виграш зник
asyncio.to_thread	≈ найдовша пауза	блокуючий виклик винесено вбік
обчислення через gather	без змін	накладати нічого

---

Завдання

🟢 Рівень 1

Напиши корутину перевірити_адресу(адреса, пауза), яка чекає пауза секунд і повертає рядок f"{адреса}: доступна". Запусти п'ять адрес із паузами 0,1 · 0,2 · 0,3 · 0,4 · 0,5 через asyncio.gather, зміряй час.

Зроблено, якщо проходить assert на те, що загальний час менший за 0,7 c (сума пауз — 1,5 c), і assert на те, що список результатів іде в порядку адрес, а не в порядку пауз.

🟡 Рівень 2

Візьми ту саму корутину й перепиши запуск на asyncio.TaskGroup. Додай шосту адресу, яка кидає ConnectionError. Злови ExceptionGroup через except* і надрукуй, скільки задач встигло завершитись до скасування.

Зроблено, якщо проходить assert на те, що спіймано рівно один ConnectionError, і assert на те, що загальний час менший за суму всіх пауз, — тобто скасування спрацювало, а не просто всі дочекались кінця.

🔴 Рівень 3

Побудуй табличку «кількість задач → час» для 1, 2, 4, 8, 16 і 32 задач із паузою 0,1 c кожна. Порахуй для кожного рядка накладні витрати: вимір − 0,1.

Зроблено, якщо:

• у табличці видно, що час майже не росте, поки задач стає більше;
• накладні витрати виміряні, а не переписані з лекції, і для 32 задач вони менші за 0,1 c;
• одним реченням пояснено, чому виграш ×32 тут не суперечить тому, що ядро одне.