Асинхронний Python: види конкурентності

З виходом останніх версій Python3 все частіше можна почути дискусії про асинхронність і конкурентність, адже нещодавно Python став їх підтримувати. Насправді це не так. Асинхронність присутня в Python вже давно. Також багато новачків думають, що asyncio це єдиний/найкращий шлях для написання конкурентного коду. Ні. В ці статті ми опишемо декілька способів досягти асинхронного виконання і покажемо їх плюси і мінуси.

Термінологія

Перед тим як почати, слід визначитися з термінами і зрозуміти різницю між ними.

Синхронність vs асинхронність

При синхронному виконанні завдання виконуються по черзі, одне після іншого. При асинхронному ж, операції виконуються незалежно одна від іншої.

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

Конкурентність vs паралельність

Конкурентність - це коли в одиницю часу може виконуватися лише одна задача. В минулому прикладі, коли ми робили асинхронний дзвінок, це була конкурентність.

Але якби ми попросили забронювати білети свого друга, в той час як би ми писали листа, то це була паралельність, тобто дві задачі виконуються одночасно.

Де-факто паралельність - одна з форм конкурентності (в Python). Але паралелізм ще залежить від вашого заліза. Наприклад, якщо ваш процесор має лише одне ядро, то справжнього паралелізму досягти не вдасться, процеси будуть просто по черзі використовувати процесор (дуже малими проміжками, так, наче вони виконуються одночасно). Але ж якщо у вас багатоядерний процесор, то ви можете запустити 2 або більше (залежить від кількості ядер) потоків одночасно.

Короткий підсумок

• Синхронність: блокуючі операції.

• Асинхронність: неблокуючі операції.

• Конкурентність: одна задача виконується в даний момент.

• Паралельність: в даний момент може виконуватися декілька задач.

Паралелізм завжди є конкурентністю, але не навпаки.

Потоки і процеси

Python вже давно підтримує потоки. Вони дозволяють запускати завдання конкурентно. Але є проблема у вигляді Global Interpreter Lock (GIL), який не дозволяє одночасно виконувати більше одного потоку. Для використання декількох ядер на Python слід використовувати multiprocessing.

Потоки

Давайте розглянемо невеличкий приклад. В наступному коді функція worker буде запускатися в декількох потоках, асинхронно і конкурентно.

import threading
import time
import random


def worker(number):
    sleep = random.randrange(1, 10)
    time.sleep(sleep)
    print("Я Worker {}, я спав {} секунд".format(number, sleep))


for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    t.start()

print("Всі потоки запущено, давайте почекаємо поки вони виконаються!")

Ось такий можливий вивід:

$ python thread_test.py
Всі потоки запущено, давайте почекаємо поки вони виконаються!
Я Worker 1, я спав 1 секунд
Я Worker 3, я спав 4 секунд
Я Worker 4, я спав 5 секунд
Я Worker 2, я спав 7 секунд
Я Worker 0, я спав 9 секунд

Як бачите, ми запускаємо 5 потоків. І кожен наступний стартує відразу після запуску попереднього, а не чекає його завершення. Тобто вони запускаються асинхронно і конкурентно.

Якщо хочете дізнатися про потоки більше, ось офіційна документація.

Global Interpreter Lock (GIL)

GIL був доданий до CPython щоб полегшити управління пам'яттю та полегшити інтеграцію розширень на C++. GIL - механізм блокування, що не дозволяє виконувати декілька потоків одночасно.

Короткі факти про GIL:

• В одиницю часу може працювати лише один поток.

• Інтерпретатор перемикається між робочими потоками для реалізації конкурентності.

• GIL є тільки в CPython, в інших реалізаціях (Jython чи IronPython) він відсутній.

• GIL робить однопоточні програми швидшими.

• Для I/O-залежних операцій він майже не наносить шкоди.

• Завдяки йому ми можемо легко інтегрувати потоконебезпечний код на С[++].

• Для CPU-залежних задач інтерпретатор кожні N тактів перемикається між потоками. Тобто один поток не блокує іншого.

Багато людей бачать в GIL слабке місце Python'у, я ж бачу в ньому те, що дозволило Python зайняти своє місце в науковій спілноті, завдяки портам бібліотек NumPy та SciPy.

Більше інформації про GIL тут.

Процеси

Щоб досягти паралельності виконання в Python існує модуль multiprocessing, що надає API схожий до того, що ви бачили перед цим в модулі threading. В попередньому прикладі нам достатньо замінити Thread на Process.

import multiprocessing
import time
import random


def worker(number):
    sleep = random.randrange(1, 10)
    time.sleep(sleep)
    print("Я Worker {}, я спав {} секунд".format(number, sleep))


for i in range(5):
    t = multiprocessing.Process(target=worker, args=(i,))
    t.start()

print("Всі процеси запущено, давайте почекаємо поки вони виконаються!")

Що змінилося? Тепер замість потоків ми використовуємо процеси, що використовують різні ядра вашого CPU (якщо вони в вас є).

З класом Pool ми можемо викликати одну функцію з різними аргументами ще легше. Ось приклад з документації:

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    p = Pool(5)
    print(p.map(f, [1, 2, 3]))

Тут замість ітерації по списку аргументів і виклику f, ми запускаємо всі процеси за раз. Для кращого ознайомлення: клік.

Модуль concurrent.futures

Цей модуль надає дуже круту штуку для написання асинхронного коду. Моїми улюбленими є ThreadPoolExecutor та ProcessPoolExecutor. Вони представляють собою пул процесів або потоків. Ми додаємо наші завдання до них, а вони виконуються в доступних потоках/процесах. Нам же повертається об'єкт Future, що представляє собою результат операції, що ще не виконалася.

Ось приклад використання ThreadPoolExecutor:

from concurrent.futures import ThreadPoolExecutor
from time import sleep
 
def return_after_5_secs(message):
    sleep(5)
    return message
 
pool = ThreadPoolExecutor(3)
 
future = pool.submit(return_after_5_secs, ("hello"))
print(future.done())
sleep(5)
print(future.done())
print(future.result())

Для подальшого ознайомлення:

• 1

• 2

Asyncio — чому, як і навіщо?

Ви, напевно, задаєтеся тими ж питаннями що і більшість Python-спільноти: навіщо нам ще один вид асинхронності? Хіба процесів і потоків замало? Давайте ж розберемося!

Чому нам потрібен asyncio?

Створення процесів коштує дорого. Тому для I/O більше підходять потоки. Також ми знаємо що швидкість I/O-операцій залежать від багатьох факторів — повільний диск або нестабільна мережа роблять I/O-операції довшими. Давайте уявимо, що в нас є 3 потоки, що роблять різні I/O-операції. Назвемо їх Т1, Т2 та Т3. Ми запустили їх. Т3 готовий до виконання, а Т2 та Т1 чекають на I/O. Інтерпретатор спочатку перемикається на Т1, бачить що він чекає, перемикається на Т2, він теж чекає, і тільки тоді він перемикається на Т3, який виконає свій код. Ви бачите проблему?

Т3 був готовий, але інтерпретатору спочатку треба було перемкнутися на Т1 та Т2, що спричиняє марні витрати. Куди краще коли б він разу перемкнувся на Т3, чи не так?

Що таке asyncio?

Asyncio надає цикл подій та ще деякі круті фічі. Цикл подій реагує на різні I/O-події та перемикається на завдання, що можуть виконуватися і призупиняє ті, що чекають на I/O. Тобто ми не витрачаємо час на завдання, що ще не готові виконуватися.

Ідея дуже проста. У нас є цикл подій (event loop). А ще в нас є асинхронні функції, I/O-операції. Ми передаємо свої функції до циклу подій, щоб він запустив їх. Цикл подій повертає нам обєкт Future. Можна сказати, що це обіцянка, що ми отримаємо якісь дані в майбутньому. Ми зберігаємо його і час від часу перевіряємо чи не має наш Future результату виконання. І якщо так, то використовуємо ці дані для подальшої обробки.

Щоб зупиняти та відновлювати завдання asyncio використовує генератори та співпрограми (generators and coroutines). Більше про asyncio:

• Один

• Два

Як використовувати asyncio

Давайте відразу почнемо з прикладу:

import asyncio
import datetime
import random


async def my_sleep_func():
    await asyncio.sleep(random.randint(0, 5))


async def display_date(num, loop):
    end_time = loop.time() + 50.0
    while True:
        print("Loop: {} Time: {}".format(num, datetime.datetime.now()))
        if (loop.time() + 1.0) >= end_time:
            break
        await my_sleep_func()


loop = asyncio.get_event_loop()

asyncio.ensure_future(display_date(1, loop))
asyncio.ensure_future(display_date(2, loop))

loop.run_forever()

Зауважте, що ключові слова async та await доступні тільки з Python 3.5+. Давайте детальніше розглянемо код:

• Ми маємо асинхронну функцію display_date, що приймає число-індентифікатор та цикл подій.

• Функція має безкінечний цикл, що переривається через 50 секунд. Але поки 50 секунд не минуло, вона друкує час і засинає на випадкову кількість секунд. Ключове слово await вказує, що під час виконання функції, що стоїть після нього, можна перемкнутися на іншу асинхронну функцію (співпрограму).

• Ми додаємо функції до циклу подій за допомогою функції ensure_future.

• Ми запускаємо наш цикл подій.

Як зробити правильний вибір

Ми пройшлися по найпопулярніших формах конкурентності. Але тепер постає питання: яку з них використовувати? Залежить від задачі. Я можу порадити вам вирішувати за таким алгоритмом:

if io_bound:
    if io_very_slow:
        print("Use Asyncio")
    else:
       print("Use Threads")
else:
    print("Multi Processing")

• CPU-залежна задача => Multi Processing

• I/O-залежна, швидкий I/O => Multi Threading

• I/O-залежна, повільний I/O => Asyncio