Процессы: multiprocessing

⏱ 35 мин · 🧩 заданий: 5

Примеры этого урока не запускаются в браузере

В браузерном Python нет процессов, поэтому код показан с выводом. Запустите его локально — увидите загрузку всех ядер в диспетчере задач.

Зачем процессы

GIL не даёт потокам CPython считать параллельно. У каждого процесса свой интерпретатор и свой GIL — процессы работают на разных ядрах по-настоящему одновременно.

import time
from concurrent.futures import ProcessPoolExecutor

def count_primes(limit):                         # честная CPU-нагрузка
    count = 0
    for n in range(2, limit):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            count += 1
    return count

if __name__ == "__main__":                       # обязательно!
    tasks = [200_000] * 8

    start = time.perf_counter()
    results = [count_primes(t) for t in tasks]
    print(f"последовательно: {time.perf_counter() - start:.1f} с")

    start = time.perf_counter()
    with ProcessPoolExecutor() as pool:           # по процессу на ядро
        results = list(pool.map(count_primes, tasks))
    print(f"процессы:        {time.perf_counter() - start:.1f} с")

последовательно: 6.4 с
процессы:        1.0 с          # на 8 ядрах

Process

from multiprocessing import Process, current_process
import os

def work(name):
    print(f"{name}: pid={os.getpid()}, родитель={os.getppid()}, {current_process().name}")

if __name__ == "__main__":
    procs = [Process(target=work, args=(f"задача {i}",)) for i in range(3)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()
    print("exitcode:", [p.exitcode for p in procs])

Pool и ProcessPoolExecutor

from multiprocessing import Pool

def square(x):
    return x * x

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        print(pool.map(square, range(10)))                 # результаты по порядку
        print(pool.starmap(pow, [(2, 3), (3, 2)]))         # несколько аргументов
        for r in pool.imap_unordered(square, range(5)):    # лениво, по мере готовности
            print(r, end=" ")
        async_result = pool.apply_async(square, (12,))
        print("\n", async_result.get(timeout=5))

concurrent.futures.ProcessPoolExecutor даёт тот же интерфейс, что и пул потоков (submit, map, as_completed, Future) — его обычно и используют в новом коде.

Почему `if name == "main"`

Методы запуска процессов:

Метод	Как создаётся процесс	По умолчанию
`fork`	копия текущего процесса (copy-on-write) — быстро	был в Linux до 3.14
`spawn`	новый интерпретатор, заново импортирует главный модуль	Windows, macOS
`forkserver`	процессы порождаются от «чистого» сервера	Linux с 3.14

При spawn дочерний процесс импортирует ваш скрипт. Если создание процессов не спрятано под if __name__ == "__main__":, каждый дочерний начнёт порождать свои — бесконечная рекурсия (и RuntimeError).

fork опасен в сочетании с потоками (копируется состояние блокировок), поэтому Python уходит от него по умолчанию.

Передача данных

Процессы не разделяют память. Всё, что передаётся — аргументы, результаты, сообщения, — сериализуется pickle и копируется:

лямбды, вложенные функции, открытые файлы, соединения с БД, блокировки потоков — не сериализуются;
функции передаются по имени — они должны быть определены на уровне модуля;
большие объекты дорого гонять туда-обратно.

import pickle, math

data = pickle.dumps(math.sqrt)                  # функция модуля передаётся по имени
print(len(data), "байт:", data)
print(pickle.loads(data)(16))

for obj in [lambda x: x, (i for i in range(3))]:
    try:
        pickle.dumps(obj)
    except Exception as e:
        print(type(e).__name__, ":", e)

Queue и Pipe

from multiprocessing import Process, Queue

def producer(q):
    for i in range(3):
        q.put({"task": i})
    q.put(None)

def consumer(q):
    while (item := q.get()) is not None:
        print("обработал", item)

if __name__ == "__main__":
    q = Queue()
    p1, p2 = Process(target=producer, args=(q,)), Process(target=consumer, args=(q,))
    p1.start(); p2.start(); p1.join(); p2.join()

Общая память

from multiprocessing import Process, Value, Array, Lock

def add(counter, lock):
    for _ in range(10_000):
        with lock:                       # общая память требует синхронизации, как потоки
            counter.value += 1

if __name__ == "__main__":
    counter, lock = Value("i", 0), Lock()
    procs = [Process(target=add, args=(counter, lock)) for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(counter.value)                 # 40000

Для больших массивов (numpy) — multiprocessing.shared_memory.SharedMemory: процессы работают с одним блоком памяти без копирования. Manager() даёт общие dict/list через прокси — удобно, но медленно.

Накладные расходы

Процесс стартует десятки миллисекунд, каждая задача — сериализация туда и обратно. Поэтому:

дробите работу на крупные куски (chunksize в map, деление диапазона на части);
передавайте параметры, а не данные: имя файла вместо его содержимого;
для числовых данных сначала попробуйте numpy — векторизация часто быстрее процессов;
max_workers ≈ число ядер (os.cpu_count()), больше для CPU-bound не имеет смысла.

Очереди задач

Когда вычисления надо вынести из веб-запроса или распределить по нескольким машинам, используют очереди задач:

Инструмент	Для чего
Celery (+ Redis/RabbitMQ)	фоновые задачи в Django/FastAPI: письма, отчёты, периодические задачи
RQ, Dramatiq, arq	проще Celery
Dask, Ray	распределённые вычисления, большие данные, ML

Вопрос на собеседовании

«Как ускорить CPU-bound задачу в Python?» — Процессы (ProcessPoolExecutor, multiprocessing.Pool) — у каждого свой GIL, работа идёт на всех ядрах. Учитывать накладные расходы pickle и запуска, дробить работу крупными кусками. Альтернативы: numpy/векторизация, C-расширения (Cython), PyPy, free-threaded Python, вынос в другой язык.

Итог

Процессы обходят GIL: свой интерпретатор на ядро.
ProcessPoolExecutor / Pool — основной инструмент; код запуска — под if __name__ == "__main__":.
Данные копируются через pickle: функции — на уровне модуля, никаких лямбд и соединений.
Queue, Pipe, Value/Array, shared_memory, Manager — обмен данными.
Крупные куски работы; для распределённых задач — Celery, Dask, Ray.

Войдите, чтобы отмечать пройденные уроки.

	Queue / Pipe	shared_memory / Array	Manager
Поток сообщений-задач от главного процесса рабочим
Большой массив чисел, который все процессы читают без копирования
Общий словарь с удобным API (медленнее, через прокси)