Обработка except в дочернем процессе пула multiprocessing

Если сделать так, то ошибку покажет:

def job_distributor(task):
    try:
        if task == 'parent':
            while True:
                print('Родитель делает ветку без ошибок')
                time.sleep(300)
        elif task != 'parent':
            print('Дочерний процесс идёт выполнять ветку с ошибкой')
            numbr(task)
    except Exception as ex:
        print(ex)

И здесь тоже ошибка поймается:

def numbr(task):
    try:
        b = int(task) * int(task)
        print(b)
    except Exception as ex:
        print(ex)

Вывод в обоих вариантах:

...
64
Дочерний процесс идёт выполнять ветку с ошибкой
invalid literal for int() with base 10: 'abc'
Дочерний процесс идёт выполнять ветку с ошибкой
81

Но при этом pool.map(job_distributor, a) в try/catch оборачивать без толку, вот так ничего не поймается:

try:
    pool = Pool(2)
    pool.map(job_distributor, a)
    print(pool)
except Exception as ex:
    print(ex)

Вызов pool.map у вас ведь не заканчивается просто потому, что "parent" у вас висит в бесконечном цикле while в отдельном потоке, а pool.map ждёт завершения этого потока. Не делайте так больше, pool.map не предназначен для того, чтобы создавать процессы, остающиеся висеть в памяти, для этого есть другие механизмы. Если же убрать бесконечный цикл и уменьшить паузу, то и этот фрагмент кода нормально поймает исключение после завершения выполнения pool.map:

64
Дочерний процесс идёт выполнять ветку с ошибкой
Дочерний процесс идёт выполнять ветку с ошибкой
81
Дочерний процесс идёт выполнять ветку с ошибкой
1
invalid literal for int() with base 10: 'abc'

Код я проверял в Google Colab, если что.

P.S. Странно, но перемещение try чуть ниже сразу ловит Exception даже не смотря на бесконечный while и тут же прекращает работу pool.map:

    pool = Pool(2)
try:
    pool.map(job_distributor, a)
    print(pool)
except Exception as ex:
    print(ex)

Вывод:

invalid literal for int() with base 10: 'abc'

Возможно, это особенности ipython в Google Colab, надо проверять на других интерпретаторах.

Во-первых, замечание по терминологии. В операционных системах уже есть отношение между процессами "родитель-потомок" и соответствующие понятия. В этом смысле в примере, который вы привели, есть один родительский процесс - тот в котором выполняется создание пула и pool.map(job_distributor, a). А все остальные процессы - дочерние. Они по отношению друг к другу - братья (англ. siblings). Это все с точки зрения ОС.

То что вы выделяете один из дочерних процессов как-то отдельно (назовем его координатором) - это логика вашей программы и не относится к ОС. В данном случае это важно и является сутью проблемы.

То что исключение возникшее в одном процессе не попадает в родительский или другой процесс в пуле это нормальное поведение.

Процессы сами по себе не умеют передавать исключений в принципе. Исключения и их обработка происходит в рамках одного потока в одном процессе. Если вы хотите передать какую-то информацию из одного процесса в другой (например, информацию о том, что возникло исключение, какое именно и где именно), нужно воспользоваться механизмами межпроцессного взаимодействия из multiprocessing такими как multiprocessing.Queue или multiprocessing.Pipe. В случае исключений это означает что, во-первых, нужно это исключение поймать в том потоке того процесса, где оно случилось, а, во-вторых, передать его куда нужно.

multiprocessing.Pool уже реализует это, но только от дочернего (в смысле ОС) процесса (грубо говоря из функции которую запускаете в пуле) к родителю (тому коду который запускал).

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

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

1. перед запуском процессов создаем Queue.
2. передаем ее все процессам
3. обработчики (в вашей терминологии дочерние процессы) перехватывают все исключения и записывают их и дополнительную информацию (что за задача выполнялась, stacktrace и т.д.) в эту очередь.
4. координатор (в вашей терминологии родитель) читает из этой очереди и решает, что с этим делать

if __name__ == '__main__':
    freeze_support()

    with Pool(2) as p:
        fut = {p.apply_async(job_distributor, args=[x]): x for x in a}
        while fut:
            for f in fut.copy():
                try:
                    r = f.get(1)
                except multiprocessing.TimeoutError:
                    continue
                except Exception as e:
                    # raise RuntimeError([fut[f], f]) from e
                    logging.exception(RuntimeError([fut[f], f]))
                else:
                    del fut[f]

вывод:

ERROR:root:['abc', <multiprocessing.pool.ApplyResult object at 0x0000024759E21190>]
multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "D:\python-3.8.6-amd64\lib\multiprocessing\pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
  File "E:\SCR\NEW_STAND\ii.py", line 21, in job_distributor
    numbr(task)
  File "E:\SCR\NEW_STAND\ii.py", line 9, in numbr
    b = int(task) * int(task)
ValueError: invalid literal for int() with base 10: 'abc'
"""
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "E:/SCR/NEW_STAND/ii.py", line 40, in <module>
    r = f.get(1)
  File "D:\python-3.8.6-amd64\lib\multiprocessing\pool.py", line 771, in get
    raise self._value
ValueError: invalid literal for int() with base 10: 'abc'
ERROR:root:['abc', <multiprocessing.pool.ApplyResult object at 0x0000024759E21190>]
multiprocessing.pool.RemoteTraceback: 
"""