При разработке сложных систем с большим количеством задач, синхронное выполнение этих задач в конечном итоге приведет к падению производительности.
Конкурентность и параллелизм – это механизмы реализованные для исправления этой ситуации путем «переплетения» нескольких задач, либо путем параллельного выполнения.
Конкурентность
Цель конкурентности – предотвратить взаимоблокировки задач путем переключения между ними, когда одна из задач вынуждена ждать внешнего ресурса. Типичный пример – обработка нескольких сетевых запросов.
Вариант реализации конкурентности – запуск запросов по очереди, пока запросы не обработаются. Здесь, однако, возникает проблема со скоростью и эффективностью.
Решение проблемы – запустить запросы одновременно, а затем переключаться между ними по мере получения ответов.
Параллелизм
Пример: Есть 10 рабочих. Не хотелось бы, чтобы 1 рабочий работал, а остальные 9 сидели и ничего не делали.
Чтобы повысить эффективность, требуется разделить работу между ними, чтобы работа выполнялась быстрее.
Параллелизм использует ту же концепцию, которая применяется к аппаратным ресурсам. Речь идет о максимальном использовании этих ресурсов путем запуска процессов или потоков, использующих все ядра процессора, которыми располагает компьютер.
Обе эти концепции полезны для одновременной обработки нескольких задач, но требуется выбрать правильный метод для конкретной задачи.
Как работать с этими концепциями в Python?
Python предоставляет механизмы для реализации конкурентности и параллелизма : для конкурентности используется многопоточность и асинхронность, для параллелизма используется многопроцессорность.
Многопоточность
Поток похож на последовательную программу: у него есть начало, процесс работы и окончание работы. В момент выполнения потока существует единственная точка выполнения, однако поток не программа — поток не может выполнятся сам по себе.
Поток выполнения — одна или несколько функций, которые возможно выполнить, независимо от остальной части программы.
Рассмотрим два способа обработки потоков в Python:
- С помощью библиотек потоков.
- С помощью
ThreadPoolExecutor, созданного в качестве менеджера контекста. Это простой способ создавать и уничтожать пул.
import logging
import threading
import time
from concurrent.futures import ThreadPoolExecutor
def main():
logging.info(f"************** Multiple threads with threading ****************")
threads = list()
for _ in range(3):
logging.info(f"Main function: create and start thread {_+1}.")
x = threading.Thread(target=thread_function, args=(_,))
threads.append(x)
x.start()
for _, thread in enumerate(threads):
logging.info(f"Main function: before joining thread {_+1}.")
thread.join()
logging.info(f"Main function: thread {_+1} done")
logging.info(f"************** Multiple threads with ThreadPoolExecutor ****************")
with ThreadPoolExecutor(max_workers=3) as executor:
executor.map(thread_function, range(3))
def thread_function(name):
logging.info(f"Thread starting: {name+1}")
time.sleep(5)
logging.info(f"Thread {name+1}: finishing")
if __name__ == "__main__":
# Setting a logger to track the progress
logging.basicConfig(format="%(message)s", level=logging.INFO)
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
logging.info(f"{__file__} executed in {elapsed:0.2f} seconds.")
В приведенном выше примере создается функция, которая запускается в отдельных потоках. Эта функция запускает поток, а затем засыпает, имитируя некоторое внешнее время ожидания.
В функции main видно, как реализованы оба метода:
- первый в строках 9-19.
- второй в строках 23-24.
from concurrent.futures import ThreadPoolExecutor
import logging
import time
import urllib.request as url_request
def main():
collected_data = list()
urls = [
"https://python.org",
"https://docs.python.org/",
"https://wikipedia.org",
"https://imdb.com",
"https://google.com"
]
logging.info(f"************** Starting threads ****************")
with ThreadPoolExecutor(max_workers=5) as executor:
for url in urls:
executor.submit(thread_function, url, collected_data)
for data_stream in collected_data:
logging.info(f"First part of data stream for {data_stream['url']}: {data_stream['data_stream'][:200]}")
logging.info(f"-------------------------------------------------------------------------------------")
logging.info(f"******************************************************")
def thread_function(url, collected_data):
connection = url_request.urlopen(url)
data_stream = connection.read()
collected_data.append({'url': url, 'data_stream': data_stream})
if __name__ == "__main__":
# Setting a logger to track the progress
logging.basicConfig(format="%(message)s", level=logging.INFO)
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
logging.info(f"{__file__} executed in {elapsed:0.2f} seconds.")
В этом куске кода использована многопоточность для одновременного чтения данных из нескольких URL-адресов, выполнения нескольких экземпляров thread_function() и сохранения результатов в списке.
Как видно из примера, ThreadPoolExecutor упрощает обработку необходимых потоков. Хотя это простой пример, возможно отправить больше URL-адресов, используя ThreadPoolExecutor не дожидаясь каждого ответа.
Из приведенных выше примеров видно, что потоки — это удобный и понятный способ обработки задач, ожидающих других ресурсов. Также видно, что стандартная библиотека Python поставляется с высокоуровневыми реализациями, которые еще больше упрощают эту задачу.
Coroutines и asyncio
Coroutines (сопрограммы) – альтернативный способ одновременного выполнения функции посредством специальных конструкций, а не системных потоков.
Сопрограмма – общая структура управления, в которой управление потоком совместно передается между двумя подпрограммами без возврата. Получаем однопоточную однопроцессорную конструкцию, использующую кооперативную многозадачность.
В то время, как многопоточность берет и запускает функции в отдельных потоках, asyncio работает в одном потоке и разрешает циклу обработки событий программы взаимодействовать с несколькими задачами, чтобы каждая из них выполнялась по очереди в оптимальное время.
Реализуем это на Python с помощью библиотеки asyncio, предоставляющей основу и API для запуска и управления сопрограммами с ключевыми словами async и await.
Пример:
import asyncio
import logging
import time
async def async_function(number):
logging.info(f"Starting Async function: {number}")
await asyncio.sleep(1)
logging.info(f"Finishing Async function: {number}")
async def main():
logging.info("************ Starting Async Program ***************")
await asyncio.gather(async_function(1), async_function(2), async_function(3))
logging.info("***************************************************")
if __name__ == "__main__":
logging.basicConfig(format="%(message)s", level=logging.INFO)
s = time.perf_counter()
asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"{__file__} executed in {elapsed:0.2f} seconds.")
Пример:
import asyncio
import sys
from aiohttp import ClientSession
import logging
import time
async def async_function(session, url):
async with session.get(url) as request:
data_stream = await request.text()
return {'url': url, 'data_stream': data_stream}
async def main():
urls = [
"https://python.org",
"https://docs.python.org/",
"https://wikipedia.org",
"https://imdb.com",
"https://google.com"
]
logging.info("************ Starting Async Program ***************")
async with ClientSession() as session:
collected_data = await asyncio.gather(*[async_function(session, _) for _ in urls])
for data_stream in collected_data:
logging.info(f"First part of data stream for {data_stream['url']}: {data_stream['data_stream'][:200]}")
logging.info(f"-------------------------------------------------------------------------------------")
logging.info("***************************************************")
if __name__ == "__main__":
logging.basicConfig(format="%(message)s", level=logging.INFO)
s = time.perf_counter()
if sys.version_info[:2] == (3, 7):
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
loop.run_until_complete(asyncio.sleep(2.0))
finally:
loop.close()
#asyncio.run(main())
elapsed = time.perf_counter() - s
logging.info(f"{__file__} executed in {elapsed:0.2f} seconds.")
В отличие от работы с потоками, где любая функция выполняется в потоке, в сопрограмме четко указывается в синтаксисе, какие функции выполняются параллельно.
Также сопрограммы не связаны архитектурными ограничениями как потоки и требуют меньше памяти из-за того, что выполняются в одном потоке.
Многопроцессорность
Многопроцессорность – механизм, разрешающий параллельно выполнять ресурсоемкие задачи, запуская независимые экземпляры интерпретатора Python.
Каждый экземпляр получает код и данные, необходимые для выполнения рассматриваемой задачи, и выполняется независимо в собственном потоке.
Реализуем приведенные выше примеры с использованием многопроцессорной обработки, чтобы увидеть отличия.
from multiprocessing import Pool
import time
import logging
def main():
logging.info(f"************** Starting Multiprocessing ****************")
with Pool() as p:
p.map(processing_function, range(3))
logging.info(f"********************************************************")
def processing_function(name):
logging.info(f"Process starting: {name+1}")
time.sleep(5)
logging.info(f"Process finishing: {name+1}")
if __name__ == "__main__":
# Setting a logger to track the progress
logging.basicConfig(format="%(message)s", level=logging.INFO)
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
logging.info(f"{__file__} executed in {elapsed:0.2f} seconds.")
Пример выше показывает многопроцессорную обработку, а ниже та же функциональность чтения данных с несколькими URL одновременно.
from multiprocessing import Pool
import time
import logging
import urllib.request as url_request
def main():
urls = [
"https://python.org",
"https://docs.python.org/",
"https://wikipedia.org",
"https://imdb.com",
"https://google.com"
]
logging.info(f"************** Starting Multiprocessing ****************")
with Pool() as p:
collected_data = p.map(request_function, urls)
for data_stream in collected_data:
logging.info(f"First part of data stream for {data_stream['url']}: {data_stream['data_stream'][:200]}")
logging.info(f"-------------------------------------------------------------------------------------")
logging.info(f"********************************************************")
def request_function(url):
connection = url_request.urlopen(url)
data_stream = str(connection.read())
return {'url': url, 'data_stream': data_stream}
if __name__ == "__main__":
# Setting a logger to track the progress
logging.basicConfig(format="%(message)s", level=logging.INFO)
s = time.perf_counter()
main()
elapsed = time.perf_counter() - s
logging.info(f"{__file__} executed in {elapsed:0.2f} seconds.")
В фрагменте выше объект Pool() представляет повторно используемую группу процессов, которая сопоставляет итерации для распределения между каждым экземпляром функции.
Преимущество: каждая операция выполняется в отдельной среде выполнения Python и на полном ядре ЦП, что разрешает одновременно запускать процессы интенсивно использующие ЦП.
Недостаток: каждый подпроцесс требует копию данных (с которой подпроцесс работает), отправленных из главного процесса и, как правило, возвращающих данные в главный процесс.
В статье были рассмотрены концепции:
- Конкурентность.
- Параллелизм.
И подходы для работы с этими концепциями:
- Многопоточность.
- Coroutines и asyncio.
- Многопроцессорность.
Комментарии