Concurrent.futures Python: выбор исполнителя по размеру пакета

Гибридный маршрутизатор пакетов

Аналитическая станция принимает входящие пакеты логов, распаковывает их по сети через ThreadPoolExecutor и выполняет для каждого пакета математическое сжатие. Чтобы распределять нагрузку эффективно, координатор оценивает объём данных сразу после завершения соответствующего сетевого Future.

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

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

Что уже дано

В среде выполнения уже объявлена функция route_packets(packet_ids, fetch, compressors, threshold, io_pool, cpu_pool), которую необходимо реализовать. Вспомогательные функции observe_router_probe и make_router_example_observation доступны во время исполнения и служат для проверки работы диспетчера.

Для наглядной демонстрации работы также доступен контекстный менеджер make_router_example_case(buffers, fail_ids=()). Он принимает словарь отображения идентификаторов пакетов в их распакованные байты и опциональный кортеж идентификаторов сбоящих сжатий, предоставляя готовые атрибуты для прямого запуска: case.fetch, словарь сериализуемых функций сжатия case.compressors (которые применяют zlib.compress к байтам), а также реальные экземпляры case.io_pool и case.cpu_pool с автоматическим освобождением ресурсов при выходе из блока with.

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

Реализуйте функцию route_packets(packet_ids, fetch, compressors, threshold, io_pool, cpu_pool).

Параметры функции:

• packet_ids — список уникальных целочисленных идентификаторов пакетов (от 0 до 12 штук);
• fetch — готовая функция fetch(packet_id), возвращающая распакованные байты пакета (длиной от 0 до 8192 байт);
• compressors — словарь, сопоставляющий каждый packet_id функции сжатия compress(data). При успехе она возвращает сжатые bytes, а при сбое выбрасывает исключение Exception;
• threshold — положительный целочисленный порог размера $S$ в байтах (от 1 до 8192);
• io_pool — переданный экземпляр concurrent.futures.ThreadPoolExecutor;
• cpu_pool — переданный экземпляр concurrent.futures.ProcessPoolExecutor.

Правила диспетчеризации и обработки:

• Загрузка каждого пакета (fetch(packet_id)) должна выполняться в переданном пуле потоков io_pool.
• Маршрутизация каждого пакета на сжатие выполняется сразу по завершении его сетевой загрузки, не дожидаясь окончания загрузки других пакетов.
• Если фактическая длина распакованных данных len(data) строго меньше threshold ($<S$), операция сжатия отправляется в io_pool. Если длина равна или превышает порог ($\ge S$), сжатие отправляется в cpu_pool.
• Сбои сжатия изолированы: если функция компрессора выбрасывает исключение, нужно сохранить имя типа этого исключения и продолжить обработку остальных пакетов.
• Переданные пулы io_pool и cpu_pool принадлежат вызывающей стороне — закрывать их (shutdown) запрещено.

Функция должна вернуть словарь, сопоставляющий каждый packet_id словарю со следующими ключами:

• 'executor' (str) — строка 'thread' или 'process' в зависимости от того, в каком пуле реально выполнялось сжатие данных этого пакета;
• 'value' (bytes | None) — полученные байты сжатия при успехе, либо None при ошибке сжатия;
• 'error_type' (str | None) — строковое имя типа исключения (например, 'CompressionError') при сбое, либо None при успешном сжатии.

Порядок ключей в результирующем словаре роли не играет. Если передан пустой список packet_ids, функция должна вернуть пустой словарь {}.

Ввод и вывод

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

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

buffers = {10: b'ab', 20: b'abcde'}
threshold = 5
with make_router_example_case(buffers) as case:
    result = route_packets([10, 20], case.fetch, case.compressors, threshold,
                           case.io_pool, case.cpu_pool)
    for packet_id in [10, 20]:
        row = result[packet_id]
        print(packet_id, row['executor'], row['value'], row['error_type'])

Вывод:

10 thread b'x\x9cKL\x02\x00\x01&\x00\xc4' None
20 process b'x\x9cKLJNI\x05\x00\x05\xc8\x01\xf0' None

В примере длина пакета 10 равна 2 байтам, что строго меньше threshold = 5, поэтому сжатие выполнено в пуле потоков ('thread'). Длина пакета 20 равна 5 байтам, что в точности равно порогу ($\ge 5$), поэтому задача ушла в пул процессов ('process'). В обоих случаях сжатие завершилось успешно: поле 'value' содержит результирующие байты, а 'error_type' равен None.