Мультипроцессорность. Создание и запуск процессов

Мультипроцессорность (Multiprocessing) представляет способ параллельного выполнения программ, при котором создаются отдельные независимые процессы. Каждый такой процесс имеет собственное пространство памяти и выполняется на отдельном ядре процессора.

В языке Python мультипроцессорность критически важна из-за наличия Global Interpreter Lock (GIL). GIL - это механизм, который позволяет только одному потоку управлять интерпретатором Python в один момент времени. Из-за этого обычные потоки (threading) не могут ускорить вычисления на многоядерных процессорах. В Python мультипроцессорность представлена модулем multiprocessing, который обходит GIL, запуская физически разные процессы.

Когда использовать мультипроцессорность, а когда - потоки:

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

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

Главные плюсы мультипроцессорности:

• Полный обход GIL: задействуются все ядра CPU

• Повышенная стабильность: падение одного процесса не ломает другие.

• Подходит для тяжелых математических вычислений.

Но также следует упомянуть и минусы:

• Потребление памяти: каждый процесс копирует окружение и дополнительно потребляет RAM

• Медленный старт по сравнению с обычными потоками.

• Сложный обмен данными (требует IPC: Queue, Pipes).

Создание процесса

Наиболее простой и распространенный способо создания и инициализации процесса представляет класс Process из модуля multiprocessing:

class multiprocessing.Process(group=None, name=None, args=(), kwargs={}, *, daemon=None)

Ключевые параметры конструктора:

• target: вызываемый объект (функция), который начнет выполняться при запуске процесса. По умолчанию он равен None, что означает отсутствие вызова (ничего не выполняется).

• name: задает имя процесса

• args: кортеж аргументов для вызываемого объекта target (аргументы для параметров функции).

• kwargs: словарь именованных аргументов для вызываемого объекта target

Фактически объекты Process представляют собой задачу, выполняемую в отдельном процессе. Стоит отметить, что класс Process содержит аналоги всех методов класса threading.Thread.

Например, создадим простейший процесс:

from multiprocessing import Process    # импортируем класс Process

# Создаем объект процесса
my_process = Process(target=print, name="Print_Process", args=("Hello METANIT.COM",))

print(my_process)

Для создания процесса сначала импортируем класс multiprocessing.Process. При создании процесса в конструкторе указываем, что процесс будет выполнять встроенную функцию print() (target=print), что процесс будет называться "Print_Process" (name="Print_Process") и что при выполнении в функцию print будет передаваться строка "Hello METANIT.COM" (args=("Hello METANIT.COM",). Обратите внимание, что после строки идет запятая, так как args представляет кортеж.

Запуск одного процесса

Как и в случае с потоками Thread, для запуска процессов применяется метод start() объекта Process. Например, запустим новый процесс, который будет просто выводить некоторое сообщение на консоль:

from multiprocessing import Process     # подключаем класс Process

if __name__ == '__main__':
    # Создаем объект процесса
    my_process = Process(target=print, args=("Hello METANIT.COM",))
    
    # Запускаем процесс
    my_process.start()
    
    # Ожидаем завершения процесса, прежде чем идти дальше
    my_process.join()
    
    print("Главный процесс завершил работу.")

Здесь вызов my_process.start() приведет к запуску нового процесса, который выполнит функцию print() с передачей ей строки "Hello METANIT.COM". В итоге мы получим следующий вывод:

Hello METANIT.COM
Главный процесс завершил работу.

Важно весь код с запуском процессов с использованием multiprocessing должен обязательно находиться внутри блока if __name__ == '__main__'. Это предотвращает бесконечную рекурсивную генерацию подпроцессов.

Аналогично мы можем запускать в созданном процессе свои функции:

from multiprocessing import Process
import os
import time

def work(name):
    pid = os.getpid()   # получаем идентификатор процесса в системе
    print(f"[{os.getpid()}] Процесс {name} запущен.")
    time.sleep(2)   # задержка на 2 секунды - имитация работы
    print(f"[{os.getpid()}] Процесс {name} завершил работу.")

if __name__ == "__main__":
    # Создаем объект процесса
    my_process = Process(target=work, args=("Process 1",))
    
    # Запускаем процесс
    my_process.start()
    
    # Ожидаем завершения процесса, прежде чем идти дальше
    my_process.join()
    
    print("Главный процесс завершил работу.")

В данном случае запускается функция work(), которая выводит пару сообщений на консоль. Здесь следует отметить использование функции os.getpid(), которая позволяет получить идентификатор процесса на уровне системы. И, кроме того, для имитации работы для задержки на 2 секунды здесь применяется вызов time.sleep(2). ПРимер консольного вывода:

[27010] Процесс Process 1 запущен.
[27010] Процесс Process 1 завершил работу.
Главный процесс завершил работу.

Свойства процесса

У класса Process стоит отметить ряд свойств:

• name: имя процесса - строковое значение, используемое исключительно для идентификации; оно не несет никакой функциональной нагрузки. Разным процессам может быть присвоено одно и то же имя.

  Если имя не передано конструктору явно, оно формируется по шаблону "Process-N1:N2:...:Nk", где каждое Nk - это порядковый номер дочернего процесса относительно родительского.

• pid: идентификатор процесса (PID). До момента запуска процесса это поле имеет значение None.

• exitcode: код завершения дочернего процесса. Если процесс еще не завершился, это поле имеет значение None.

Применим свойства:

from multiprocessing import Process

if __name__ == "__main__":
    # Создаем объект процесса
    my_process = Process(target=print, name="Print_Process", args=("Hello METANIT.COM",))
    
    my_process.start()  # Запускаем процесс
    
    my_process.join()   # Ожидаем завершения процесса

    print("name = ", my_process.name)
    print("pid = ", my_process.pid)
    print("exitcode = ", my_process.exitcode)

Пример консольного вывода:

Hello METANIT.COM
name =  Print_Process
pid =  40098
exitcode =  0

Здесь мы видим, что имя процесса, как и было указано в конструкторе, - "Print_Process". Если бы в конструктор не передали бы имя процесса, по умолчанию оно бы было "Process-1". Ну и кроме того, свойство exitcode равно 0, это значит, что процесс завершился успешно. Другое числовое значение сигнализировало бы об ошибке. Соответственно при завершении мы можем проверить успешность завершения процесса.

Пул процессов (Process Pool) и метод map

Если необходимо обработать большой список данных, удобнее всего использовать класс multiprocessing.Pool. Он автоматически распределяет задачи по доступным ядрам процессора. Рассмотрим небольшой пример:

from multiprocessing import Pool

import time

def square_number(number):
    time.sleep(0.5)  # Имитация тяжелых вычислений
    result = number ** 2
    print(f"Число {number} возведено в квадрат: {result}")
    return result

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]
    
    start_time = time.time()
    
    # Создаем пул процессов (по умолчанию равен количеству ядер CPU)
    with Pool() as pool:
        # pool.map берет функцию и список аргументов, выполняя их параллельно
        results = pool.map(square_number, numbers)
        
    end_time = time.time()
    
    print(f"Результаты: {results}")
    print(f"Затраченное время: {end_time - start_time:.2f} сек.")

Здесь для удобного управления ресурсами для создания пула используем конструкцию with

with Pool() as pool:

Внури этой конструкции запускаем процессы с помощью метода map():

pool.map(square_number, numbers)

Первый параметр этого метода - выполняемое в процессе действие - в данном случае это функция square_number, которая возводит число в квадрат.

Второй параметр - итерируемый объект - в данном случае список чисел numbers. Метод map разбивает итерируемый объект на несколько фрагментов (частей) и передает их в пул процессов как отдельные параметры выполняемой функции.

ПРимер консольного вывода:

Число 1 возведено в квадрат: 1
Число 2 возведено в квадрат: 4
Число 3 возведено в квадрат: 9
Число 4 возведено в квадрат: 16
Число 5 возведено в квадрат: 25
Результаты: [1, 4, 9, 16, 25]
Затраченное время: 0.58 сек.

Если бы мы выполняли этот код последовательно, он бы занял порядка 2,5 секунды. На 4-ядерном процессоре он выполнится примерно за 1 секунду.

Передав в метод map() положительное число, мы можем указать конкретное количество запускаемых процессов:

from multiprocessing import Pool
import time
import os

def square_number(number):
    pid = os.getpid()
    time.sleep(0.5)  # Имитация тяжелых вычислений
    result = number ** 2
    print(f"[PID {pid}] Число {number} возведено в квадрат: {result}")
    return result

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]
    
    start_time = time.time()
    
    # Создаем пул процессов (по умолчанию равен количеству ядер вашего CPU)
    with Pool(2) as pool:
        # pool.map берет функцию и список аргументов, выполняя их параллельно
        results = pool.map(square_number, numbers)
        
    end_time = time.time()
    
    print(f"Результаты: {results}")
    print(f"Затраченное время: {end_time - start_time:.2f} сек.")

Для большей наглядности в функции square_number с помощью функции os.getpid() получаем идентификатор текущего процесса. Пример консольного вывода:

[PID 41380] Число 1 возведено в квадрат: 1
[PID 41381] Число 2 возведено в квадрат: 4
[PID 41380] Число 3 возведено в квадрат: 9
[PID 41381] Число 4 возведено в квадрат: 16
[PID 41380] Число 5 возведено в квадрат: 25
Результаты: [1, 4, 9, 16, 25]
Затраченное время: 1.56 сек.

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