За время менторства начинающих разработчиков заметил, что большинство вопросов связаны с темами: конкурентность, параллелизм, асинхронность. Их часто задают на собеседованиях, а в работе эти знания позволяют писать более эффективные и производительные системы. Поэтому я решил систематизировать свое видение и опыт в виде статьи.
Цель — c помощью примеров рассказать:
-
О потоках, процессах и корутинах.
-
О Global Interpreter Lock и накладываемых им ограничениях.
-
CPU и I/O bound нагрузке.
-
В чем польза от ThreadPool и ProcessPool.
Задача №1: Работа с JSON или CPU-bound task
Рассмотрим функцию:
# CPU bound task: generate list and dump it into JSON def work(size): json.dumps(list(range(size)))Она делает 2 вещи: генерирует список и превращает его в JSON. Подобный код — отличный пример CPU bound задачи — задачи, скорость выполнения которой зависит от мощности процессора.
Представим что нам нужно выполнить эту тяжеловесную задачу N раз. Последовательный алгоритм будет выглядеть так:
# Do work sequentially, one by one def sequential(size, count): for _ in range(count): work(size)Очевидно, что последовательное выполнение кода неэффективно. Какие варианты ускорить программу нам доступны?
Вариант №1 — создать по одному потоку на каждый вызов функции и попробовать добиться параллелизма:
# Do work using OS Threads def run_threads(size, executionUnitsCount): with concurrent.futures.ThreadPoolExecutor() as executor: executor.map(work, [size] * executionUnitsCount)Вариант №2 — вместо потоков создавать процесс:
# Do work using OS Processes def run_processes(size, executionUnitsCount): with concurrent.futures.ProcessPoolExecutor() as executor: executor.map(work, [size] * executionUnitsCount)В обоих случаях используются executor классы из пакета сoncurrent.futures. Если использовать пакеты multithreading и multiprocessing то код бы выглядел примерно так:
Hidden text
# Do work using OS Threads def run_threads(size, executionUnitsCount): threads = [threading.Thread(target=work, args=(size,)) for _ in range(executionUnitsCount)] for t in threads: t.start() for t in threads: t.join() # Do work using OS Processes def run_processes(size, executionUnitsCount): processes = [multiprocessing.Process(target=work, args=(size,)) for _ in range(executionUnitsCount)] for p in processes: p.start() for p in processes: p.join()Код тестирующий производительность всех реализаций:
if __name__ == '__main__': # Disable GC for better benchmarking and avoid pauses gc.disable() jsonSize = 1000000 testCases = [ (jsonSize, i) for i in range(1, 11)] variants = [sequential, run_threads, run_processes] for i, t in enumerate(testCases): size, executionUnitsCount = t print(f"Parallelism: {executionUnitsCount}, JSON Size: {size}") for j, variant in enumerate(variants): start = time.perf_counter() variant(size, executionUnitsCount) end = time.perf_counter() print(f"{variant.__name__}, elapsed: {round(end - start, 2)}") print()Результаты:
Какие выводы можно сделать?
-
Использование потоков для параллельного исполнения CPU bound задач не дает преимущества, результаты сопоставимы с кодом не использующим потоки и работающим последовательно. С чем это связано? Все дело в GIL — механизме синхронизации потоков, именно он не позволяет программе «набрать мощность» и выполнять потоки параллельно.
-
На процессы GIL не распространяется и создав дочерние процессы мы начинаем утилизировать ресурсы процессора по максимуму и как следствие видим уменьшение времени исполнения на графике.
-
Максимальный параллелизм которого можно добиться c помощью процессов равен количеству физических ядер CPU. На графике можно заметить что время исполнения теста Processes растет примерно лесенкой с шагом 4 — это количество физических ядер CPU на моем ноутбуке.
Задача №2: Работа с внешним API или IO-bound task
Помимо CPU bound существуют задачи IO bound. В такой задаче производительность зависит не от процессора а от подсистемы ввода-вывода, а также устройств с которыми мы взаимодействуем, например файловая система или сеть.
В примере ниже я буду эмулировать IO нагрузку через sleep (представьте что вместо sleep — запрос по HTTP к API вашего любимого сервиса).
def work(latency): time.sleep(latency)Задача: сделать N вызовов функции (например, мы сервис агрегатор услуг и ходим к партнерам за данными). Последовательный вариант выполнения задачи будет выглядеть так:
def sequential(latency, count): for _ in range(count): work(latency)Конкурентная реализация на потоках:
# Do work using OS Threads def run_threads(latency, executionUnitsCount): with concurrent.futures.ThreadPoolExecutor(max_workers=executionUnitsCount) as executor: executor.map(work, [latency] * executionUnitsCount)Конкурентная реализация на процессах:
# Do work using OS Processes def run_processes(latency, executionUnitsCount): with concurrent.futures.ProcessPoolExecutor() as executor: executor.map(work, [latency] * executionUnitsCount)Конкурентная реализация на основе asyncio:
def run_async_io(latency, executionUnitsCount): asyncio.run(async_io_tasks(latency, executionUnitsCount)) async def async_work(latency): await asyncio.sleep(latency) async def async_io_tasks(latency, executionUnitsCount): tasks = [asyncio.create_task(async_work(latency)) for _ in range(executionUnitsCount)] await asyncio.gather(*tasks)Бенчмарк (Чтобы показать разницу между подходами тестируем от 1000 до 2000 конкурентных задач):
if __name__ == '__main__': testCases = [ (0.1, i) for i in range(1000, 20001, 200)] variants = [run_threads, run_async_io, run_processes] for i, t in enumerate(testCases): latency, executionUnitsCount = t print(f"Parallelism\: {executionUnitsCount}") for j, variant in enumerate(variants): start = time.perf_counter() r = variant(latency, executionUnitsCount) end = time.perf_counter() print(f"{case.__name__}, elapsed: {round(end - start, 2)}") print()Результаты:
Увеличим масштаб и сравним потоки и корутины:
Какие выводы можно сделать?
-
Процессы в чистом виде непригодны для подобного класса задач, слишком большой оверхед на создание процесса и его убийство, взаимодействовать с ОС так часто — дорогое удовольствие.
-
Потоки справились намного лучше процессов, и GIL в данной задаче не стал помехой, так как он запрещает «исполняться» нескольким потокам одновременно, а в случае IO нагрузки потоки дольше находятся в ожидании чем в работе . Но оверхед по сравнению с корутинами присутствует так как мы взаимодействуем с ОС и аллоцируем память.
-
Корутины — абсолютный лидер, за счет того что ими управляет рантайм языка а не ОС, к тому же с точки зрения памяти корутине нужно ее меньше чем потоку.
Есть ли ограничения по количеству процессов/потоков/корутин?
В примерах выше мы запускали разное количество задач и у любопытного читателя может возникнуть вопрос: Сколько можно создавать процессов и потоков в программе? Чем мы платим за их создание? Ответ довольно прозаичен: мы платим оперативной памятью нашей машины. Для каждого процесса и потока ОС резервирует определенное количество памяти, а так как память конечна то без остановки создавая процесс или поток мы рано или поздно столкнемся с OOM (Out of memory error).
Чтобы избегать подобных ошибок в production системах ограничивается количество активных процессов и потоков с помощью популярных механизмов ThreadPool и ProcessPool.
Итоги
В данной статье я показал практическое применение потоков, процессов и корутин. На примерах разобрались в каких задач применим тот или иной инструмент. Для того чтобы упростить код были выбраны простые синтетические примеры, но при этом не противоречащие production коду который мы встречаем ежедневно в работе.
Полный код из статьи доступен на GitHub.
Статьи для дополнительного ознакомления:
Спасибо что прочитали до конца, буду рад любой конструктивной обратной связи в комментариях!
ссылка на оригинал статьи https://habr.com/ru/articles/773376/
Добавить комментарий