Graceful shutdown Python: завершение текущей работы перед остановкой воркера

Остановка океанического пула

Океанический комплекс регистрирует замеры и складывает их в очередь, откуда данные разбирают воркеры-упаковщики. При подаче сигнала на подъём зонд не должен выкачивать всю оставшуюся очередь: диспетчер сразу прекращает приём новых данных и отменяет задачи воркеров.

Если воркер в момент отмены просто ждал очередной замер на queue.get(), он сразу завершает работу. Если же он уже успел извлечь замер и начал сохранение, ему дают упаковать этот конкретный образец, подтвердить его через task_done() и только потом выйти.

Ожидание остановки пула ограничено asyncio.timeout. Если время истекает раньше, чем занятые воркеры закончат начатую упаковку, диспетчер фиксирует таймаут и возвращает список фактически запечатанных замеров вместе со списком брошенных воркеров, чьи задачи так и остались незавершёнными.

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

Что уже дано

При выполнении кода доступны вспомогательные инструменты для моделирования станции и ожидания задач, включая wait_for_task_completion(tasks), фабрику make_shutdown_station с возможностью настройки задержек упаковки образцов и средства наблюдения observe_owned_shutdown. Вам необходимо реализовать только саму асинхронную функцию shutdown_owned_work.

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

Напишите асинхронную функцию shutdown_owned_work(station, workers, timeout_seconds).

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

• station — объект станции. Предоставляет синхронный метод close_admission(), очередь asyncio.Queue в атрибуте queue и список запечатанных замеров в атрибуте sealed. Упаковка замеров реализуется самими воркерами станции: ожидающий воркер завершается при отмене, а уже начавший упаковку воркер доводит её до конца, вызывает queue.task_done() и завершается.
• workers — список переданных запущенных задач (asyncio.Task), представляющих целевых воркеров пула (от 0 до 4 задач).
• timeout_seconds — положительное число секунд на ожидание остановки пула (от 0.01 до 0.1 с).

Правила остановки:

1. При вызове функции немедленно остановите приём замеров станцией вызовом station.close_admission() и запросите отмену всех переданных задач воркеров из workers вызовом метода cancel().
2. Не выкачивайте оставшиеся в очереди элементы и не запускайте обработку новых замеров.
3. Ограничьте ожидание завершения переданных воркеров с помощью контекстного менеджера asyncio.timeout(timeout_seconds). Сами воркеры выполняют полезную работу кооперативно.
4. Функция должна вернуть словарь ровно с тремя ключами:

• 'sealed' — список или список-коллекция запечатанных станцией замеров (station.sealed) в порядке их фактического запечатывания;
• 'pending' — список тех задач из переданного списка workers, у которых на момент возврата из функции task.done() == False (каждая незавершённая задача входит ровно один раз, порядок элементов в списке не фиксирован);
• 'timed_out' — булево значение (bool), показывающее, истёк ли таймаут ожидания asyncio.timeout.

5. Не изменяйте и не отменяйте сторонние фоновые задачи вызывающего кода, не входящие в workers. Не применяйте принудительное завершение, повторные попытки или аварийный сброс очереди — незавершённые воркеры передаются вызывающей стороне как есть.

Ввод и вывод

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

В примере ниже станция инициализируется двумя записями: к моменту вызова функции единственный воркер уже успел забрать замер 'owned' и начал его упаковку, а замер 'queued' остался в очереди. При задержке завершения упаковки в 0.001 с воркер успевает запечатать взятый образец до истечения таймаута в 0.04 с. В результате замер оказывается в списке запечатанных, незавершённых задач не остаётся, признак таймаута равен False, а хвост очереди сохраняет исходный элемент.

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

station = make_shutdown_station(
    ['owned', 'queued'], release_delays=[0.001, None])
observation = observe_owned_shutdown(station, worker_count=1, timeout_seconds=0.04)
print(observation['reported_sealed'])
print(observation['reported_pending_ids'])
print(observation['reported_timed_out'])
print(observation['queue_at_return'])

Вывод:

['owned']
[]
False
['queued']