ProcessPoolExecutor Python: обработка результатов и ошибок сериализации

Реестр исходов пула процессов

Испытательный стенд запускает пакет прикладных расчётов в ProcessPoolExecutor. Часть задач передаётся в виде обычных функций верхнего уровня модуля, но некоторые по ошибке оформлены как анонимные лямбды или вложенные функции. Модуль pickle не может разрешить имена таких объектов при передаче в дочерний процесс, что приводит к ошибкам сериализации.

Метод submit при вызове не падает синхронно: отказ проявляется только позже, когда стенд опрашивает полученные объекты Future. Поэтому системе нужно аккуратно сопоставить исходный идентификатор каждого расчёта с его фактическим исходом — полученным значением либо перехваченным исключением сериализации (PicklingError или AttributeError).

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

Что уже дано

В среде выполнения уже объявлены готовые вспомогательные инструменты для тестирования пулов и сериализации:

• make_process_probe(rows, workers=2) и make_archive_probe(files, failures, workers=2);
• метод probe.run(), который запускает тестируемую функцию в рабочем пространстве имён;
• вспомогательная функция сжатия compress_block(file_id, data).

Вам предстоит реализовать функцию collect_process_outcomes.

Что нужно сделать

Реализуйте функцию collect_process_outcomes(jobs, pool).

Функция принимает два аргумента:

• jobs — упорядоченный список кортежей вида (job_id, function, args) (от 0 до 8 элементов). Поле job_id — уникальный непустой строковый идентификатор из символов ASCII длиной до 30 знаков. function — вызываемый объект (импортируемая функция модуля, лямбда или вложенная функция). args — кортеж позиционных аргументов для функции (числовые значения по модулю не превосходят 1 000 000, списки суммирования содержат до 32 элементов);
• pool — уже созданный экземпляр ProcessPoolExecutor.

Правила выполнения:

• Каждую функцию с её исходными аргументами необходимо отправить в переданный пул pool через метод pool.submit(function, *args) ровно один раз, сохранив связь между job_id и полученным объектом Future.
• Опросите каждый полученный Future на предмет результата или исключения. Ошибка сериализации одной задачи не должна приводить к сбросу или пропуску остальных задач.
• Функция должна вернуть словарь ровно с двумя ключами: "successes" и "errors".
• Ключом словаря "successes" является job_id успешно выполненной задачи, а значением — фактически вычисленный результат. Порядок добавления записей должен строго соответствовать исходному порядку задач во входном списке jobs.
• Ключом словаря "errors" является job_id задачи, завершившейся ошибкой, а значением — точный исходный объект возникшего исключения (PicklingError или AttributeError), сохранённый в Future. Нельзя заменять объект ошибки строкой, типом или заново созданным исключением.
• Если список jobs пуст или в какую-либо категорию не попало ни одной задачи, соответствующий словарь должен быть пустым ({}).

Ввод и вывод

Стандартные потоки stdin и stdout не используются: считывать данные и выводить результат через print не требуется. Тестирующая система импортирует и вызывает функцию верхнего уровня collect_process_outcomes напрямую после выполнения кода решения.

Пример использования

probe = make_process_probe([{'id': 'good', 'kind': 'square', 'value': 6}, {'id': 'bad', 'kind': 'lambda', 'value': 3}])
observed = probe.run()
print(observed['successes'])
print([row['id'] for row in observed['errors']])

Вывод:

[('good', 36)]
['bad']