ThreadPoolExecutor и ProcessPoolExecutor: загрузка и обработка данных

Конвейер карты высот

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

Однако некоторые варианты алгоритмов интерполяции передаются в виде локальных функций или замыканий. При попытке получить результат из их Future в пуле процессов возникает отложенная ошибка сериализации pickle. Сервис строго разделяет причины сбоев: если проверка упавшей задачи подтверждает, что не удалось сериализовать аргументы или саму вызываемую функцию, сервис запускает для этого фрагмента запасную триангуляцию в ThreadPoolExecutor и отмечает использование резерва. Если же внутри алгоритма произошла вычислительная ошибка, она считается неустранимым браком фрагмента и не должна приводить к запуску резервных вычислений.

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

Что уже дано

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

• is_submission_serialization_failure(future) — возвращает True, если задача пула процессов завершилась доказанным системным сбоем сериализации переданной функции или её аргументов при отправке в очередь воркеров;
• make_fallback_example_case(specifications) — контекстный менеджер для локального тестирования конвейера, принимающий словарь спецификаций фрагментов. Внутри контекста доступны готовые колбэки fetch, interpolate (вычисляет сумму квадратов значений фрагмента), backup (вычисляет обычную сумму значений фрагмента), а также действующие пулы io_pool и cpu_pool;
• observe_fallback_probe(specifications) и make_fallback_example_observation() — инструменты для проверки работы конвейера.

Вам предстоит написать тело функции build_height_map.

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

Реализуйте функцию build_height_map(fragment_ids, fetch, interpolators, backup, io_pool, cpu_pool).

Функция принимает шесть аргументов:

• fragment_ids — список уникальных целочисленных идентификаторов фрагментов (от 0 до 12 идентификаторов);
• fetch — функцию fetch(fragment_id), возвращающую сырые данные фрагмента (содержит от 0 до 32 целочисленных значений в диапазоне от -1000 до 1000);
• interpolators — отображение идентификаторов фрагментов на соответствующие им функции основной интерполяции от одного аргумента interpolator(fragment);
• backup — функцию backup(fragment) для резервного расчета в пуле потоков из того же скачанного объекта фрагмента;
• io_pool — экземпляр concurrent.futures.ThreadPoolExecutor для скачивания и резервных расчетов;
• cpu_pool — экземпляр concurrent.futures.ProcessPoolExecutor для попыток основной интерполяции.

Порядок и правила обработки:

1. Загрузка каждого фрагмента из списка fragment_ids выполняется через переданный io_pool вызовом fetch(fragment_id).
2. После успешной загрузки фрагмента запустите основную интерполяцию в cpu_pool: вызовите соответствующую функцию из interpolators, передав ей полученный объект фрагмента.
3. Если Future из пула процессов завершился успехом, сохраните вычисленное значение (value), установив fallback=False и error=None.
4. Если Future завершился ошибкой, проверьте причину с помощью предоставленной функции is_submission_serialization_failure(future):

• При подтверждении сбоя сериализации задачи или аргументов запустите для этого фрагмента функцию backup(fragment) в пуле потоков io_pool. В результат запишите полученное значение резервного расчета, флаг fallback=True и error=None.
• Если проверка не подтвердила сбой сериализации (включая любые вычислительные исключения внутри самого алгоритма, такие как AttributeError, ValueError или даже возбуждённый внутри тела функции PicklingError), резервный расчет не запускается. В результат запишите value=None, fallback=False, а в поле error сохраните исходный объект исключения из Future пула процессов.

5. Обработайте все переданные идентификаторы. Завершать или закрывать переданные пулы io_pool и cpu_pool (вызывать shutdown) запрещено: пулы принадлежат вызывающей стороне.

Функция должна вернуть словарь, сопоставляющий каждый идентификатор из fragment_ids с итоговым словарем вида: {'value': ..., 'fallback': ..., 'error': ...}. Порядок ключей в словаре не имеет значения. Если передан пустой список идентификаторов, верните пустой словарь {}.

Ввод и вывод

Стандартные потоки ввода stdin и вывода stdout в решении не используются: считывать ввод и вызывать print() не нужно. После выполнения кода учащегося проверяющая система считывает из пространства имён модуля функцию верхнего уровня build_height_map и вызывает её с проверочными аргументами.

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

def make_local_interpolator():
    def interpolate(fragment):
        return sum(value * value for value in fragment['values'])
    return interpolate


specifications = {10: {'values': [2, 3]}, 20: {'values': [2, 3]}}
with make_fallback_example_case(specifications) as case:
    interpolators = {10: case.interpolate, 20: make_local_interpolator()}
    result = build_height_map([10, 20], case.fetch, interpolators, case.backup,
                              case.io_pool, case.cpu_pool)
    for fragment_id in [10, 20]:
        row = result[fragment_id]
        print(fragment_id, row['value'], row['fallback'], row['error'])

Вывод:

10 13 False None
20 5 True None

Для фрагмента 10 функция case.interpolate успешно выполняется в пуле процессов: сумма квадратов значений [2, 3] даёт 13, резервный расчёт не требуется (fallback=False), ошибок нет (error=None). Для фрагмента 20 передана локальная функция make_local_interpolator(), которая не может быть сериализована для пула процессов. Диагностика подтверждает ошибку сериализации вызова, поэтому конвейер выполняет резервную функцию case.backup в пуле потоков, вычисляя сумму 2 + 3 = 5, отмечая срабатывание резерва (fallback=True) и завершая обработку без ошибки (error=None).