Остановка пула процессов Python: Queue, маркеры завершения и join

Остановка пула процессов с общей очередью

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

Для этого была написана функция остановки, однако в ней отправка маркера None и вызов join() объединены в один общий цикл и выполняются попарно для каждого воркера по очереди. Поскольку общая очередь не привязана к конкретным исполнителям, отправленный маркер легко перехватывает произвольный активный процесс, а вовсе не тот, чей join() ожидает диспетчер прямо сейчас. Из-за этого система может зависнуть намертво: диспетчер ждёт завершения процесса, до которого сигнал остановки ещё не дошёл.

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

Что уже дано

В редакторе уже содержится исходный код функции stop_process_pool(work_queue, workers), допускающий зависание при конкурентной выборке из очереди.

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

• make_pool_stop_probe(worker_count, documents, delayed_first_reader=False) — создаёт тестовый сценарий с реальными процессами и общей очередью;
• observe_pool_stop(probe) — выполняет функцию остановки на созданном сценарии и возвращает структуру с результатами наблюдений;
• make_pool_stop_example_observation() — запускает готовый типовой сценарий и возвращает его результаты.

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

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

Исправьте функцию stop_process_pool(work_queue, workers).

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

• work_queue — реальный экземпляр multiprocessing.Queue, общий для всех запущенных воркеров. В очередь заранее помещено от 0 до 24 документов (произвольные JSON-совместимые значения, кроме None верхнего уровня; ложные значения и вложенные None считаются обычными документами). Новые задачи в очередь больше не поступают.
• workers — непустой список из 1–4 уже запущенных процессов multiprocessing.Process. Каждый процесс в цикле читает документы из общей очереди, а при извлечении первого встреченного значения None верхнего уровня завершает работу.

Требования к поведению функции:

• К моменту возврата из функции каждый из переданных процессов workers должен штатно завершиться.
• Каждый документ, находившийся в очереди до начала остановки, должен быть обработан ровно один раз без изменений.
• Значения в очереди не привязаны к конкретным процессам: любой активный воркер может забрать как очередной документ, так и отправленный сигнал остановки.
• Время выполнения функции ограничено бюджетом в 5 секунд на каждый вызов (допускаются безопасные ожидания с таймаутом в пределах этого лимита).
• Возвращаемое значение функции не используется.

Ввод и вывод

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

В примере ниже создаётся сценарий с двумя воркерами и двумя документами в общей очереди. Функция observe_pool_stop передаёт очередь и процессы в stop_process_pool, после чего фиксирует состояние системы: флаг all_stopped подтверждает, что оба процесса штатно завершили работу, а processed_count показывает, что оба документа были успешно обработаны.

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

worker_count = 2
documents = ['draft.txt', 'notes.txt']
probe = make_pool_stop_probe(worker_count, documents)
observed = observe_pool_stop(probe)
print(observed['all_stopped'])
print(observed['processed_count'])

Вывод:

True
2