Asyncio: две очереди с приоритетом и общим семафором

Приоритетный конвейер разгрузки

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

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

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

Что уже дано

В среде выполнения предоставлен класс PriorityContext, управляющий очередями, синхронизацией и регистрацией воркеров. В объекте контекста доступны:

• context.fast_queue и context.standard_queue — экземпляры асинхронных очередей со скоростными и стандартными объектами (всего от 0 до 1000 уникальных элементов);
• context.sem — общий асинхронный семафор с лимитом одновременных операций разгрузки от 1 до 20;
• context.lock — блокировка для безопасной записи в журнал;
• context.log — список для фиксации разгруженных объектов;
• context.pool — реестр пула воркеров с методами start(worker_coroutine) и stop();
• вспомогательные методы и флаги жизненного цикла поставщика: context.producer_finished, await context.wait_producer(), а также асинхронная операция await context.unload(item).

Вам необходимо реализовать только корутину dispatch_priority.

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

Напишите асинхронную функцию dispatch_priority(context, worker_count) -> list.

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

• context — описанный объект PriorityContext;
• worker_count — целое число от 1 до 20, задающее количество параллельных воркеров.

Функция запускает воркеры через context.pool.start(...), координирует разгрузку элементов и возвращает список всех успешно разгруженных оригинальных объектов.

Правила диспетчеризации и синхронизации:

• Воркер берёт элемент из стандартной очереди standard_queue строго тогда, когда скоростная очередь fast_queue пуста.
• Если в fast_queue ожидает элемент, следующее освобождающееся разрешение семафора context.sem должно достаться скоростной задаче. Нельзя заранее захватывать слот семафора под стандартный элемент, если есть или появляется скоростная работа.
• Уже запущенные операции разгрузки не прерываются. Дополнительное взвешивание очередей или перенаправление элементов при переполнении не требуются.
• Разгрузка выполняется через await context.unload(item) воркером, удерживающим слот семафора. Функция context.unload не выбрасывает исключений на корректных вызовах.
• Сразу после успешного вызова context.unload(item) воркер обязан добавить оригинальный объект item в context.log.append(item), удерживая context.lock.
• После записи в журнал воркер должен вызвать task_done() у той очереди, из которой был получен item.
• Входящий поставщик конечен: внешняя сторона сообщает об окончании поступления элементов через context.producer_finished (ожидается через await context.wait_producer()). Завершать самого продюсера не нужно.
• После завершения поставщика функция должна дождаться полной обработки обеих очередей через await context.fast_queue.join() и await context.standard_queue.join(), корректно остановить воркеров через await context.pool.stop() и вернуть журнал разгруженных элементов.
• При выходе из функции все занятые слоты семафора и блокировка должны быть освобождены. Порядок элементов в журнале не регламентирован, но каждый разгруженный объект должен встретиться в нём ровно один раз.

Ввод и вывод

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

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

import asyncio
async def unload(item):
    await asyncio.sleep(0)
context = PriorityContext(["fast"], ["ordinary"], unload, 1)
print(asyncio.run(dispatch_priority(context, 2)))

Вывод:

['fast', 'ordinary']