Multiprocessing в Python: как загрузить все ядра без хаоса

Вы запускаете расчёт на четырёх потоках, ждёте четырёхкратного ускорения, а получаете то же время или хуже. Причина в GIL: в CPython в каждый момент байткод исполняет только один поток. Для CPU-задач это тупик. multiprocessing обходит его, запуская отдельные процессы с собственным интерпретатором и собственным GIL — но платит за это сериализацией данных и накладными расходами на старт. Разбираем, что из модуля брать под конкретную задачу и на чём люди спотыкаются. Материал для Python 3.11–3.12, но принципиальные вещи верны с 3.8.

Когда процессы, а когда нет

Первое решение — вообще не про API, а про природу задачи.

• CPU-bound (перемножение матриц, парсинг, хеширование, обработка изображений на чистом Python) — здесь потоки бесполезны, нужны процессы.
• I/O-bound (запросы к сети, чтение файлов, обращения к БД) — потоки или asyncio обычно лучше: GIL освобождается на время ожидания, а процессы только добавят расходов на сериализацию.
• Задача уже в C-библиотеке — NumPy, Pandas, многие операции которых отпускают GIL внутри. Иногда достаточно потоков, иногда векторизация быстрее любого параллелизма.

Процесс — это не бесплатный поток. Старт процесса и передача в него данных стоят миллисекунды и копирование памяти. Если каждая единица работы длится микросекунды, вы утонете в накладных расходах.

Три способа раздать работу

В модуле есть низкоуровневый Process, пул Pool и более новый concurrent.futures.ProcessPoolExecutor. Выбор зависит от того, насколько однородна работа и нужен ли вам контроль над каждым процессом.

Pool.map — когда работа однородна

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

# Python 3.11
import multiprocessing as mp

def heavy(n: int) -> int:
    # имитация CPU-нагрузки: сумма квадратов
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    tasks = [10_000_00] * 8
    # число процессов по умолчанию = os.cpu_count()
    with mp.Pool(processes=4) as pool:
        results = pool.map(heavy, tasks)
    print(results)

with закрывает пул и дожидается завершения. Без контекстного менеджера не забудьте pool.close() и pool.join(), иначе процессы повиснут.

ProcessPoolExecutor — когда нужны будущие результаты

Тот же пул, но интерфейс Future: можно получать результаты по мере готовности через as_completed, а не ждать самый долгий.

# Python 3.11
from concurrent.futures import ProcessPoolExecutor, as_completed

def heavy(n: int) -> int:
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    payloads = [10_000_00, 5_000_00, 20_000_00, 8_000_00]
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(heavy, n): n for n in payloads}
        for fut in as_completed(futures):
            n = futures[fut]
            print(f"n={n} -> {fut.result()}")

Важное отличие поведения: если функция внутри процесса кинула исключение, оно всплывёт при вызове fut.result(), а не в момент submit. У Pool.map исключение всплывает при итерации по результату.

Process — когда процессы разнородны

Один процесс слушает очередь, другой пишет в файл, третий считает. Здесь пул не подходит — берите Process напрямую и Queue для обмена.

Сравнение подходов

Метод старта: fork, spawn, forkserver

Это то, обо что чаще всего спотыкаются при переезде между ОС. Способ создания дочернего процесса задаётся через mp.set_start_method или mp.get_context.

• fork — дочерний процесс наследует память родителя копированием при записи. Быстрый старт. Был методом по умолчанию на Linux; в Python 3.12 использование fork в многопоточном процессе помечено как ненадёжное. Ломается с некоторыми библиотеками, которые держат открытые дескрипторы или потоки.
• spawn — запускает чистый интерпретатор и заново импортирует ваш модуль. Медленнее, но предсказуемо. По умолчанию на macOS и Windows.
• forkserver — компромисс: отдельный сервер-процесс форкает воркеры из чистого состояния. Доступен на Unix.

Практическое следствие spawn: код верхнего уровня выполнится в каждом дочернем процессе повторно. Поэтому весь запуск обязан быть под if __name__ == "__main__": — иначе на Windows и macOS получите бесконечный рекурсивный запуск процессов. Это не стилистика, а условие работоспособности.

# явно задаём контекст, не полагаясь на дефолт ОС
import multiprocessing as mp

if __name__ == "__main__":
    ctx = mp.get_context("spawn")
    with ctx.Pool(4) as pool:
        ...

Что реально ломается

Несериализуемый аргумент

Всё, что передаётся в процесс, идёт через pickle. Лямбды, вложенные функции, открытые сокеты и файловые объекты не пиклятся. Типичная ошибка: Can't pickle local object или PicklingError. Решение — функция верхнего уровня модуля вместо лямбды, а вместо файлового объекта передавайте путь и открывайте внутри воркера.

Тихая смерть воркера

Если процесс убит по OOM, у Pool вы можете получить BrokenProcessPool (в ProcessPoolExecutor) или зависание. Причина обычно в том, что каждый воркер держит копию больших данных. Считайте память: N воркеров — это до N копий того, что не разделяется.

Копирование больших входных данных

Передача гигабайтного массива в каждый вызов map означает сериализацию его заново на каждую задачу. Для больших read-only данных инициализируйте их один раз через параметр initializer пула или используйте multiprocessing.shared_memory (доступен с Python 3.8).

Ошибки не видно

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

Сколько процессов запускать

Дефолт os.cpu_count() — разумная отправная точка для чистого CPU-bound. Но:

1. Если задача частично ждёт I/O, воркеров может быть больше числа ядер.
2. Если каждый воркер сам параллелит через NumPy/BLAS, вы получите переподписку: N процессов × M потоков BLAS. Ограничьте потоки библиотеки через переменные окружения (например, для OpenBLAS это OMP_NUM_THREADS) до запуска, иначе ядра будут драться за время.
3. В контейнере os.cpu_count() может вернуть число ядер хоста, а не лимит cgroup. Проверяйте фактически доступные ядра, а не доверяйте функции вслепую.

Как измерить, а не угадать

Не берите цифры ускорения на веру — замерьте на своих данных. Минимальный протокол: зафиксируйте объём входа и число ядер, прогоните последовательную версию и версию с пулом на 2, 4, 8 воркерах, постройте зависимость. Если после какого-то числа воркеров время растёт — вы упёрлись либо в память, либо в сериализацию, либо в переподписку BLAS. Мерьте wall-clock через time.perf_counter(), а не time.time().

Что делать дальше

Начните с ProcessPoolExecutor и as_completed — это самый предсказуемый вход. Убедитесь, что запуск под __main__, а функция-воркер лежит на верхнем уровне модуля. Потом посмотрите на два показателя в проде: пиковую память на воркер и долю времени, которую съедает передача аргументов. Если данные большие и read-only — читайте про shared_memory и initializer пула. Обмен данными между процессами через Queue и Manager — отдельная тема, здесь мы её не разворачивали.