Многопроцессорность в Python от multiprocessing до Ray и Dask накладные расходы сериализации и выбор инструмента

Сервер на тридцать два ядра принимает ночную задачу по расчёту скорингов, и наблюдать за ним грустно: одно ядро загружено на сто процентов, остальные тридцать один отдыхают. Разбить работу на процессы несложно, в стандартной библиотеке для этого есть готовый модуль, а в экосистеме ещё пара фреймворков. Настоящая трудность начинается позже: каждый аргумент и каждый результат обязан пересечь границу процесса, а на границе стоит таможня по имени pickle, и её пошлина способна съесть весь выигрыш от параллельности. Ниже разбор того, что multiprocessing реально берёт за пересылку значений, как Ray и Dask снижают эту цену через общую память и умный планировщик, и в какой ситуации какой инструмент окупается без сюрпризов.

Как GIL отправляет вычислительную задачу в одиночное ядро

В классическом CPython глобальная блокировка интерпретатора позволяет ровно одному потоку исполнять байткод в каждый момент времени. Для задач, ждущих сеть или диск, это не помеха: ожидание освобождает блокировку, и потоков в задачах ввода-вывода может быть сотни. Для вычислительного кода картина обратная: десять потоков честно делят одно ядро и тратят время на переключение контекста. Счёт идёт не в пользу распараллеливания, а в пользу процессов, у каждого из которых собственный интерпретатор, собственная память и собственная блокировка. С Python 3.13 существует экспериментальная сборка без глобальной блокировки, но для продакшена она пока экзотика, поэтому рабочая схема осталась прежней: потоков для ожидания, процессов для вычислений.

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

Что забирает pickle при трёх способах запуска дочерних процессов

multiprocessing умеет рожать процессы тремя способами, и разница между ними ощущается сразу. Вариант fork копирует родителя целиком: память наследуется, а копируется лениво, при первой записи. Десятилетиями это был дешёвый и любимый путь в Linux, пока не выяснилось, что форк процесса с несколькими потоками небезопасен: в Python 3.12 такой вызов начал выдавать предупреждение, а в 3.14 fork перестал быть умолчанием на любой платформе. Теперь код, которому нужен именно fork, обязан просить его явно.

Вариант spawn запускает свежий интерпретатор: дочерний процесс заново импортирует модули, и старт получается заметно медленнее, зато предсказуемым и переносимым. Windows и macOS живут на spawn по умолчанию, причём macOS перешла на него ещё в версии 3.8. Третий путь, forkserver, держит наготове чистый служебный процесс и раздаёт форки от него: с Python 3.14 это умолчание на Linux и прочих POSIX-системах. Общее правило безопасности одно и работает на spawn и forkserver: точку входа ограждают проверкой, иначе дочерние процессы начнут импортировать главный модуль и заново порождать себя:

import multiprocessing as mp

def hard(chunk):
    return sum(i * i for i in chunk)

if __name__ == "__main__":
    ctx = mp.get_context("spawn")
    with ctx.Pool(processes=8) as pool:
        totals = pool.map(hard, [range(10**6) for _ in range(32)])

Очереди и пайпы где пересылка значений съедает выигрыш от ядер

Связь между процессами строится на очередях и пайпах, и у каждого из них за кулисами одна и та же процедура: объект превращается в байты функцией pickle, байты уходят через канал, на той стороне собираются обратно. Для мелких значений процедура незаметна. Для numpy-массива на полгигабайта он превращается в долгую историю: сериализация делает копию в памяти отправителя, пересылка занимает заметное время, приёмник собирает вторую копию, и то же самое предстоит результату на обратном пути. Память процесса в пике удваивается, а ускорение вместо ожидаемых семи раз на восьми воркерах останавливается на полутора.

Пайп устроен проще очереди: Connection.send сериализует значение и пишет его в канал напрямую, без фидерного потока, который Queue держит для сглаживания пересылок. Для пары процессов, обменивающихся гигантскими пакетами, пайп даёт заметно меньше служебной работы, очередь выигрывает удобством: она безопасна при десятке потребителей и не требует разруливания, кто и когда читает.

Ещё дороже обходятся менеджеры. Manager создаёт объект-посредник в отдельном процессе, и каждое обращение к его методам или элементам превращается в сетевое сообщение с сериализацией туда и обратно. Цикл по списку через менеджер упирается не в вычисления, а в конвейер пересылок. Отдельная экономия прячется в размере порций: pool.map, вызванный без chunksize, раздаёт воркерам элементы по одному, и конвейер захлёбывается мелкими пересылками, где сериализация съедает больше, чем само вычисление. Порции побольше лечат болезнь, а imap_unordered возвращает результаты по мере готовности, не выстраивая очередь из тех, кто опоздал:

with ctx.Pool(processes=8) as pool:
    for total in pool.imap_unordered(hard, chunks, chunksize=4):
        handle(total)

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

Разделяемая память как обходной путь для больших массивов и её границы

Модуль shared_memory, появившийся в Python 3.8, даёт способ вообще не пересылать большие массивы: сегмент общей памяти создаётся один раз, а воркеры прикладывают к нему представление numpy и работают с теми же байтами:

from multiprocessing import shared_memory
import numpy as np

payload = np.random.rand(10**6)
shm = shared_memory.SharedMemory(create=True, size=payload.nbytes)
view = np.ndarray(payload.shape, dtype=payload.dtype, buffer=shm.buf)
view[:] = payload
    # процессам передаётся имя shm.name, а не сам массив

Имя сегмента пересылается как обычная строка, и это единственное, что проходит через pickle. Дальше каждый воркер открывает сегмент по имени и получает массив без единого байта сериализации. Границы у приёма честные: за сегментом нужно убирать вручную, забытая ссылка удерживает память, а синхронизацию записи придётся строить самостоятельно через Lock, потому что общая память не даёт защиты от одновременных изменений. За сиротами присматривает отдельный трекер ресурсов: процесс, умерший без закрытия сегмента, оставляет после себя утечку, трекер замечает её и печатает предупреждение, но страховка не отменяет дисциплины, где close и unlink остаются прямой обязанностью кода. На старте fork существовала и бесплатная альтернатива: наследованные страницы копируются лениво, и чтение большого массива в дочернем процессе ничего не стоит. Печаль в том, что любая запись в такую страницу запускает копирование, а отказ от fork по умолчанию закрыл и эту лазейку для новых проектов.

Ray и объектное хранилище с нулевым копированием внутри узла

Ray подходит к той же проблеме с другой стороны и берёт на себя обе работы: и пересылку значений, и расписание задач. Функция помечается декоратором remote и превращается в задачу, выполнение уезжает в воркер, а вызывающий код получает на руки ссылку на будущий результат:

import ray

ray.init()

@ray.remote
def fit_shard(shard_ref):
    shard = ray.get(shard_ref)
    return train(shard)

shards = [ray.put(chunk) for chunk in chunks]
refs = [fit_shard.remote(s) for s in shards]
scores = ray.get(refs)

Ключевая механика спрятана в ray.put: значение уезжает в объектное хранилище узла, размещённое в общей памяти, и все локальные воркеры читают его напрямую, для массивов numpy без копирования. Сериализация случается один раз при размещении, а не при каждой передаче между задачами. Когда хранилище переполняется, холодные объекты выгружаются на диск, и вычисления продолжаются. Для состояния, которое должно переживать между вызовами, есть акторы: класс с декоратором remote превращается в объект с собственным процессом, и запросы к нему выстраиваются в очередь. Так живёт модель машинного обучения, которую дорого грузить при каждом вызове:

@ray.remote
class Scorer:
    def __init__(self, path):
        self.model = load_model(path)

    def score(self, features):
        return self.model.predict(features)

engine = Scorer.remote("model.bin")
value = ray.get(engine.score.remote(features))

Плата очевидна: фреймворк поднимает собственную инфраструктуру, его изучение само по себе задача, и на одиночной машине с двумя воркерами вся эта машинерия не окупится.

Dask и планировщик с ленивыми графами для числовых и табличных конвейеров

Dask решает проблему пересылок третьим способом: вместо ручной раздачи кусков он строит граф задач и сам решает, где и что исполнять. Библиотека предлагает коллекции, повторяющие привычные интерфейсы: dask.array похож на numpy, dask.dataframe на pandas, а вычисления откладываются до вызова compute, за который планировщик успевает спланировать и склеить операции:

from dask.distributed import Client
import dask.array as da

client = Client(n_workers=8)
x = da.random.random((50000, 50000), chunks=(5000, 5000))
result = (x * 2).mean().compute()

Умение сливать соседние операции в один прогон уменьшает число пересылок между воркерами, а конвейер из десяти шагов по фильтрации и агрегации исполняется без материализации промежуточных гигабайт. Для кода, который не сводится к массивам и таблицам, есть прозрачный путь через dask.delayed: обычные функции оборачиваются декоратором, граф собирается из обычных выражений, а compute запускает весь маршрут разом:

from dask import delayed

frames = [delayed(read_csv)(p) for p in paths]
merged = delayed(concat)(frames)
table = merged.compute()

Планировщик из пакета distributed поднимает веб-панель на порту 8787, где видно загрузку, память и очередь задач, поддерживает адаптивное изменение числа воркеров и работает как на ноутбуке, так и в кластере. Слабое место симметрично силе: библиотека великолепна там, где конвейер выражается в терминах её коллекций или собирается из отложенных вызовов через dask.delayed, но за пределами этих сценариев она превращается в тяжёлый универсальный планировщик, которому нужна дисциплина.

Границы применимости трёх инструментов и порядок выбора под задачу

multiprocessing остаётся честным выбором для одиночной машины: нет зависимостей, нет фоновых процессов, полный контроль над каналами, а код читается как обычный Python. Плата: пересылки строятся вручную, за общей памятью следит разработчик, и за пределы одного хоста схема не масштабируется. Ray силён там, где нужен кластер, состояние в акторах и потоковая обработка с устойчивостью к сбоям отдельных задач. Dask уместен, когда работа выглядит как массивы, таблицы и графы из сотен мелких операций, и когда хочется видеть в панели, что именно происходит. Перед выбором полезно пройти короткий проверочный маршрут:

1. Замерить время сериализации ключевых объектов простым тестом с двумя процессами и сравнить его с временем вычисления;
2. Для задач, ждущих сеть или диск, сначала испытать потоки и asyncio и не плодить процессы без нужды;
3. Если вычисления укладываются в одну машину и пересылаемых объектов немного, остановиться на multiprocessing;
4. Если объекты измеряются гигабайтами и лежат в памяти, перейти на shared_memory или на объектное хранилище Ray;
5. Если конвейер выражается массивами и таблицами, отдать его коллекциям Dask и проверить граф в панели планировщика;
6. Если нужна распределённая система с состоянием и восстановлением после сбоев, брать Ray и выстроить его вокруг акторов.

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