Передача исключений в concurrent.futures и multiprocessing.Pool, в чём разница

Параллелизм в Python с использованием процессов выглядит обманчиво простым, пока всё работает. Но стоит первой задаче упасть с исключением где-то в дочернем процессе, как два главных инструмента для многопроцессорной обработки начинают вести себя совершенно по-разному. concurrent.futures.ProcessPoolExecutor и multiprocessing.Pool снаружи похожи, оба раздают задачи воркерам, оба возвращают результаты. Внутри же механика передачи исключений между процессами устроена настолько по-разному, что выбор инструмента может определить, насколько быстро команда найдёт причину сбоя в продакшене или потратит вечер на разбор оборванного трейсбэка.

Почему передача исключений между процессами это вообще проблема

В отличие от потоков, которые делят память внутри одного процесса, дочерние воркеры в многопроцессорной обработке живут в собственных интерпретаторах. Между ними нет общей памяти, любой обмен данными идёт через сериализацию (pickle) и межпроцессорные очереди. Когда в воркере возникает исключение, это объект со своим типом, аргументами, цепочкой cause и context и, что особенно важно, объектом traceback. Передать такое целиком в родительский процесс в общем случае невозможно. Объекты traceback в Python не поддерживают сериализацию pickle напрямую, потому что содержат ссылки на frame-объекты с локальными переменными, которые могут содержать что угодно, включая открытые сокеты или объекты файловой системы.

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

Как concurrent.futures обращается с упавшими задачами

ProcessPoolExecutor построен вокруг объекта Future, который инкапсулирует и результат, и возможную ошибку. Когда воркер падает с исключением, executor сериализует это исключение вместе с информацией о трейсбэке через специальный механизм. Внутри используется обёртка _RemoteTraceback, которая сохраняет текстовое представление трейсбэка из дочернего процесса и подменяет им cause у исключения в родительском. При попытке получить результат через future.result() оригинальное исключение перевозбуждается, а в выводе появляется текст вложенного удалённого трейсбэка с пометкой "The above exception was the direct cause of the following exception".

Выглядит это так. Допустим, есть функция, которая иногда падает:

import concurrent.futures


def risky_division(x: int, y: int) -> float:
    return x / y


if __name__ == "__main__":
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as pool:
        future = pool.submit(risky_division, 10, 0)
        try:
            result = future.result()
        except ZeroDivisionError as e:
            print(f"Поймали: {e}")

В трейсбэке при падении видно сразу два блока. Сначала идёт _RemoteTraceback с указанием, где именно в дочернем процессе случилась проблема (файл, строка, контекст вызова), а затем родительский трейсбэк, показывающий, как до места падения дошло уже в основном процессе. Это даёт почти столько же информации, как если бы код выполнялся синхронно.

У Future есть отдельный метод exception(), который позволяет проверить наличие ошибки, не возбуждая её:

future = pool.submit(risky_division, 10, 0)
err = future.exception(timeout=5)
if err is not None:
    print(f"В задаче была ошибка: {type(err).__name__}: {err}")
else:
    print(f"Результат: {future.result()}")

Такой раздельный доступ удобен, когда логика обработки ошибок и логика использования результата разнесены по разным местам. Можно перебрать список futures, отметить упавшие, а успешные обработать дальше, не теряя при этом информации о причинах сбоев.

Как multiprocessing.Pool ведёт себя в тех же ситуациях

multiprocessing.Pool появился раньше, и его подход проще. При вызове apply_async или map_async задача попадает в воркер, а результат (или исключение) возвращается через объект AsyncResult. Метод get() либо отдаёт результат, либо перевозбуждает исключение. На первый взгляд то же самое, что и у Future.result().

Разница в том, что получается с трейсбэком. Pool сериализует только само исключение, без расширенного контекста удалённого вызова. В современных версиях Python (начиная примерно с 3.4) к классу подмешивается простое представление того, где случилась ошибка, через атрибут cause с RemoteTraceback-подобным эффектом, но реализация менее богатая, чем у concurrent.futures. Часть деталей о переменных и цепочке вызовов теряется.

import multiprocessing


def risky_division(x: int, y: int) -> float:
    return x / y


if __name__ == "__main__":
    with multiprocessing.Pool(processes=4) as pool:
        async_result = pool.apply_async(risky_division, (10, 0))
        try:
            value = async_result.get(timeout=5)
        except ZeroDivisionError as e:
            print(f"Поймали: {e}")

При запуске Pool тоже покажет двухуровневый трейсбэк, но без отдельного метода для проверки факта ошибки. Чтобы узнать, упала задача или отработала успешно, есть только метод successful(), и его можно вызывать только после ready(), иначе он сам возбудит AssertionError. В коде это часто выглядит костыльно:

async_result = pool.apply_async(risky_division, (10, 0))
async_result.wait(timeout=5)
if async_result.ready():
    if async_result.successful():
        print(f"Результат: {async_result.get()}")
    else:
        try:
            async_result.get()
        except Exception as exc:
            print(f"Задача упала: {type(exc).__name__}: {exc}")

Очевидно, что та же логика на Future пишется лаконичнее и без обходных манёвров. Это первый практический довод в пользу concurrent.futures - современный код с тщательной обработкой ошибок там получается чище.

Что происходит при падении в map

Метод map есть у обоих API, и поведение при ошибке у них разное. У ProcessPoolExecutor исключение от любой задачи прорастает наружу в момент итерации по результатам. Если одна из задач падает, исключение поднимается на той итерации, к которой относится упавший результат, а остальные продолжают обрабатываться (хотя получить их уже не выйдет - ошибка прервёт цикл):

with concurrent.futures.ProcessPoolExecutor(max_workers=4) as pool:
    try:
        for result in pool.map(risky_division, [10, 20, 30, 40], [2, 0, 5, 4]):
            print(result)
    except ZeroDivisionError as e:
        print(f"Цепочка прервана: {e}")

У multiprocessing.Pool.map поведение похожее, но есть нюанс. Метод по умолчанию синхронный, он блокирует поток до завершения всех задач, и если хоть одна упала, исключение возбуждается уже после завершения остальных. Это удобно тем, что не теряется работа уже отработанных воркеров, но неудобно тем, что нельзя поймать ошибку на конкретном элементе входной последовательности.

Для получения частичных результатов в реальном коде чаще используют комбинацию submit и as_completed, которая работает только с concurrent.futures:

with concurrent.futures.ProcessPoolExecutor(max_workers=4) as pool:
    futures = {
        pool.submit(risky_division, x, y): (x, y)
        for x, y in [(10, 2), (20, 0), (30, 5), (40, 4)]
    }
    for future in concurrent.futures.as_completed(futures):
        args = futures[future]
        try:
            result = future.result()
            print(f"{args} -> {result}")
        except ZeroDivisionError as e:
            print(f"{args} упало: {e}")

В таком виде ни одна успешно отработавшая задача не теряется. Аналог на multiprocessing.Pool пишется заметно сложнее с использованием apply_async и ручного отслеживания готовности.

Ловушка с несериализуемыми исключениями

Есть отдельная категория проблем, которая бьёт оба API одинаково больно. Это случаи, когда само исключение нельзя сериализовать pickle. Сторонние библиотеки порой создают классы исключений с лямбдами, замыканиями или ссылками на несериализуемые объекты внутри. Когда такое исключение возникает в воркере, попытка отправить его в родительский процесс падает с PicklingError. И вот тут начинается интересное.

В multiprocessing.Pool такая ситуация приводит к тому, что воркер либо зависает в ожидании отправки, либо роняет весь пул, либо просто молча проглатывает результат, оставляя AsyncResult в состоянии never-ready. Диагностика сложна, потому что в логах появляется ошибка про pickle, не имеющая видимой связи с исходным кодом задачи.

ProcessPoolExecutor в этой ситуации ведёт себя предсказуемее. При невозможности сериализовать исключение пул переходит в состояние broken, и все ожидающие futures получают BrokenProcessPool с разумно понятным сообщением. Это всё ещё неприятно, но хотя бы заметно сразу.

Универсальное решение - оборачивать тело задачи в try/except и сериализовать только текстовое представление ошибки:

import traceback
from dataclasses import dataclass


@dataclass
class TaskFailure:
    error_type: str
    message: str
    traceback_text: str


def safe_task(func, *args, **kwargs):
    try:
        return func(*args, **kwargs)
    except Exception as exc:
        return TaskFailure(
            error_type=type(exc).__name__,
            message=str(exc),
            traceback_text=traceback.format_exc(),
        )

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

BrokenProcessPool и абортивные завершения воркеров

Отдельная категория проблем не связана с исключениями в коде задачи как таковыми. Воркер может умереть по причинам, лежащим вне его контроля - получить SIGKILL от OOM-киллера, упасть с segfault из-за бажной C-библиотеки, быть прибитым операционной системой. В таких случаях нет ни исключения для сериализации, ни возможности передать что-либо обратно.

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

В свежих версиях интерпретатора расширена и диагностика. В Python 3.14 при крахе воркера в BrokenProcessPool попадает информация о том, какой именно процесс упал и с каким кодом возврата:

import ctypes
import concurrent.futures


def worker_segfault():
    ctypes.string_at(0)


if __name__ == "__main__":
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as pool:
        future = pool.submit(worker_segfault)
        try:
            future.result()
        except concurrent.futures.process.BrokenProcessPool as e:
            print(f"Пул сломался: {e}")

Сообщение об ошибке теперь содержит примерно такую информацию - "Process 48939 terminated abruptly with exit code -11". Это сильно облегчает корреляцию с логами ядра и помогает быстро понять причину (-11 - это SIGSEGV, -9 - SIGKILL и так далее).

У multiprocessing.Pool аналогичная ситуация исторически решена ещё проще и хуже. По умолчанию умерший воркер не заменяется автоматически (если не использовать maxtasksperchild), а ожидающие задачи могут зависнуть навсегда. Поведение настолько неудобное, что многие проекты вынуждены оборачивать вызовы Pool в свои собственные сторожевые таймеры.

Когда какой инструмент выбирать в реальных проектах

Картина после всех разборов получается такой. ProcessPoolExecutor выигрывает почти по всем пунктам, когда речь идёт о работе с ошибками:

Отдельные методы result() и exception() позволяют проверять ошибку без её возбуждения;
Удалённый трейсбэк сохраняется и показывается в понятном виде;
as_completed даёт возможность забирать результаты по мере готовности и обрабатывать падения индивидуально;
BrokenProcessPool возникает предсказуемо и не оставляет задачи висящими;
В свежих версиях интерпретатора появляется всё больше деталей о причинах краха воркеров.

multiprocessing.Pool остаётся актуальным там, где нужно более тонкое управление процессами - кастомные начальные функции, контекст fork против spawn, прямая работа с очередями и пайпами, использование imap_unordered для потоковой обработки больших последовательностей с ленивой загрузкой. Под капотом ProcessPoolExecutor сам использует multiprocessing, так что выбор не всегда сводится к "одно вместо другого", иногда оба инструмента уживаются в одном проекте.

Главный принцип - если код задач может падать с нетривиальными исключениями и эти исключения нужно разбирать, то ProcessPoolExecutor дешевле в поддержке. Если же задачи в основном чистые вычисления, ошибки в них редки и однородны, а важнее тонкая настройка процесса - multiprocessing.Pool остаётся достойным вариантом, особенно с настройками maxtasksperchild и кастомным контекстом старта.

Несколько практических привычек, которые экономят часы отладки

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

Не возбуждать в воркерах кастомные исключения, которые могут оказаться несериализуемыми. Лучше возвращать структурированные объекты-результаты с явным полем статуса;
Логировать падения внутри самой задачи, а не надеяться на трейсбэк в родительском процессе. Локальный log внутри воркера сохранится в любом случае, даже если потом передача исключения провалится;
Устанавливать pool_pre_ping-аналоги для своих пулов - функцию-инициализатор воркера, которая проверяет, что окружение поднялось корректно. Падение в инициализаторе ProcessPoolExecutor превращается в BrokenProcessPool сразу, на старте, а не через час работы;
Не забывать про защиту if name == "main" во входной точке программы. На Windows и macOS со стартом spawn её отсутствие приводит к каскадному размножению пулов и красочным крашам, которые потом долго объяснять команде;
Включать максимально подробное логирование в воркерах хотя бы временно, когда воспроизводится плавающая проблема. Трейсбэк, попавший в файл лога дочернего процесса, спасал больше отладочных сессий, чем все остальные инструменты вместе взятые.

В сухом остатке многопроцессорная обработка в Python остаётся одной из тех областей, где знание тонкостей инструментов окупается многократно. ProcessPoolExecutor современнее и предлагает существенно лучше проработанную семантику ошибок. multiprocessing.Pool более низкоуровневый и гибкий, но требует больше внимания к деталям. Выбор между ними правильнее делать осознанно, понимая, как именно будут передаваться сбои между процессами, а не по привычке к одному из API.