Поиск причины BrokenProcessPool в concurrent.futures Python

Кратко

Я получаю исключение BrokenProcessPool при распараллеливании моего кода с concurrent.futures. Дальнейшая ошибка не отображается. Я хочу найти причину ошибки и попросить идеи, как это сделать.

Полная проблема

Я использую concurrent.futures для распараллеливания некоторого кода.

with ProcessPoolExecutor() as pool:
    mapObj = pool.map(myMethod, args)

Я получаю (и только) следующее исключение:

concurrent.futures.process.BrokenProcessPool: A child process terminated abruptly, the process pool is not usable anymore

К сожалению, программа сложная и ошибка появляется только после 30 минут работы программы. Поэтому я не могу предоставить хороший минимальный пример.

Чтобы найти причину проблемы, я обернул метод, который запускаю параллельно, блоком try-except:

def myMethod(*args):
    try:
        ...
    except Exception as e:
        print(e)

Проблема осталась прежней, и блок исключений так и не был введен. Я делаю вывод, что исключение не исходит из моего кода.

Следующим моим шагом было написать собственный класс ProcessPoolExecutor, который является дочерним по отношению к исходному ProcessPoolExecutor и позволяет мне заменить некоторые методы на настраиваемые. Я скопировал и вставил исходный код метода _process_worker и добавил несколько операторов печати.

def _process_worker(call_queue, result_queue):
    """Evaluates calls from call_queue and places the results in result_queue.
        ...
    """
    while True:
        call_item = call_queue.get(block=True)
        if call_item is None:
            # Wake up queue management thread
            result_queue.put(os.getpid())
            return
        try:
            r = call_item.fn(*call_item.args, **call_item.kwargs)
        except BaseException as e:
                print("??? Exception ???")                 # newly added
                print(e)                                   # newly added
            exc = _ExceptionWithTraceback(e, e.__traceback__)
            result_queue.put(_ResultItem(call_item.work_id, exception=exc))
        else:
            result_queue.put(_ResultItem(call_item.work_id,
                                         result=r))

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

Теперь мне не хватает идей, как я могу найти ошибку. Здесь возникает исключение:

def submit(self, fn, *args, **kwargs):
    with self._shutdown_lock:
        if self._broken:
            raise BrokenProcessPool('A child process terminated '
                'abruptly, the process pool is not usable anymore')
        if self._shutdown_thread:
            raise RuntimeError('cannot schedule new futures after shutdown')

        f = _base.Future()
        w = _WorkItem(f, fn, args, kwargs)

        self._pending_work_items[self._queue_count] = w
        self._work_ids.put(self._queue_count)
        self._queue_count += 1
        # Wake up queue management thread
        self._result_queue.put(None)

        self._start_queue_management_thread()
        return f

Здесь пул процессов должен быть разбит:

def _queue_management_worker(executor_reference,
                             processes,
                             pending_work_items,
                             work_ids_queue,
                             call_queue,
                             result_queue):
    """Manages the communication between this process and the worker processes.
        ...
    """
    executor = None

    def shutting_down():
        return _shutdown or executor is None or executor._shutdown_thread

    def shutdown_worker():
        ...

    reader = result_queue._reader

    while True:
        _add_call_item_to_queue(pending_work_items,
                                work_ids_queue,
                                call_queue)

        sentinels = [p.sentinel for p in processes.values()]
        assert sentinels
        ready = wait([reader] + sentinels)
        if reader in ready:
            result_item = reader.recv()
        else:                               #THIS BLOCK IS ENTERED WHEN THE ERROR OCCURS
            # Mark the process pool broken so that submits fail right now.
            executor = executor_reference()
            if executor is not None:
                executor._broken = True
                executor._shutdown_thread = True
                executor = None
            # All futures in flight must be marked failed
            for work_id, work_item in pending_work_items.items():
                work_item.future.set_exception(
                    BrokenProcessPool(
                        "A process in the process pool was "
                        "terminated abruptly while the future was "
                        "running or pending."
                    ))
                # Delete references to object. See issue16284
                del work_item
            pending_work_items.clear()
            # Terminate remaining workers forcibly: the queues or their
            # locks may be in a dirty state and block forever.
            for p in processes.values():
                p.terminate()
            shutdown_worker()
            return
        ...

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

Я использую Python 3.5 на 64-битном Linux.


person Samufi    schedule 03.01.2017    source источник
comment
я получил эту ошибку, и этот пост прояснил мою проблему. stackoverflow .com/questions/15900366/   -  person kmh    schedule 14.09.2017
comment
У меня такая же ошибка, код выхода из нескольких процессов - -11. Хотя эта же функция отлично работает в многопоточном режиме.   -  person WeiChing 林煒清    schedule 17.10.2019


Ответы (2)


Я думаю, что смог получить как можно дальше:

Я изменил метод _queue_management_worker в моем измененном модуле ProcessPoolExecutor таким образом, чтобы выводился код выхода сбойного процесса:

def _queue_management_worker(executor_reference,
                             processes,
                             pending_work_items,
                             work_ids_queue,
                             call_queue,
                             result_queue):
    """Manages the communication between this process and the worker processes.
        ...
    """
    executor = None

    def shutting_down():
        return _shutdown or executor is None or executor._shutdown_thread

    def shutdown_worker():
        ...

    reader = result_queue._reader

    while True:
        _add_call_item_to_queue(pending_work_items,
                                work_ids_queue,
                                call_queue)

        sentinels = [p.sentinel for p in processes.values()]
        assert sentinels
        ready = wait([reader] + sentinels)
        if reader in ready:
            result_item = reader.recv()
        else:                               

            # BLOCK INSERTED FOR DIAGNOSIS ONLY ---------
            vals = list(processes.values())
            for s in ready:
                j = sentinels.index(s)
                print("is_alive()", vals[j].is_alive())
                print("exitcode", vals[j].exitcode)
            # -------------------------------------------


            # Mark the process pool broken so that submits fail right now.
            executor = executor_reference()
            if executor is not None:
                executor._broken = True
                executor._shutdown_thread = True
                executor = None
            # All futures in flight must be marked failed
            for work_id, work_item in pending_work_items.items():
                work_item.future.set_exception(
                    BrokenProcessPool(
                        "A process in the process pool was "
                        "terminated abruptly while the future was "
                        "running or pending."
                    ))
                # Delete references to object. See issue16284
                del work_item
            pending_work_items.clear()
            # Terminate remaining workers forcibly: the queues or their
            # locks may be in a dirty state and block forever.
            for p in processes.values():
                p.terminate()
            shutdown_worker()
            return
        ...

После этого я посмотрел значение кода выхода:

from multiprocessing.process import _exitcode_to_name
print(_exitcode_to_name[my_exit_code])

где my_exit_code — это код выхода, напечатанный в блоке, который я вставил в _queue_management_worker. В моем случае код был -11, что означает, что я столкнулся с ошибкой сегментации. Поиск причины этой проблемы будет огромной задачей, но выходит за рамки этого вопроса.

person Samufi    schedule 04.01.2017

Если вы используете macOS, существует известная проблема, связанная с тем, как некоторые версии macOS используют разветвление, которое Python в некоторых сценариях не считает безопасным для разветвления. Обходной путь, который сработал для меня, — использовать переменную среды no_proxy.

Отредактируйте ~/.bash_profile и включите следующее (может быть лучше указать здесь список доменов или подсетей вместо *)

no_proxy='*'

Обновить текущий контекст

source ~/.bash_profile

Мои локальные версии, с которыми проблема была замечена и устранена: Python 3.6.0 на macOS 10.14.1 и 10.13.x.

Источники: Ошибка 30388 Ошибка 27126

person gowthamnvv    schedule 12.11.2018
comment
Та же проблема с MacOS 10.14.6 (18G87) и Python 3.7.2. - person joe; 29.08.2019