Многопоточность и многопроцессорность

Введение

Многозадачность - это способность операционной системы или окружения обеспечивать одновременное или псевдоодновременное выполнение нескольких задач.

Точно так же и сложные приложения являются многозадачными: они обрабатывают пользовательские запросы, обращаются к базе данных, выполняют вычисления и поддерживают отзывчивость пользовательского интерфейса, всё это должно происходить плавно, создавая иллюзию одновременной работы всех связанных задач. Как же это происходит?

Существуют два вида многозадачности:

Процессная многозадачность.
Поточная многозадачность.

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

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

Каждое приложение имеет как минимум один процесс, и у каждого процесса есть как минимум один поток. Процесс представляет собой более крупную единицу работы, в то время как поток - более мелкую единицу, которая выполняется внутри процесса.

Примечание 1.

В этом курсе будет расмотрено лишь самое главное для парсинга, для более детального понимая советую пройти курсы по этим темам.

Использование многопоточности и многопроцессорности в Python зависит от конкретной задачи и её требований.

Многопоточность (threading) подходит для ситуаций, когда задачи могут выполняться параллельно в рамках одного процесса. Это особенно полезно в задачах, где большая часть времени уходит на ввод-вывод (I/O-bound tasks), таких как чтение и запись файлов, работа с сетью, обработка запросов к базе данных и т.д. Использование потоков позволяет избежать блокировки процесса на операциях ввода-вывода, так как потоки могут работать параллельно и не ждать завершения друг друга.

Многопроцессорность (multiprocessing), с другой стороны, используется в ситуациях, когда задачи требуют интенсивных вычислений (CPU-bound tasks). В этом случае создаются отдельные процессы, каждый из которых выполняет свою задачу. Поскольку каждый процесс имеет собственное пространство памяти, он может использоваться для распределения вычислительных ресурсов более эффективно на многоядерных процессорах. Однако стоимость создания и управления процессами выше, чем у потоков, из-за накладных расходов на передачу данных между процессами и синхронизацию.

Итак, когда выбирать многопоточность:

Когда большая часть времени уходит на операции ввода-вывода.
Когда требуется обработка большого количества клиентских запросов (например, веб-сервер).
Когда необходимо работать с несколькими параллельными задачами.

А многопроцессорность подходит:

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

Пул потоков

Пул потоков (Threading pool) - это механизм, используемый для эффективного управления и переиспользования потоков в приложении. Он позволяет создавать заранее определенное количество потоков, которые могут выполнять различные задачи из очереди задач.

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

Преимущества использования пула потоков включают:

Уменьшение накладных расходов на создание и уничтожение потоков: Поскольку потоки создаются заранее и переиспользуются, избегается издержка на создание и уничтожение потоков при выполнении задач.
Контроль нагрузки: Пул потоков позволяет контролировать количество одновременно выполняющихся задач и предотвращает излишнюю нагрузку на систему.
Улучшение производительности: Пул потоков может улучшить производительность приложения, особенно при обработке множества коротких и независимых задач.
Упрощение управления потоками: Использование пула потоков упрощает управление потоками в приложении, поскольку разработчику не нужно явно создавать и уничтожать потоки.
Предотвращение перегрузки системы: Пул потоков может помочь предотвратить перегрузку системы, ограничивая количество одновременно выполняющихся задач.

Общая схема использования пула потоков включает в себя создание пула потоков с заданным количеством потоков, добавление задач в очередь задач и их последующее выполнение с использованием доступных потоков в пуле.

Python предоставляет модуль concurrent.futures, который содержит класс ThreadPoolExecutor для работы с пулом потоков. Этот класс позволяет легко создавать и управлять пулом потоков в Python.

Пул потоков создается классом
concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())

В concurrent.futures.ThreadPoolExecutor все аргументы необязательны, что означает, что вы можете указать только max_workers, если нужно. Однако некоторые из аргументов имеют значения по умолчанию:

max_workers - это параметр, определяющий максимальное количество потоков в пуле. Если он не указан или равен None, то устанавливается значение по умолчанию, которое вычисляется как минимум из 32 и числа ядер процессора плюс 4. Это значение по умолчанию гарантирует, что как минимум 5 потоков останется доступными для операций ввода/вывода и предотвращает необоснованное использование большого количества ресурсов на многоядерных системах.
thread_name_prefix - это префикс, добавляемый к именам всех потоков, создаваемых пулом. Это полезно для отладки, так как позволяет легче отслеживать, какой поток относится к какой задаче.
initializer - это объект, который вызывается перед выполнением каждой задачи в каждом рабочем потоке. Это может быть полезно, например, для инициализации какого-то общего состояния, используемого в задачах. Если initializer выбрасывает исключение, все текущие ожидающие задачи будут отменены, и любая попытка отправить задачу в пул потоков также завершится этим исключением.
initargs - это кортеж аргументов, которые будут переданы в initializer при его вызове.

В классе пула потоков обычно есть несколько методов для управления его поведением и выполнения задач. Вот несколько обычных методов и их назначение:

submit(func, *args, **kwargs): Этот метод используется для отправки задачи на выполнение в пул потоков. Он принимает функцию func, которая будет выполнена в отдельном потоке, а также любые аргументы и ключевые аргументы, необходимые для выполнения функции. Метод возвращает объект Future, который представляет собой результат выполнения задачи и позволяет отслеживать ее статус и получать результат, когда он будет готов.
map(func, iterable, chunksize=1): Этот метод аналогичен встроенной функции map(), но выполняет функцию func параллельно в пуле потоков. Он принимает функцию func и итерируемый объект iterable, применяет функцию к каждому элементу итерируемого объекта, используя пул потоков для ускорения выполнения. chunksize определяет размер порции данных(по умолчанию равно 1), которые передаются каждому потоку для обработки.
shutdown(wait=True): Этот метод используется для остановки пула потоков после завершения всех задач. Если wait установлен в True (по умолчанию), метод блокирует вызывающий поток до завершения всех задач и остановки потоков. Если wait установлен в False, метод завершает выполнение текущих задач и останавливает потоки, не дожидаясь завершения оставшихся задач.
shutdown_now(): Этот метод немедленно останавливает пул потоков без ожидания завершения текущих задач. Все невыполненные задачи будут отменены, и потоки будут немедленно остановлены.
submit: Этот метод используется для отправки отдельной задачи на выполнение в пуле потоков. Вы передаете ему функцию (и, возможно, аргументы для этой функции), которую вы хотите выполнить параллельно, и он возвращает объект Future, представляющий результат выполнения этой задачи. submit позволяет отправлять задачи на выполнение по одной и имеет более гибкое управление над каждой задачей, так как вы можете отслеживать ее статус и получать результаты индивидуально.
map: Этот метод похож на встроенную функцию map() в Python. Он принимает функцию и итерируемый объект (например, список) и применяет эту функцию к каждому элементу итерируемого объекта, выполняя это параллельно в пуле потоков. В отличие от submit, map обрабатывает несколько задач одновременно и возвращает результаты в том же порядке, в котором они присутствовали в исходном итерируемом объекте. Это удобно, когда требуется применить одну и ту же функцию к каждому элементу данных и получить результаты в том же порядке.

Примечание 1.

Метод submit возвращает результаты выполнения задач в порядке их завершения, а не в том порядке, в котором задачи были отправлены на выполнение.

Чтобы избежать необходимости явного закрытия пула потоков, рекомендуется использовать контекстный менеджер. Он автоматически обеспечивает корректное завершение работы пула после выполнения всех задач. Пример использования контекстного менеджера выглядит так:

import concurrent.futures

with concurrent.futures.ThreadPoolExecutor() as executor: # Здесь можно использовать методы пула потоков, например, executor.submit() или executor.map() # После выхода из блока контекстного менеджера пул потоков будет автоматически закрыт

Такой подход обеспечивает удобство и безопасность, так как гарантирует, что пул потоков будет закрыт правильно даже в случае возникновения исключений в процессе выполнения задач.

Вот несколько примеров, демонстрирующих, как использование пула потоков может оптимизировать скорость выполнения программы. Для измерения времени выполнения используются функции time.perf_counter() из модуля time.

import time

def calculate_square(n):
    return n * n

start_time = time.perf_counter()

results = []
for i in range(1, 11):
    results.append(calculate_square(i))

end_time = time.perf_counter()

print("Потрачено времени:", end_time - start_time) # Потрачено времени: 6.299989763647318e-0

Взглянем на этот же пример с пулом потоков:

import time
import concurrent.futures

def calculate_square(n):
    return n * n

start_time = time.perf_counter()

with concurrent.futures.ThreadPoolExecutor() as executor:
    results = list(executor.map(calculate_square, range(1, 11)))

end_time = time.perf_counter()


print("Потрачено времени:", end_time - start_time) # 0.003899499977706

Посмотрим пример, приближенный к нашему курсу:

import time
import requests
import concurrent.futures

urls = [
    'https://example.com/page1',
    'https://example.com/page2',
    'https://example.com/page3',
]

def fetch_url(url):
    response = requests.get(url)
    return response.text

start_time = time.perf_counter()

with concurrent.futures.ThreadPoolExecutor() as executor:
    results = list(executor.map(fetch_url, urls))

end_time = time.perf_counter()

print("Время потрачено:", end_time - start_time)

Здесь также использование пула потоков сократило время выполнения за счет параллельной загрузки нескольких веб-страниц.

Использование пула потоков особенно эффективно при выполнении задач, которые могут быть выполнены параллельно и не зависят друг от друга. Однако следует помнить, что использование потоков также имеет свои ограничения, такие как конкуренция за ресурсы процессора и возможные проблемы с синхронизацией данных.

Пул процессов

Пул процессов - это механизм, используемый в Python для создания и управления группой процессов, которые могут параллельно выполнять задачи. Подобно пулу потоков, пул процессов позволяет распределить задачи на выполнение между несколькими процессами, что может привести к ускорению выполнения программы, особенно на многоядерных системах.

Основные характеристики пула процессов включают в себя:

Многозадачность: Пул процессов позволяет выполнять несколько задач параллельно в разных процессах, что повышает эффективность использования вычислительных ресурсов.
Отделение ресурсов: Каждый процесс в пуле процессов работает в своем собственном адресном пространстве памяти, что гарантирует изоляцию данных между процессами и минимизирует возможность конфликтов.
Безопасность: Поскольку процессы работают независимо друг от друга, ошибки или сбои в одном процессе обычно не влияют на работу остальных процессов в пуле.
Поддержка многоядерных систем: Пул процессов особенно полезен на многоядерных или многопроцессорных системах, где можно достичь параллельного выполнения задач и эффективного использования всех доступных ядер процессора.
Управление жизненным циклом: Пул процессов обычно предоставляет удобный интерфейс для создания, управления и завершения процессов, а также для сбора результатов выполнения задач.

Использование пула процессов может быть особенно полезным для выполнения CPU-интенсивных задач, таких как обработка больших объемов данных, вычислительные расчеты и параллельные вычисления.

Процесс:

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

2. Каждый процесс имеет свой уникальный набор ресурсов, таких как открытые файлы и сетевые соединения, в то время как все потоки внутри процесса используют общие ресурсы.

3. Процессы более надежно изолированы друг от друга за счет наличия собственных адресных пространств памяти, что снижает вероятность взаимной блокировки и гонок данных.

4. Процессы могут работать на разных ядрах процессора и использовать многопроцессорные системы.

5. При сбое одного процесса остальные могут продолжать работу, в то время как сбой одного потока в процессе может повлиять на остальные потоки в этом процессе и, в конечном итоге, привести к сбою всего процесса.

Поток:

1. Потоки делят запоминающее устройство с другими потоками, что может вызывать проблемы с взаимной блокировкой и гонками данных.

2. Потоки являются более легковесными, чем процессы, и требуют меньше ресурсов при создании и управлении.

3. Потоки используют общие ресурсы внутри процесса, такие как открытые файлы и сетевые соединения.

4. Потоки не могут работать на разных ядрах процессора и не могут использовать преимущества многопроцессорных систем (пока что в Python).

5. При сбое одного потока это может привести к сбою всего процесса, так как они используют общие ресурсы.

В конечном итоге, запуск нового процесса требует больше системных ресурсов (памяти, времени и т. д.), по сравнению с запуском нового потока. Это обусловлено необходимостью формирования нового адресного пространства памяти и повторного использования ресурсов процесса.

Пул процессов (Process Pool) представляет собой механизм параллелизма, который используется для эффективного управления и распределения задач между несколькими процессами. В отличие от пула потоков, где используются потоки, пул процессов использует отдельные процессы для выполнения задач.

Основная идея пула процессов состоит в том, чтобы создать заранее определенное количество процессов, которые будут готовы к выполнению задач. Когда поступает новая задача, она отправляется в очередь задач пула. Свободный процесс из пула забирает задачу из очереди и выполняет её. После завершения задачи процесс возвращается в пул, готовый к выполнению следующей задачи.

Использование пула процессов имеет несколько преимуществ:

Изоляция: Каждый процесс в пуле имеет собственное адресное пространство памяти, что уменьшает вероятность возникновения конфликтов при одновременном доступе к ресурсам.
Масштабируемость: Пул процессов может использоваться на многопроцессорных системах, так как каждый процесс может выполняться на отдельном ядре процессора, что увеличивает общую производительность.
Устойчивость к блокировкам: Поскольку каждый процесс работает независимо, блокировка в одном процессе не затрагивает другие процессы, что обеспечивает более стабильное выполнение задач.

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

Приступим к разбору пула процессов:

concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=(), max_tasks_per_child=None)

В concurrent.futures.ProcessPoolExecutor обязательных аргументов нет, все они являются необязательными, то есть можем указать только аргумент max_workers, к примеру. Однако, некоторые из них имеют значения по умолчанию:

max_workers: Этот аргумент определяет максимальное количество одновременно запущенных процессов в пуле. По умолчанию max_workers=None, что означает, что количество процессов будет равно количеству доступных ядер процессора. Если вы хотите ограничить количество процессов, вы можете указать конкретное число.
mp_context: Этот аргумент позволяет указать контекст многопроцессорности (multiprocessing context), который будет использоваться при создании пула процессов. По умолчанию mp_context=None, что означает использование стандартного контекста multiprocessing.
initializer: Это функция, которая будет вызвана при инициализации каждого процесса в пуле. Она позволяет выполнить какие-либо начальные настройки или подготовку перед выполнением основных задач. По умолчанию initializer=None.
initargs: Это кортеж аргументов, передаваемых в функцию initializer. По умолчанию initargs=().
max_tasks_per_child: Этот аргумент позволяет указать максимальное количество задач, которые будет выполнено каждым процессом перед его перезапуском. После выполнения указанного количества задач процесс будет завершен и создан новый процесс для дальнейшей работы. По умолчанию max_tasks_per_child=None, что означает отсутствие ограничения на количество задач.

Примечание 1.

В контексте многопроцессорности существуют два пула процессов, оба нацелены на одни и те же цели и выполняют аналогичные действия. Однако рассмотрим лишь тот, который используется в модуле concurrent.futures, поскольку на практике этот пул немного удобнее в использовании.

Примечание 2.

В классе ProcessPoolExecutor из модуля concurrent.futures содержатся те же методы и их функциональность, что и в классе ThreadPoolExecutor. Но здесь map() предпочтительнее для выполнения однотипных задач над списком данных, в то время как submit() предоставляет более гибкий интерфейс для выполнения различных задач.

Посмотрим на примеры:

import time

def fibonacci(n):
    if n <= 1:
        return n
    else:
        return fibonacci(n-1) + fibonacci(n-2)

if __name__ == "__main__":
    n = 42  # Вычисляем значение 40-го числа Фибоначчи
    start_time = time.perf_counter()
    result = fibonacci(n)
    end_time = time.perf_counter()
    print("Время выполнения без использования ProcessPoolExecutor:", end_time - start_time) #36.23401099999319

Выполним этот же пример с пулом процессов:

import concurrent.futures
import time

def fibonacci(n):
    if n <= 1:
        return n
    else:
        return fibonacci(n-1) + fibonacci(n-2)

if __name__ == "__main__":
    n = 40  # Вычисляем значение 40-го числа Фибоначчи
    start_time = time.perf_counter()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        result = executor.submit(fibonacci, n).result()
    end_time = time.perf_counter()
    print("Время выполнения с использованием ProcessPoolExecutor:", end_time - start_time) # 14.430688399996143

Использование пула процессов приводит к распараллеливанию выполнения задачи на несколько процессов, что позволяет производить вычисления параллельно. Это увеличивает эффективность обработки данных и приводит к сокращению времени выполнения программы в 2.5 по сравнению с последовательным выполнением задач."