Asyncio shield: сохранение начатой работы при повторной отмене

Экстренная эвакуация линии фасовки

Автоматизированная линия фасовки распределяет поступающие партии грузов по воркерам-упаковщикам через общую очередь. При получении сигнала тревоги диспетчер немедленно прекращает приём новых партий и отменяет задачи всех воркеров.

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

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

Помогите диспетчеру скоординировать экстренную остановку линии: выясните, уложилась ли остановка в установленный лимит времени, сколько коробок рабочие успели упаковать до отмены и какие именно партии удалось вернуть в аварийный буфер.

Что уже дано

При выполнении кода доступны готовые классы и вспомогательные функции имитации линии:

• объект линии line с очередью line.queue (содержит партии со строковым полем 'id' длиной от 1 до 40 символов и целочисленным 'boxes' от 0 до 1000; изначально в очереди от 0 до 12 партий);
• корутина воркера line.worker(), выполняющая упаковку и защищённую от повторных отмен эвакуацию партии в line.recovered;
• методы line.close_admission() для закрытия приёма новых партий и line.shutdown_timeout(delay), возвращающий контекстный менеджер asyncio.timeout;
• счётчик упакованных коробок line.completed_boxes и список спасённых партий line.recovered;
• функции make_recovery_probe(spec) и observe_recovery_run(probe, coordinator_coroutine) для локальной проверки сценария.

Вам необходимо реализовать только координирующую функцию evacuate_line.

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

Реализуйте асинхронную функцию evacuate_line(line, worker_count, stop_signal, stop_timeout).

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

• line — объект линии RecoveryLine;
• worker_count — целое число от 1 до 4, определяющее точное количество независимых задач воркеров, которые нужно запустить через asyncio.create_task для корутины line.worker();
• stop_signal — объект asyncio.Event, устанавливаемый извне при сигнале тревоги;
• stop_timeout — вещественное число от 0.001 до 1.0 (секунды), задающее лимит времени на ожидание финализации воркеров.

Порядок и правила выполнения эвакуации:

1. Запустите ровно worker_count задач со стандартной корутиной line.worker() и дождитесь установки внешнего события stop_signal.
2. Сразу после сигнала тревоги вызовите line.close_admission(), закрывая приём новых задач, и запросите отмену всех созданных задач воркеров с помощью вызова .cancel() на каждой из них. Не пытайтесь предварительно вычитывать или вырабатывать оставшиеся в очереди неразобранные партии.
3. Ограничьте ожидание завершения пула воркеров с помощью контекстного менеджера line.shutdown_timeout(stop_timeout). Если интервал времени истёк, зафиксируйте флаг превышения таймаута.
4. Превышение таймаута не разрешает бросать задачи воркеров незавершёнными: обязательно дождитесь фактического завершения всех запущенных задач и выполнения их защищённых процедур возврата. На момент выхода из evacuate_line ни одна из созданных задач не должна оставаться в состоянии ожидания (pending).
5. Функция должна вернуть словарь строго следующего вида: {"timed_out": bool, "packed": int, "recovered": list[dict]}, где:

• timed_out — истинно (True), если контекст таймаута истёк до завершения воркеров, иначе False;
• packed — итоговое количество успешно упакованных коробок из line.completed_boxes;
• recovered — список партий, спасённых в line.recovered (каждая взятая в работу и прерванная партия сохраняется ровно один раз; порядок элементов в списке произвольный).

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

Ввод и вывод

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

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

probe = make_recovery_probe({'name': 'demo', 'workers': 1, 'jobs': [{'id': 'demo_0', 'boxes': 2}, {'id': 'demo_1', 'boxes': 7}], 'completed_before_alarm': 1, 'contended_recovery_lock': False, 'repeat_worker_cancel': False, 'stop_timeout': 1.0, 'lock_hold': 0.06, 'expected_timeout': False})
summary = observe_recovery_run(probe, evacuate_line(probe.line, 1, probe.stop_signal, 1.0))["summary"]
print(summary["timed_out"], summary["packed"])
print([item["id"] for item in summary["recovered"]])

Вывод:

False 2
['demo_1']