async и await в Python: event loop и race

Проблем

Последната бройка, продадена два пъти

Разпродажбата е в разгара си. От едни слушалки остава една бройка. checkout проверява наличността, чака банката да потвърди плащането и чак тогава намалява наличността. Двама купувачи натискат „Купи“ в един и същи момент:

# Банката потвърждава плащането за 100 ms.
async def pay(buyer):
    await asyncio.sleep(0.1)


class Shop:
    def __init__(self):
        self.stock = 1
        self.sold = 0

    async def checkout(self, buyer):
        left = self.stock
        if left == 0:
            return
        await pay(buyer)
        self.stock = left - 1
        self.sold += 1


async def main():
    shop = Shop()
    await asyncio.gather(shop.checkout("Ана"), shop.checkout("Борис"))
    print(f"Продадени: {shop.sold}, наличност: {shop.stock}")


asyncio.run(main())

Продадени: 2, наличност: 0

Продадени са две, а бройката е една. Единият купувач ще получи имейл „Поръчката ви е изпратена“ за слушалки, които ги няма.

В asyncio това изглежда невъзможно: кодът върви в една нишка. В урока за race condition две нишки се сблъскваха на count += 1. Тук втора нишка няма.

Изходът е един и същ при всяко пускане: това не е лош късмет, а редът, в който върви кодът. За да го видим, трябва да знаем какво прави await.

Идеята

Една нишка и много чакания

Повечето работа в онлайн магазин е работа, която чака (виж работа, която чака, и работа, която смята): плащането чака банката, имейлът чака пощенския сървър, наличността чака склада. Нишка, която стои и чака, е похабена.

Event loop е една нишка и опашка от готови парчета код. Нишката взима следващото парче и го изпълнява докрай. Когато кодът стигне до чакане, той не държи нишката: казва „събуди ме, когато банката отговори“ и я пуска. Нишката взима следващото готово парче. Когато банката отговори, остатъкът от функцията (продължението) влиза в опашката и чака своя ред.

В Python event loop-ът е в asyncio. async def прави корутина: функция, която може да спре по средата. await е мястото, където спира и пуска нишката. asyncio.run(main()) пуска event loop-а и корутината main в него.

Три заявки към три услуги, пуснати наведнъж:

class Calls:
    def __init__(self):
        self.waiting = 0
        self.most = 0

    async def call(self, name, ms):
        self.waiting += 1
        self.most = max(self.most, self.waiting)
        await asyncio.sleep(ms / 1000)
        self.waiting -= 1
        print(f"{name}: готово след {ms} ms")


async def main():
    calls = Calls()
    await asyncio.gather(
        calls.call("плащане", 300),
        calls.call("имейл", 100),
        calls.call("склад", 200),
    )
    print(f"Чакаха наведнъж: {calls.most}")


asyncio.run(main())

имейл: готово след 100 ms
склад: готово след 200 ms
плащане: готово след 300 ms
Чакаха наведнъж: 3

asyncio.gather пуска трите корутини и чака всички. Всяка върви до await asyncio.sleep(...) и пуска нишката, затова и трите влизат в call, преди някоя да е свършила.

„Чакаха наведнъж: 3“: и трите заявки чакат едновременно, а нишката е само една. Всичко отнема около 300 ms (най-дългата), а не 600 (сборът). Нишката не е чакала нито една от тях.

Пробвай

Колко чакат наведнъж

Същото call, но първата заявка се чака сама:

Какво ще отпечата програмата?

async def main():
    calls = Calls()
    await calls.call("плащане", 300)
    await asyncio.gather(calls.call("имейл", 100), calls.call("склад", 200))
    print(f"Чакаха наведнъж: {calls.most}")


asyncio.run(main())

Отговор

плащане: готово след 300 ms
имейл: готово след 100 ms
склад: готово след 200 ms
Чакаха наведнъж: 2

плащане: готово след 300 ms
имейл: готово след 100 ms
склад: готово след 200 ms
Чакаха наведнъж: 3

имейл: готово след 100 ms
склад: готово след 200 ms
плащане: готово след 300 ms
Чакаха наведнъж: 3

Първото await calls.call(...) чака плащането докрай. Чак после gather пуска другите две наведнъж, затова най-много са 2. Цялото отнема около 500 ms.

await веднага след извикването значи „чакай това, преди да продължиш“. За да чакат заедно, първо ги пусни всичките, после чакай.

Виж го

Race между два await-а

Обратно към последната бройка. Ето как се преплитат двете поръчки:

class Shop:
    def __init__(self):
        self.stock = 1
        self.sold = 0

    async def checkout(self, buyer):
        left = self.stock
        if left == 0:
            return
        await pay(buyer)
        self.stock = left - 1
        self.sold += 1

Стъпка по стъпка, с думи

Остава една бройка. Две поръчки идват в един и същи момент: на Ана и на Борис.
checkout на Ана чете наличността: left е 1. На await пуска нишката, докато банката отговори.
Нишката е свободна, затова върви checkout на Борис. И той чете 1: Ана още не е записала нищо.
Банката отговаря на Ана. Кодът ѝ продължава след await: stock = left - 1 = 0, sold = 1. Дотук е вярно.
Борис продължава със старото left = 1: пак stock = 0, sold = 2. Една бройка, две продажби, и то с една нишка.

Между две await кодът върви без прекъсване. Но на await функцията пуска нишката, и през това време може да върви всичко друго. Когато продължи, светът може да е друг: left е стойност, прочетена преди чакането.

Това е същото „прочети, промени, запиши“ като при нишките, само че прекъсването не е където и да е, а точно на await. Затова и изходът е един и същ при всяко пускане: прекъсването не е случайно.

Идеята

Поправката: запази бройката преди await

Проверката и намаляването трябва да са едно парче код без await между тях. Поръчката запазва бройката веднага и чак тогава чака банката:

class Shop:
    def __init__(self):
        self.stock = 1
        self.sold = 0

    async def checkout(self, buyer):
        if self.stock == 0:
            return
        self.stock -= 1  # запази бройката преди await
        await pay(buyer)
        self.sold += 1

Продадени: 1, наличност: 0

Между if self.stock == 0 и self.stock -= 1 няма await, затова никоя друга корутина не може да влезе между тях. Борис вижда 0 и си тръгва.

Ако плащането се провали, бройката трябва да се върне. Това е упражнението накрая.

По-точно, за любопитните

Понякога цялото парче трябва да е заключено заедно с чакането: например само една поръчка наведнъж да пише в стар склад, който не понася две. Тогава трябва заключване, което разбира от await:

asyncio.Lock(): async with lock: пуска нишката, докато чака заключването. threading.Lock тук не става: той блокира цялата нишка, а с нея и event loop-а.

Как е в другите езици: в JavaScript и в asyncio нишката е една и парчето между две await върви само. В C# продълженията са в басейна от нишки, затова там трябва и lock. Java няма await: там същото е с CompletableFuture, а от Java 21 и с виртуални нишки, където пишеш обикновен код, който блокира, а JVM-ът сам сваля нишката, докато тя чака.

Пробвай

Намери грешката

Количката събира цените на три поръчки. Цената идва от склада с await. Трите поръчки струват по 25 €, а сборът излиза 25:

class Cart:
    def __init__(self):
        self.total = 0

    async def add_order(self, order_id):
        self.total += await price_of(order_id)


async def main():
    cart = Cart()
    await asyncio.gather(cart.add_order(1), cart.add_order(2), cart.add_order(3))
    print(f"Общо: {cart.total} €")


asyncio.run(main())

Общо: 25 €

Избери реда с грешката.

self.total += await price_of(order_id) прочита self.total преди await. И трите прочитат 0 и записват 0 + 25. Първо чакай, после събирай: price = await price_of(order_id), после self.total += price.

Грешката не е в нишките, а в реда: стойност, прочетена преди await, е стара след него.

Идеята

Не блокирай event loop-а

Event loop-ът работи, само ако никое парче код не държи нишката дълго. Отчетът за продажбите се прави с time.sleep(0.3): заспива, без await. В същото време купувач отваря страница на продукт, която чака склада 50 ms:

async def report():
    time.sleep(0.3)  # блокира: чака 300 ms без await
    print("Отчетът е готов")


async def product_page(buyer, start):
    await asyncio.sleep(0.05)
    print(f"{buyer} чака над 0,25 s: {time.monotonic() - start > 0.25}")


async def main():
    start = time.monotonic()
    await asyncio.gather(report(), product_page("Ана", start))


asyncio.run(main())

Отчетът е готов
Ана чака над 0,25 s: True

Страницата е чакала над 0,25 s вместо 50 ms. Докато отчетът държи нишката, никое друго продължение не може да върви: нито тази страница, нито плащания, нито таймери. На сървър с хиляди купувачи всички замръзват едновременно.

Поправката е тежката работа да не е в нишката на event loop-а. Тя отива в пул от нишки, а event loop-ът само чака резултата:

def build_report():
    time.sleep(0.3)


async def report():
    await asyncio.to_thread(build_report)
    print("Отчетът е готов")


async def product_page(buyer):
    await asyncio.sleep(0.05)
    print(f"{buyer}: страницата е готова")


async def main():
    await asyncio.gather(report(), product_page("Ана"))


asyncio.run(main())

Ана: страницата е готова
Отчетът е готов

await asyncio.to_thread(build_report) пуска функцията в пул от нишки и чака резултата, без да държи event loop-а. Ако работата смята (а не спи), заради GIL трябва пул от процеси: await asyncio.get_running_loop().run_in_executor(processes, build_report), където processes е ProcessPoolExecutor.

Пробвай

Попълни

Пусни трите заявки наведнъж и ги чакай заедно:

async def main():
    calls = Calls()
    await (
        calls.call("плащане", 300),
        calls.call("имейл", 100),
        calls.call("склад", 200),
    )
    print(f"Чакаха наведнъж: {calls.most}")


asyncio.run(main())

Упражнение

Запази, плати, върни

Реши задачата в редактора си.

Остават две бройки. Четирима купувачи поръчват наведнъж: Ана, Борис, Вера и Галя. Плащането отнема 100 ms. Картата на Борис е отказана: плащането хвърля изключение с текст „картата е отказана“.

checkout запазва бройката преди await, после плаща. Ако плащането падне, бройката се връща. Отпечатай резултата на всеки купувач по реда на поръчките, а накрая наличността и продадените:

Ана: платено
Борис: картата е отказана, бройката е върната
Вера: няма наличност
Галя: няма наличност
Наличност: 1, продадени: 1

Вера и Галя виждат „няма наличност“, защото бройката на Борис още е запазена, докато банката отговори. Това е честно: върнатата бройка ще купи следващият.

Още задачи

Опит отново. Ако купувач няма наличност, но друг ще върне бройка, нека чака до 200 ms и опита пак.
Таймаут. Банката понякога не отговаря. Ако плащането отнеме над 1 s, откажи поръчката и върни бройката. (asyncio.wait_for)
Отчет. Сметни сбора на продажбите (по 25 €) и го отпечатай накрая.

Тест

Провери се

1. В колко нишки върви твоят код в JavaScript (Node) и в asyncio на Python?В толкова, колкото са ядратаВ по една нишка за всяко awaitВ една: event loop-ът пуска парчетата код едно по едно

2. Три заявки чакат 100, 200 и 300 ms. Колко време общо, ако ги чакаш една по една с await, и колко с Promise.all, gather, Task.WhenAll или allOf?Около 600 ms и около 300 msОколо 600 ms и в двата случаяОколо 300 ms и в двата случая

3. В JavaScript и Python async функция чете общите данни, после прави await, после записва. Кога друг код може да ги промени?Никога: нишката е еднаМежду всеки два редаСамо докато тя чака на await

4. В обработката на заявка има сметка, която отнема 2 секунди, без нито един await. Какво става с другите заявки в Node или asyncio?Минават в друга нишкаЧакат 2 секунди: event loop-ът е заетНищо: async функцията върви отделно

5. Как да не се продаде последната бройка два пъти?Провери и намали наличността без await между тях, после плащай; при отказ върни бройкатаПусни поръчките в две нишкиСложи още един await преди записа

async не е нишка

Бисквитки