Асинхронные возможности Python

Асинхронность:

• Прямая (параллелизм) — в языке нет
• С последовательной активацией (обратные вызовы функций, callbacks) — полностью моделируется имеющимся синтаксисом
• Сопрограммная (внутри сопрограммы есть синхронные участки, выполнение их произвольно) — вот!

Модель

Предполагается, что весь предлагаемый код вы запускаете и смотрите на результат; без этого понять намного сложнее, if even possible ☺

• Как работает yield from (повторение)

     1 def subr():
     2     yield "One"
     3     yield "Two"
     4 
     5 def task():
     6     for i in range(3):
     7         yield from subr()
     8         yield f"{i} Pass"
     9 
    10 for res in task():
    11     print(res)
  

  • На время yield from код генератора task() логически не исполняется, можно считать, что на это время его замещает subr()

• Ловля return из генератора с помощью yield from (два способа)

     1 def subr(n):
     2     yield f"One: {n}"
     3     yield f"Two: {n}"
     4     return f"Done: {n}"
     5 
     6 def task():
     7     for i in range(3):
     8         result = yield from subr(i)
     9         yield result
    10     return "*END*"
    11 
    12 core = task()
    13 try:
    14     while (res := next(core)):
    15         print(res)
    16 except StopIteration as E:
    17     print(E.value)
  

  • Оператор return в генераторе откладывает свой параметр в поле .value исключения StopIteration

  • А в конструкции yield from … это значение приезжает прямо так!

• Передача параметра в генератор с помощью .send() (повторение):

     1 def task(initial):
     2     value = initial
     3     while True:
     4         value = yield f"<{value * 2}>"
     5 
     6 core = task(100500)
     7 print(f"Start: {next(core)}")
     8 for i in range(5):
     9     print(core.send(i + 1))
  

  • Особенность: самый первый .send() должен быть генератор.send(None) (или, что то же самое, next(генератор), потому что в синтаксисе нет способа передать какое-то значение в начало генератора, а не в yield.

    • initial — это параметр генератор-функции, он передаётся в момент создания генератора, а не при его проходе

    • Мы договорились считать этот первый next() запуском генератора.

• Куда происходит .send() в случае yield from?

     1 def subr():
     2     x = yield "Wait for x"
     3     y = yield f"Wait for y ({x=})"
     4     return x * y
     5 
     6 def task():
     7     while True:
     8         value = yield from subr()
     9         _ = yield value
    10 
    11 core = task()
    12 print(next(core))
    13 for i in range(8):
    14     print(core.send(i))
  

  • Ничего неожиданного: .send() попадает в тот итератор, который сейчас yield-ит

  • Внимательно посмотрим, куда что send-илось…

Асинхронность как произвольное исполнение частей кода между yield-ами

• Понятие синхронного фрагмента — непрерывно выполняемого кода между yield-ами (а также стартом и return-ом)

• Понятие образующего цикла (main loop)

• Тот же пример, но с двумя асинхронно выполняющимися задачами:

     1 def subr(n):
     2     x = yield f"({n}) Wait for x"
     3     y = yield f"({n}) Wait for y ({x=})"
     4     return x * y
     5 
     6 def task(n):
     7     while True:
     8         value = yield from subr(n)
     9         _ = yield f"[{n}]: {value}"
    10 
    11 cores = task(0), task(1)
    12 print(next(cores[0]), next(cores[1]), sep="\n")
    13 for i in range(20):
    14     print(cores[not i % 3].send(i))
  

  • Здесь из образующего цикла поступает поток целых чисел, subr() их попарно умножает, а две задачи складывают эти произведения

  • Очередной число попадает в subr() выбранной задачи, а выбор задач делает образующий цикл

  • Синхронные фрагменты из task[0] выполняются в два раза чаще синхронных фрагментов из task[1]

  • Можно попробовать разобраться, что с чем складывалось…
• Более сложный пример: три конечных задачи с разным количеством синхронных фрагментов

     1 from random import randint
     2 
     3 def subr():
     4     x = yield
     5     y = yield
     6     return x * y
     7 
     8 def task(num):
     9     res = 0
    10     for i in range(num):
    11         res += yield from subr()
    12     return res
    13 
    14 def loop(*tasks):
    15     queue, result = list(tasks), []
    16     print("Start:", *queue, sep="\n\t")
    17     for task in tasks:
    18         next(task)
    19     while queue:
    20         task = queue.pop(0)
    21         try:
    22             task.send(randint(1, 9))
    23         except StopIteration as ret:
    24             result.append((task, ret.value))
    25         else:
    26             queue.append(task)
    27     return result
    28 
    29 print("Done:", *loop(task(10), task(3), task(5)), sep="\n\t")
  

  • Образующий цикл вынесен в отдельную функцию и стал сложнее. В нём генерируется непрерывный поток случайных целых и отдаётся поштучно на обработку очередному заданию. Если задание закончилось, запоминается его результат, а если нет — ставится в конец очереди.

  • Для реализации этой логики пришлось снова «вытащить» явную обработку StopIteration

  • Значения, возвращаемые yield, при этом не используются вообще: yield служит только для разметки синхронных фрагментов

• Если ещё усложнить логику образующего цикла, мы сможем управлять его поведением с помощью возвращаемых yield значений:

     1 from random import randint, choice
     2 from string import ascii_uppercase
     3 from collections import deque
     4 
     5 def subr():
     6     return (yield int) * (yield str)
     7 
     8 def task(num):
     9     res = ""
    10     for i in range(num):
    11         res += yield from subr()
    12     return res
    13 
    14 def loop(*tasks):
    15     queue, result = deque((task, None) for task in tasks), []
    16     print("Start:", *queue, sep="\n\t")
    17     while queue:
    18         task, request = queue.popleft()
    19         if request is int:
    20             data = randint(1, 4)
    21         elif request is str:
    22             data = choice(ascii_uppercase)
    23         else:
    24             data = request
    25         try:
    26             request = task.send(data)
    27         except StopIteration as ret:
    28             result.append((task, ret.value))
    29             task.close()
    30         else:
    31             queue.append((task, request))
    32     return result
    33 
    34 print("Done:", *loop(task(10), task(3), task(5)), sep="\n\t")
  

  • subr() возвращает тип параметра, который она хотела бы получить в следующем yield-е

  • Этот тип хранится в очереди вместо с заданием, чей subr() запросил данный параметр

  • Образующий цикл генерирует параметр сообразно типу

  • А ещё мы храним очередь в очереди, а не в списке, не надо привыкать к плохому!

Можно и дальше усложнять, но и так уже непросто!

Ещё модели

• Цикл событий: образующий цикл получает откуда-то «события», определяет, кто их должен обрабатывать и вызывает функции-обработчики с параметром обработчик(событие) (возможно, не функции, а генераторы облаботчик.send(событие), не слишком важно).

• Цикл обратных вызовов (callback-ов): частный случай того же самого: каждый обработчик «регистрируется» — по заранее определённому протоколу указывает, в каких случаях его надо вызывать (это и есть событие), а образующий цикл при наступлении события вызывает все обработчики, которые на нём зарегистрировались (опять-таки, можно организовать в виде функций, а можно в виде генераторов)

• Цикл с future: унификация управления образующим циклом

  • future — это генератор из двух синхронных сегментов

    1. Настройка и yield в образующий цикл

    2. return возвращаемого значения

  • Кроме того, в future есть поле готовности / результата

  • Алгоритм работы:
    1. Генератор-сервис заводит неготовую фьючу в данном образующем цикле

    2. Генератор-пользователь делает yield from фьюча

    3. Фьюча выпадает в образующий цикл
    4. В какой-то момент генератор-сервис выставляет в фьюче готовность / результат
    5. На этом основании образующий цикл возвращает управление фьюче (во второй сегмент)
    6. А та возвращает значение генератору-пользователю
  • Фактически это частный случай обратных вызовов
• … более сложная логика (например, приоритизация событий) …

Синтаксис Async

• async def + return ≈ генератор

• await ≈ yield from

• @types.coroutine: низкоуровневая сопрограмма, которая может делать и return значение, и yield

• async def + yield — это именно то, чем кажется: генераторы, про которые сразу известно, что они асинхронные:

  • Их можно проходить async for (причём в конструкторах вида [… async for i in асинхронный-гененратор …] тоже)

Перепишем предыдущий пример на async

•    1 from random import randint, choice
     2 from string import ascii_uppercase
     3 from types import coroutine
     4 from collections import deque
     5 
     6 @coroutine
     7 def subr():
     8     return (yield int) * (yield str)
     9 
    10 async def task(num):
    11     res = ""
    12     for i in range(num):
    13         res += await subr()
    14     return res
    15 
    16 def loop(*tasks):
    17     queue, result = deque((task, None) for task in tasks), []
    18     print("Start:", *queue, sep="\n\t")
    19     while queue:
    20         task, request = queue.popleft()
    21         if request is int:
    22             data = randint(1, 4)
    23         elif request is str:
    24             data = choice(ascii_uppercase)
    25         else:
    26             data = request
    27         try:
    28             request = task.send(data)
    29         except StopIteration as ret:
    30             result.append((task, ret.value))
    31             task.close()
    32         else:
    33             queue.append((task, request))
    34     return result
    35 
    36 print("Done:", *loop(task(10), task(3), task(5)), sep="\n\t")
  

  • Наш subr() использует прямое управление образующим циклом с помощью yield

    • В готовом инструментарии это практически никогда не нужно: и образующий цикл, и инструменты управления им должны входить в такой инструментарий

Asyncio

Базоавя документация

• Самое сложное — это логика образующего цикла

• Самое ненужное — это логика образующего цикла (достаточно знать, как он работает, а не что делает)

⇒

• Запрограммируем образующий цикл заранее, насуём туда инструментов

• Упростим протокол управления до одного понятия — Future

• Обмажем протокол верхним уровнем (задания, события, очереди и т. п.)

  • До такой степени, что ни одна из наших сопрограмм не делает yield (если это не асинхронный генератор)

• (asyncio specific) обмажем огромным количеством применений IRL

Основные понятия:

• Mainloop — образующий цикл. Полностью под капотом, мы его не видим.

• Task — асинхронное задание

     1 import asyncio
     2 from time import strftime
     3 
     4 async def late(delay, msg):
     5     await asyncio.sleep(delay)
     6     print(msg)
     7 
     8 async def main():
     9     print(f"> {strftime('%X')}")
    10     await late(1, "One")
    11     print(f"> {strftime('%X')}")
    12     await late(2, "Two")
    13     print(f"> {strftime('%X')}")
    14 
    15     task3 = asyncio.create_task(late(3, "Three"))
    16     task4 = asyncio.create_task(late(4, "Four"))
    17     await(task3)
    18     print(f"> {strftime('%X')}")
    19     await(task4)
    20     print(f"> {strftime('%X')}")
    21 
    22 asyncio.run(main())
  

  • asyncio.run(main()) — запуск «приложения» main() в образующем цикле asyncio()

  • «приложение» asyncio — корутина, который заполняет очередь mainloop-а и немножко командует им

  • Если просто написать await — корутина «просто запустится», в чём асинхроннотсь, непонятно (даже если она и выходила в mainloop)

  • В примере первая корутина спит секунду, а вторая — после этого ещё две

  • Если написать create_task(корутина), корутина регистрируется в mainloop-е, а возвращется нечто вроде фьючи — задание

  • await(здадание) запускает его

  • В примере ещё две корутины планируются одновременно, первая из них спит три секунды, а вторая — четыре, так что отрабатывает через секунду после первой

  • asyncio.sleep(тайм-аут) — это команда mainlop-у «верни мне управление после тайм-аута»

    • Чуть ли не единственная команда mainlop-у на поверхности

• Gather — атомарная операция create_task() / await над несколькими корутинами

     1 import asyncio
     2 
     3 async def late(delay, msg):
     4     await asyncio.sleep(delay)
     5     print(msg)
     6     return delay
     7 
     8 async def main():
     9     res = await asyncio.gather(
    10             late(3, "A"),
    11             late(1, "B"),
    12             late(2, "C"),
    13     )
    14     print(res)
    15 
    16 asyncio.run(main())
  

  • Тут всё понятно, запустились, повисели сколько сказано, завершились

  • (Python 3.11) группы заданий —запуск в виде контекстного менеджера

       1 import asyncio
       2 
       3 async def late(delay, msg):
       4     await asyncio.sleep(delay)
       5     print(msg)
       6     return delay
       7 
       8 async def main():
       9     async with asyncio.TaskGroup() as tg:
      10         tg.create_task(late(3, "A"))
      11         tg.create_task(late(1, "B"))
      12         tg.create_task(late(2, "C"))
      13     print("Done")
      14 
      15 asyncio.run(main())
    

• Синхронизация

  • Например события

       1 async def waiter(name, event):
       2     print(f'{name} waits for {event}…')
       3     await event.wait()
       4     print(f'…{name} got it!')
       5 
       6 async def eventer(wait, event):
       7     print(f"Emitting {event} in {wait} seconds")
       8     await asyncio.sleep(wait)
       9     print(f"Emitting {event}…")
      10     event.set()
      11 
      12 async def main():
      13     event = asyncio.Event()
      14     await asyncio.gather(
      15         waiter("One", event),
      16         waiter("Two", event),
      17         eventer(1, event))
      18 
      19 asyncio.run(main())
    

• Очереди

  •    1 async def ham(queue, size):
       2     for i in range(size):
       3         await asyncio.sleep(1)
       4         res = await queue.get()
       5         print(f"\tGot {res}")
       6 
       7 async def spam(wait, queue):
       8     for i in range(6):
       9         await asyncio.sleep(wait)
      10         val = f"{wait}:{i}"
      11         await queue.put(val)
      12         print(f"Put {val}")
      13 
      14 async def main():
      15     queue = asyncio.Queue()
      16     await asyncio.gather(
      17         ham(queue, 12),
      18         spam(0.4, queue),
      19         spam(1.6, queue))
      20 
      21 asyncio.run(main())
    

  • Есть и приоритетные очереди

И толстый-толстый слой шоколада!

• Параллелизм (внешний, следите за тредобезопасностью или не используйте треды)

• Изменение логики работы mainloop (aka Policies)

• Сеть (I/O, IPC и всё остальное), сигналы

• Потоки (над этим всем)

• Вброс/перехват исключений
• …

• Дикая туча модулей на основе asyncio

Д/З

1. Попробовать прочитать всю документацию и прощёлкать всё, до чего дотянетесь.

2. EJudge: FilterQueue 'Очередь с фильтром'

   Напишите класс FilterQueue со следующими свойствами:

   • Это потомок asyncio.Queue

   • В экземпляре класса атрибут очередь.window содержит первый элемент очереди, или None, если очередь пуста

   • С помощью операции фильтр in очередь можно определить, присутствуют ли в очереди такие элементы, что выражение фильтр(элемент) истинно

   • Метод .later() синхронно переставляет первый элемент очереди в её конец, или вызывает исключение asyncio.QueueEmpty, если очередь пуста

   • Метод .get() содержит необязательный параметр фильтр. Вызов очередь.get(фильтр) работает так:

     • Если в очереди нет элементов, на которых фильтр(элемент) истинно, работает как обычный .get().

     • Если в очереди есть элементы, на которых фильтр(элемент) истинно, переставляет первый элемент очереди в её конец до тех пор, пока фильтр(элемент) не истинно, а затем выполняет обычный .get().

   • Разрешается воспользоваться внутренним представлением Queue

   Input:

      1 async def putter(n, queue):
      2     for i in range(n):
      3         await queue.put(i)
      4 
      5 async def getter(n, queue, filter):
      6     for i in range(n):
      7         await asyncio.sleep(0.1)
      8         yield await queue.get(filter)
      9 
     10 async def main():
     11     queue = FilterQueue(10)
     12     asyncio.create_task(putter(20, queue))
     13     async for res in getter(20, queue, lambda n: n % 2):
     14         print(res)
     15 
     16 asyncio.run(main())
   

   Output:

   1
   3
   5
   7
   9
   11
   13
   15
   17
   4
   19
   12
   6
   16
   8
   14
   0
   10
   2
   18

3. EJudge: NotifyEvent 'Оповещения'

   Написать класс NotifyEvent (унаследованный от asyncio.Event) со следующими свойствами

   • В методе оповещение.set(имя) присутствует строка-имя адресата уведомления, но по умолчанию это None

   • Перед каждым оповещение.set() (кроме самого первого) требуется вызов оповещение.clear()

   • Метод await оповещение.wait() возвращает имя адресата уведомления (но в остальном работает как event.wait())

   Написать также сопрограмму task(имя, оповещение) со следующими свойствами:

   • Если уведомление «своё» — адресат уведомления совпадает с именем, — выводится имя, количество принятых «своих» уведомлений и количество принятых «чужих» уведомлений

   • Если вместо имени await оповещение.wait() вернул None, работа завершается

   • Использовать внутреннюю реализацию asyncio.Event в этой задаче нельзя

   Input:

      1 async def sender(names, notify):
      2     for name in names:
      3         notify.set(name)
      4         await asyncio.sleep(0.1)
      5         notify.clear()
      6     notify.set()
      7 
      8 async def main():
      9     notify = NotifyEvent()
     10     tasks = {n: task(n, notify) for n in "12"}
     11     targets = "1", "2", "2", "2", "2", "1", "2", "1", "1"
     12     await asyncio.gather(*(list(tasks.values()) + [sender(targets, notify)]))
     13 
     14 asyncio.run(main())
   

   Output:

   1: 1 / 0
   2: 1 / 1
   2: 2 / 1
   2: 3 / 1
   2: 4 / 1
   1: 2 / 4
   2: 5 / 2
   1: 3 / 5
   1: 4 / 5

TODO Тесты