Модуль `multiprocessing` в Python

В этой статье объясняется модуль multiprocessing в Python.

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

YouTube Video

Модуль multiprocessing в Python

Основы: почему стоит использовать multiprocessing?

multiprocessing позволяет выполнять параллелизацию на уровне процессов, поэтому вы можете параллельно обрабатывать задачи, требующие большого количества вычислений, не ограничиваясь GIL (Global Interpreter Lock) в Python. Для задач, связанных с вводом-выводом (I/O-bound), threading или asyncio могут быть проще и предпочтительнее.

Простейшее использование Process

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

 1# Explanation:
 2# This example starts a separate process to run `worker` which prints messages.
 3# It demonstrates starting, joining, and passing arguments.
 4
 5from multiprocessing import Process
 6import time
 7
 8def worker(name, delay):
 9    # English comment in code per user's preference
10    for i in range(3):
11        print(f"Worker {name}: iteration {i}")
12        time.sleep(delay)
13
14if __name__ == "__main__":
15    p = Process(target=worker, args=("A", 0.5))
16    p.start()
17    print("Main: waiting for worker to finish")
18    p.join()
19    print("Main: worker finished")

• Этот код демонстрирует, как основной процесс запускает дочерний процесс worker и ждет его завершения с помощью метода join(). Вы можете передавать аргументы с помощью параметра args.

Простая параллелизация с помощью Pool (высокоуровневый API)

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

 1# Explanation:
 2# Use Pool.map to parallelize a CPU-bound function across available processes.
 3# Good for "embarrassingly parallel" workloads.
 4
 5from multiprocessing import Pool, cpu_count
 6import math
 7import time
 8
 9def is_prime(n):
10    # Check primality (inefficient but CPU-heavy for demo)
11    if n < 2:
12        return False
13    for i in range(2, int(math.sqrt(n)) + 1):
14        if n % i == 0:
15            return False
16    return True
17
18if __name__ == "__main__":
19    nums = [10_000_000 + i for i in range(50)]
20    start = time.time()
21    with Pool(processes=cpu_count()) as pool:
22        results = pool.map(is_prime, nums)
23    end = time.time()
24    print(f"Found primes: {sum(results)} / {len(nums)} in {end-start:.2f}s")

• Pool автоматически регулирует количество рабочих процессов, а метод map возвращает результаты в исходном порядке.

Межпроцессное взаимодействие: паттерн производитель/потребитель с использованием Queue

Queue — это очередь First-In-First-Out (FIFO), которая безопасно передает объекты между процессами. Ниже приведены некоторые типичные шаблоны.

 1# Explanation:
 2# Demonstrates a producer putting items into a Queue
 3# and consumer reading them.
 4# This is useful for task pipelines between processes.
 5
 6from multiprocessing import Process, Queue
 7import time
 8import random
 9
10def producer(q, n):
11    for i in range(n):
12        item = f"item-{i}"
13        print("Producer: putting", item)
14        q.put(item)
15        time.sleep(random.random() * 0.5)
16    q.put(None)  # sentinel to signal consumer to stop
17
18def consumer(q):
19    while True:
20        item = q.get()
21        if item is None:
22            break
23        print("Consumer: got", item)
24        time.sleep(0.2)
25
26if __name__ == "__main__":
27    q = Queue()
28    p = Process(target=producer, args=(q, 5))
29    c = Process(target=consumer, args=(q,))
30    p.start()
31    c.start()
32    p.join()
33    c.join()
34    print("Main: done")

• Queue позволяет безопасно передавать данные между процессами. Обычно для обозначения завершения используется специальное значение, например, None.

Разделяемая память: Value и Array

Value и Array используются для обмена небольшими числами или массивами между процессами. Для избежания конфликтов необходимо использовать блокировки (locks).

 1# Explanation:
 2# Use Value to share a single integer counter
 3# and Array for a small numeric array.
 4# Show how to use a Lock to avoid race conditions.
 5
 6from multiprocessing import Process, Value, Array, Lock
 7import time
 8
 9def increment(counter, lock, times):
10    for _ in range(times):
11        with lock:
12            counter.value += 1
13
14def update_array(arr):
15    for i in range(len(arr)):
16        arr[i] = arr[i] + 1
17
18if __name__ == "__main__":
19    lock = Lock()
20    counter = Value('i', 0)  # 'i' = signed int
21    shared_arr = Array('i', [0, 0, 0])
22
23    p1 = Process(target=increment, args=(counter, lock, 1000))
24    p2 = Process(target=increment, args=(counter, lock, 1000))
25    a = Process(target=update_array, args=(shared_arr,))
26
27    p1.start(); p2.start(); a.start()
28    p1.join(); p2.join(); a.join()
29
30    print("Counter:", counter.value)
31    print("Array:", list(shared_arr))

• Value и Array передают данные между процессами через низкоуровневые механизмы (разделяемая память на уровне языка C), а не средствами Python. Следовательно, он подходит для быстрого чтения и записи небольших объёмов данных, но не подходит для обработки больших объёмов данных..

Более продвинутый обмен: разделяемые объекты (словари, списки) с помощью Manager

Если нужно делиться более сложными объектами, такими как списки или словари, используйте Manager().

 1# Explanation:
 2# Manager provides proxy objects like dict/list
 3# that can be shared across processes.
 4# Good for moderate-size shared state and easier programming model.
 5
 6from multiprocessing import Process, Manager
 7import time
 8
 9def worker(shared_dict, key, value):
10    shared_dict[key] = value
11
12if __name__ == "__main__":
13    with Manager() as manager:
14        d = manager.dict()
15        processes = []
16        for i in range(5):
17            p = Process(target=worker, args=(d, f"k{i}", i*i))
18            p.start()
19            processes.append(p)
20        for p in processes:
21            p.join()
22        print("Shared dict:", dict(d))

• Manager удобен для совместного использования словарей и списков, но при каждом обращении данные передаются между процессами и требуется конвертация с помощью pickle. Следовательно, частое обновление больших объемов данных замедлит обработку.

Механизмы синхронизации: использование Lock и Semaphore

Используйте Lock или Semaphore для управления одновременным доступом к разделяемым ресурсам. Их удобно применять через оператор with.

 1# Explanation:
 2# Demonstrates using Lock to prevent simultaneous access to a critical section.
 3# Locks are necessary when shared resources are not atomic.
 4
 5from multiprocessing import Process, Lock, Value
 6
 7def safe_add(counter, lock):
 8    for _ in range(10000):
 9        with lock:
10            counter.value += 1
11
12if __name__ == "__main__":
13    lock = Lock()
14    counter = Value('i', 0)
15    p1 = Process(target=safe_add, args=(counter, lock))
16    p2 = Process(target=safe_add, args=(counter, lock))
17    p1.start(); p2.start()
18    p1.join(); p2.join()
19    print("Counter:", counter.value)

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

Различия между fork в UNIX и поведением в Windows

В UNIX-подобных системах процессы по умолчанию копируются с помощью fork, что обеспечивает эффективное использование памяти благодаря Copy-On-Write. На Windows процессы запускаются через spawn (что повторно импортирует модули), поэтому важно защищать точку входа и глобальную инициализацию.

 1# Explanation: Check start method (fork/spawn) and set it if needed.
 2# Useful for debugging platform-dependent behavior.
 3
 4from multiprocessing import get_start_method, set_start_method
 5
 6if __name__ == "__main__":
 7    print("Start method:", get_start_method())
 8
 9    # uncomment to force spawn on Unix for testing
10    # set_start_method('spawn')

• set_start_method можно вызывать только один раз при запуске вашей программы. Безопаснее не изменять этот метод произвольно внутри библиотек.

Практический пример: тест производительности для задач, нагружающих процессор (сравнение)

Ниже приведён скрипт, который просто сравнивает, насколько быстрее может быть обработка с параллелизацией с помощью multiprocessing. Здесь используется Pool.

 1# Explanation:
 2# Compare sequential vs parallel execution times for CPU-bound task.
 3# Helps understand speedup and overhead.
 4
 5import time
 6from multiprocessing import Pool, cpu_count
 7import math
 8
 9def heavy_task(n):
10    s = 0
11    for i in range(1, n):
12        s += math.sqrt(i)
13    return s
14
15def run_sequential(nums):
16    return [heavy_task(n) for n in nums]
17
18def run_parallel(nums):
19    with Pool(processes=cpu_count()) as p:
20        return p.map(heavy_task, nums)
21
22if __name__ == "__main__":
23    nums = [2000000] * 8  # heavy tasks
24    t0 = time.time()
25    run_sequential(nums)
26    seq = time.time() - t0
27    t1 = time.time()
28    run_parallel(nums)
29    par = time.time() - t1
30    print(f"Sequential: {seq:.2f}s, Parallel: {par:.2f}s")

• В этом примере видно, что в зависимости от нагрузки и числа процессов параллелизация иногда оказывается неэффективной из-за издержек. Чем больше (тяжелее) и независимее задачи, тем больше выигрыш от параллелизма.

Важные основные правила

Ниже приведены основные моменты для безопасного и эффективного использования multiprocessing.

• В Windows при запуске дочерних процессов модули импортируются заново, поэтому необходимо защищать точку входа скрипта через if __name__ == "__main__":.
• Межпроцессное взаимодействие сериализуется (с помощью конвертации pickle), поэтому передача крупных объектов становится затратной.
• Так как multiprocessing создает процессы, обычно число процессов выбирают исходя из значения multiprocessing.cpu_count().
• Создание другого Pool внутри рабочего процесса становится сложным, поэтому по возможности избегайте вложенных экземпляров Pool.
• Поскольку исключения, возникающие в дочерних процессах, трудно обнаружить из главного процесса, необходимо явно реализовать логирование и обработку ошибок.
• Количество процессов определяйте по количеству ядер CPU; для задач, связанных с вводом-выводом, рассмотрите использование потоков.

Практические советы по проектированию

Ниже приведены некоторые полезные концепции и шаблоны для проектирования параллельной обработки.

• Эффективно разделять процессы на роли, такие как чтение входных данных (I/O), предварительная обработка (multicpu) и агрегация (последовательно), реализуя 'конвейер' (pipelining).
• Чтобы упростить отладку, сначала проверьте работу кода в одном процессе до параллелизации.
• Для логирования разделяйте логи по процессам (например, включайте PID в имена файлов), чтобы упростить обнаружение проблем.
• Реализуйте механизмы повторных попыток и тайм-аутов, чтобы можно было корректно восстановиться при зависании процесса.

Итоги (основные моменты, которые можно сразу применить)

Параллельная обработка — мощный инструмент, но важно правильно оценивать природу задач, объем данных и стоимость межпроцессного взаимодействия. multiprocessing эффективен для задач, нагружающих процессор, но плохой дизайн или ошибки синхронизации могут снизить производительность. Если следовать основным правилам и шаблонам, можно создавать безопасные и эффективные параллельные программы.

Вы можете следовать этой статье, используя Visual Studio Code на нашем YouTube-канале. Пожалуйста, также посмотрите наш YouTube-канал.

YouTube Video