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

Что нужно из теории

• Что такое и как работает yield from

• Как поймать значение из return значение в генераторе (res = yield from генератор или поле исключения StopIteration)

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

• await ≈ yield from

Использование asyncio

• Асинхронность в понимании asyncio:

  • TODO: нужно описать совсем_вводную демку про asyncio. Первая явно выделенная демка (см. ниже) уже не совсем вводная. ВОЗМОЖНО, это демка на основе первой врезки кода из раздела "Asyncio" со страницы лекции.
  • Образующий цикл не виден вообще

  • Сопрограмма — это обычный async def (то есть генератор-функция, но с дополнительными свойствами)

  • Передача управления образующему циклу:

    • await asyncio.sleep(время) (в том числе 0)

    • сеть

    • А вот yield используется для асинхронных итераторов, применяемых в async for

  • Сопрограммы нужно регистрировать в образующем цикле в виде заданий (.create_task())

  • Запуск образующего цикла: asyncio.run(сопрограмма)

• Отличие await сопрограмма() от await asyncio.create_task(сопрограмма)

• Запуск заданий с помощью .gather()

  • ручная барьерная синхронизация:

    • напишите сопрограмму squarer(), возвращающую квадрат параметра, и сопрограмму doubler(), возвращающую удвоенное значение параметра

    • по заданному списку из двух чисел [x, y] получите список [2*x^2, 2*y^2], вызвав при помощи .gather() сначала две сопрограммы squarer() на элементах этого списка, потом на элементах списка-результата две сопрограммы doubler()

    • выведите полученный в итоге список

  • Переписать решение с использованием TaskGroup() / async with

• Использование событий

  • цепочка событий:

    • напишите три сопрограммы:

      • snd() однократно отправляет событие evsnd

      • mid(k) ожидает событие evsnd и после получения отправляет событие evmid<k>, где <k> — это 0 или 1

      • rcv() ожидает событий evmid0, а затем evmid1

      • каждая сопрограмма

        • выводит после генерации события строку "<имя_сопрограммы>: generated <имя_события>",

        • а после получения события — строку "<имя_сопрограммы>: received <имя_события>"

    • запустите четыре задания: rcv(), mid(1), mid(0) и snd() (именно в таком порядке)

• Использование очереди

  • перекладывание из очереди в очередь

    • напишите три сопрограммы:

      • prod(): в цикле кладет в очередь q1 строку "value_<счетчик_цикла>", после каждой укладки ждет 1 секунду (через await asyncio.sleep())

      • mid(): в цикле ожидает получения элемента из очереди q1, кладет полученный элемент в очередь q2

      • cons(): в цикле ожидает получения элемента из очереди q2

      • цикл в prod() на 5 итераций, в остальных сопрограммах - бесконечный

      • каждая сопрограмма после укладки значения выводит строку "<имя_сопрограммы>: put <значение> to <имя_очереди>", а после получения значения - строку "<имя_сопрограммы>: got <значение> from <имя_очереди>"

    • запустите эти три сопрограммы
• Начать разбор второй задачи Д/З

Д/З

1. Задача_1: Перекладывание из очереди в очередь

   • написать три сопрограммы:

     1. writer(очередь, задержка) — каждые задержка секунд помещает в очередь строку номер_задержка, где номер растёт от 0 до бесконечности. При наступлении заранее заданного события завершает работу. Работа сопрограммы начинается с задержки.

     2. stacker(очередь, стек) — бесконечно перекладывает содержимое очереди в стек. При наступлении заранее заданного события завершает работу.

     3. reader(стек, количество, задержка) — снимает и выводит значение со стека каждые задержка секунд. После вывода количество значений посылает то самое событие и завершает работу. Работа сопрограммы начинается с задержки.

   • Считать через запятую четыре значения: задержка1, задержка2, задержка3, количество. Это, соответственно, задержки двух writer-ов и reader-а, и количество чтений.

     • In:

        2,3,1,10

     • Out:

       0_2
       0_3
       1_2
       1_3
       2_2
       3_2
       2_3
       4_2
       3_3
       5_2

2. Задача_2: сортировка слиянием

   • написать сопрограмму «слияние двух соседних упорядоченных отрезков» merge(A1, A2, start, middle, finish, event_in1, event_in2, event_out)

     • ждёт наступления обоих _in-событий (признак того, что отрезки отсортированы)

     • объединяет два упорядоченных отрезка A1[start…middle) и A1[middle…finish) в один A2[start…finish)

     • посылает событие event_out

   • Написать сопрограмму mtasks(A), в которой

     • Запланировать сначала объединение отрезков A единичной длины, помещая их в B, потом — отрезков B длины 2, помещая в A, потом длины 4 и т. д.

     • Должно получиться от N - 1 до N + log₂N заданий, зависимость между которыми задана событиями

     • Возвращается список заданий task, пригодный для одновременного запуска посредством asyncio.gather(*tasks), и список B, который в конце концов окажется отсортированным

   • In:

        1 import asyncio
        2 
        3 async def main(A):
        4     tasks, B = await mtasks(A)
        5     print(len(tasks))
        6     random.shuffle(tasks)
        7     await asyncio.gather(*tasks)
        8     return B
        9 
       10 
       11 random.seed(1337)
       12 A = random.choices(range(10), k=33)
       13 B = asyncio.run(main(A))
       14 print(*A)
       15 print(*B)
       16 print(B == sorted(A))
     

   • Out

     37
     6 5 3 5 1 8 3 7 9 3 9 2 6 9 1 8 3 0 6 3 4 8 8 6 8 5 4 9 7 3 6 9 4
     0 1 1 2 3 3 3 3 3 3 4 4 4 5 5 5 6 6 6 6 6 7 7 8 8 8 8 8 9 9 9 9 9
     True