Concurrent.futures Python: первые K успешных результатов двух пулов

Кворум опережающего опроса

Поисковый индекс обновляет данные параллельно из двух каналов: через сетевое чтение удалённых зеркал в ThreadPoolExecutor и локальную генерацию векторов в ProcessPoolExecutor. Клиент запрашивает пакет из $K$ успешно вычисленных значений, поэтому диспетчер объединяет разнородные объекты Future и отслеживает их готовность через общий as_completed.

В зачёт кворума идут только успешные ответы future.result() без исключений: упавшие с ошибкой или отменённые задачи пропускаются, а среди одновременно готовых объектов сохраняется порядок их извлечения итератором. Как только набрано ровно $K$ валидных результатов, диспетчер отменяет оставшиеся ожидающие Future, закрывает оба пула с ожиданием завершения уже запущенной работы через shutdown(wait=True) и возвращает список пар «источник и результат» в порядке их поступления. Если же все задачи в пулах исчерпаны раньше, чем удалось набрать $K$ значений, выполнение не прерывается ошибкой: возвращаются все успевшие накопиться валидные данные вместе с признаком недобора кворума.

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

Что уже дано

В редакторе представлена заготовка функции collect_success_quorum(io_pool, cpu_pool, future_sources, k). При выполнении кода в окружении уже доступны вспомогательные инструменты make_quorum_probe и observe_success_quorum для локальных проверок; реализовывать их не требуется.

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

Напишите тело функции collect_success_quorum(io_pool, cpu_pool, future_sources, k).

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

• io_pool — экземпляр ThreadPoolExecutor, управляющий задачами ввода-вывода;
• cpu_pool — экземпляр ProcessPoolExecutor, управляющий вычислительными задачами;
• future_sources — словарь, сопоставляющий ранее созданные объекты concurrent.futures.Future со строками их источников: 'io' или 'cpu';
• k — положительное целое число необходимых успешных результатов (от 1 до 20).

Общее количество задач во future_sources составляет от 0 до 12, а число рабочих потоков и процессов — от 1 до 2 на каждый пул.

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

• Выполните один общий обход всех переданных Future с помощью concurrent.futures.as_completed. Порядок извлечения объектов из этого итератора (включая одновременную готовность) должен строго сохраняться.
• Для каждого полученного Future запросите значение через метод .result(). Пропускайте задачи, которые были отменены (future.cancelled()), а также задачи, вызов .result() у которых возбуждает исключение подкласса Exception.
• Любые успешно возвращённые значения (включая 0, False, пустые строки, пустые коллекции и None) считаются валидными и идут в зачёт кворума.
• Как только набрано ровно k успешных результатов, немедленно прекратите сбор, вызовите метод .cancel() у всех оставшихся ожидающих объектов Future и закройте оба пула вызовом shutdown(wait=True), позволяя уже исполняющимся задачам корректно завершиться.
• Если все переданные задачи исчерпаны раньше, чем удалось набрать k успехов, также закройте оба пула через shutdown(wait=True) и зафиксируйте недобор кворума.
• Функция должна вернуть словарь ровно с двумя ключами:
• 'results' — список кортежей вида (source, value), где source — строка 'io' или 'cpu', а value — полученный результат, в порядке их успешного извлечения;
• 'reached' — булево значение: True, если набрано ровно k успешных значений, и False, если кворум не достигнут.

Не создавайте новые пулы, не перезапускайте упавшие задачи и не генерируйте искусственные результаты.

Ввод и вывод

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

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

probe = make_quorum_probe({
    'io': [{'value': 'mirror'}],
    'cpu': [{'value': [1, 2]}],
})
observation = observe_success_quorum(probe, 2)
print(observation['summary']['reached'])
print(len(observation['summary']['results']))

Вывод:

True
2