Потоки и процессы Python: аварийная остановка двухэтапного конвейера

Координатор аварийного останова конвейера

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

После срабатывания останова координатор больше не отправляет новые задачи в пул процессов и пытается снять уже переданные вычислительные задания через Future.cancel(). Если вызов вернул True, задача считается отменённой ('cancelled'). Если же cancel() вернул False, задание успело начаться или полностью завершиться к моменту отмены; оно доходит до своего естественного финала и получает статус 'processed'.

Когда оба пула полностью завершают работу, координатор обязан сопоставить каждому входному пакету один из пяти взаимоисключающих статусов, не нарушая их исходный порядок. Пакет помечается как 'io_error', если чтение завершилось ошибкой; как 'processed', если вычисления в пуле процессов завершились успешно или с программной ошибкой; как 'cancelled', если его расчётный Future удалось отменить; как 'unsubmitted', если данные были успешно считаны, но не ушли на обработку из-за останова конвейера; либо как 'aborted_read', если сетевое чтение для него даже не начиналось.

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

Что уже дано

В заготовке представлена сигнатура функции coordinate_cutoff(packet_ids, read_packet, analyze_packet, io_pool, cpu_pool, byte_limit), которую вам предстоит реализовать.

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

• make_cutoff_probe(rows, io_workers=2, cpu_workers=1, related_values=False) — создаёт тестовый сценарий с реальными пулами и функциями чтения/анализа;
• observe_cutoff_pipeline(probe, byte_limit) — запускает координатор и собирает системные наблюдения о вызовах и переходах состояний.

Реализовывать эти вспомогательные функции самостоятельно не нужно.

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

Реализуйте функцию coordinate_cutoff(packet_ids, read_packet, analyze_packet, io_pool, cpu_pool, byte_limit) -> list[str].

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

• packet_ids — список от 0 до 16 уникальных строковых идентификаторов пакетов (длина каждого идентификатора — от 1 до 20 ASCII-символов);
• read_packet — вызываемый объект read_packet(packet_id), возвращающий байты (от 0 до 64 байт на пакет) или возбуждающий исключение чтения. Задачу чтения пакета необходимо отправлять в io_pool;
• analyze_packet — вызываемый объект analyze_packet(packet_id, data), выполняющий вычислительный анализ и способный завершиться ошибкой. Задачу вычисления с исходными байтами необходимо отправлять в cpu_pool;
• io_pool — переданный экземпляр ThreadPoolExecutor (1–2 рабочих потока), изначально без отправленных задач;
• cpu_pool — переданный экземпляр ProcessPoolExecutor (1–2 рабочих процесса), изначально без отправленных задач;
• byte_limit — неотрицательное целое число (от 0 до 1024), пороговый предел суммарного объёма успешно извлечённых байтов.

Правила координации:

• Задачи сетевого чтения выполняются в io_pool, а вычисления — в cpu_pool с использованием переданных колбэков.
• Порядок опроса результатов чтения через вызовы методов result() или exception() у соответствующих Future определяет момент наступления останова.
• Аварийный останов конвейера фиксируется при наступлении любого из двух событий: при обнаружении первой ошибки чтения (exception() is not None) либо когда накопленный объём успешно считанных байтов становится строго больше byte_limit.
• До момента фиксации останова успешно считанные пакеты разрешено передавать в cpu_pool. Как только останов зафиксирован, отправка новых вычислительных задач в cpu_pool строго запрещается.
• В момент первого срабатывания останова координатор обязан вызвать .cancel() для всех задач, которые к этому моменту были отправлены в cpu_pool (включая уже завершившиеся к этому времени задачи). Если .cancel() возвращает True, задача признаётся успешно отменённой. Если возвращается False, задача либо уже исполнялась, либо успела завершиться до отмены, и она продолжает выполнение штатно.
• Не начатые к моменту останова операции чтения могут быть отменены или не запущены.
• Координатор обязан гарантированно закрыть оба пула, вызвав для каждого из них метод .shutdown(wait=True).
• Функция должна вернуть список строковых статусов длины, равной len(packet_ids), строго в порядке следования идентификаторов во входном списке packet_ids. Допустимы ровно пять значений статуса:
• 'io_error' — фактическая ошибка при чтении пакета;
• 'processed' — вычислительная задача была отправлена в cpu_pool и не была отменена (включая случаи, когда сама функция analyze_packet выбросила исключение);
• 'cancelled' — задача была передана в cpu_pool, но вызов Future.cancel() вернул True;
• 'unsubmitted' — пакет был успешно прочитан, но не был передан в cpu_pool из-за наступления останова;
• 'aborted_read' — сетевое чтение для пакета так и не было начато из-за останова конвейера.

Ввод и вывод

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

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

probe = make_cutoff_probe([
    {'id': 'first', 'data': [1, 2]},
    {'id': 'second', 'data': [3]},
])
observation = observe_cutoff_pipeline(probe, 3)
print(observation['summary'])

Вывод:

['processed', 'processed']