Asyncio Queue: обработка ошибок воркеров и подтверждение taskdone

Конвейер досмотра багажа

Автоматизированный пункт досмотра направляет багаж через очередь в сканер, где одновременно может проверяться не более заданного числа мест. Несколько воркеров-инспекторов забирают чемоданы из очереди и выполняют асинхронную проверку, при этом любое место багажа способно вызвать аварийный сигнал сканера.

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

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

Что уже дано

В редакторе представлена заготовка асинхронной функции screen_baggage(station, scanner, worker_count), которую необходимо реализовать.

Объект пункта досмотра station создаётся фабрикой make_baggage_station(bags, max_scans) и предоставляет следующие атрибуты и методы:

• station.queue — очередь asyncio.Queue, в которую уже помещены все кортежи багажа (новые элементы извне не поступают);
• station.slots — семафор asyncio.Semaphore, ограничивающий одновременные проверки числом слотов сканера max_scans (от 1 до 16);
• station.lock — замок asyncio.Lock для синхронизации записи в весовую сводку;
• station.total — целочисленный счётчик суммарного веса (изначально 0);
• station.failures — список для фиксации сбоев проверки (изначально пустой);
• station.start_workers(worker_factory, worker_count, *, task_group=None) — готовый вспомогательный метод создания пула воркеров. Он принимает фабрику worker_factory (вызываемый объект без аргументов, возвращающий новую корутину одного воркера), создаёт ровно worker_count задач asyncio.Task и возвращает их список. Если передан параметр task_group, задачи создаются через task_group.create_task, иначе через asyncio.create_task.

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

Реализуйте асинхронную функцию screen_baggage(station, scanner, worker_count) -> tuple[int, list[tuple[tuple[str, int], str]]].

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

• station — объект пункта досмотра BaggageStation;
• scanner — асинхронная функция scanner(bag), выполняющая досмотр конкретного места багажа;
• worker_count — целое число воркеров в пуле (от 1 до 16).

Каждое место багажа представлено кортежем bag = (label, weight), где label — строка метки багажа длиной от 1 до 64 символов (метки могут повторяться), а weight — неотрицательный целый вес от 0 до 1 000 000. Общее количество чемоданов — от 0 до 200.

Правила работы конвейера:

• Сформируйте фиксированный пул воркеров однократным вызовом метода station.start_workers(worker_factory, worker_count, ...). Непосредственное извлечение чемоданов из очереди и вызовы сканера должны производиться именно внутри задач этого зарегистрированного пула.
• Каждый воркер обрабатывает по одному чемодану за раз, забирая его из station.queue.
• Вызов await scanner(bag) для каждого чемодана выполняется строго при удерживаемом разрешении семафора station.slots.
• Если scanner(bag) завершился штатно, вес чемодана bag[1] прибавляется к station.total. Каждое обновление station.total обязано происходить строго под защитой замка station.lock.
• Если scanner(bag) выбросил обычное исключение (Exception), этот чемодан считается аварийным. Исключение должно быть перехвачено: в список station.failures добавляется кортеж (bag, type(error).__name__). Сбой не должен останавливать обработку остальных чемоданов очереди.
• Для каждого извлечённого из очереди чемодана метод station.queue.task_done() вызывается ровно один раз и строго после завершения его досмотра (как успешного, так и аварийного). Раннее подтверждение до завершения проверки запрещено.
• После обработки всех чемоданов из очереди все воркеры должны быть корректно остановлены и ожидаемы через await, а все удерживаемые разрешения семафора и замка — освобождены.
• Функция должна вернуть кортеж (station.total, failure_list). В failure_list должны войти все возникшие сбои (включая повторяющиеся места багажа); порядок записей в списке сбоев не регламентируется.

Ввод и вывод

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

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

async def example():
    station = make_baggage_station([("A", 4), ("B", 9), ("C", 3)], 2)
    async def scanner(bag):
        await asyncio.sleep(0)
        if bag[0] == "B":
            raise ValueError("scan")
    total, failures = await screen_baggage(station, scanner, 2)
    print(total)
    print(sorted((bag[0], name) for bag, name in failures))

asyncio.run(example())

Вывод:

7
[('B', 'ValueError')]