Asyncio Queue: резервная очередь при перегрузке основного конвейера

Диспетчер перегрузки телеметрии

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

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

Реализуйте асинхронный диспетчер перегрузки и определите итоговые значения раздельных счётчиков успешно отправленных кадров для основного и резервного контуров.

Что уже дано

В среде выполнения доступен класс TelemetryContext со следующими атрибутами и методами:

• context.receive() — асинхронный метод получения очередного кадра; возвращает объект кадра или маркер завершения TelemetryContext.END;
• context.send(frame, route) — асинхронный метод отправки кадра, где route принимает значение 'primary' или 'reserve'; метод не выбрасывает ошибок;
• context.main_queue — ограниченная очередь asyncio.Queue с вместимостью от 1 до 100 элементов;
• context.reserve_queue — неограниченная очередь asyncio.Queue;
• context.sem — общий семафор asyncio.Semaphore, ограничивающий суммарное число одновременных вызовов send (значение от 1 до 20);
• context.lock — общий замок asyncio.Lock для синхронизации записи статистики;
• context.stats.record(route) — метод фиксации успешной отправки кадра для маршрута 'primary' или 'reserve';
• context.pool.start(coroutine, lane) — запускает воркер в пуле задач и регистрирует за ним контур 'primary' или 'reserve';
• context.pool.stop() — останавливает зарегистрированные задачи пула.

Вам предстоит реализовать асинхронную функцию route_overflow.

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

Напишите асинхронную функцию route_overflow(context, main_workers, reserve_workers).

Параметры main_workers и reserve_workers задают количество воркеров для основного и резервного контуров соответственно (целые положительные числа от 1 до 20). Общее число входящих кадров составляет от 0 до 1000.

Функция должна вернуть объект, совместимый с collections.abc.Mapping (например, словарь), со строковыми ключами 'primary' и 'reserve' и целочисленными количествами успешно отправленных кадров по каждому контуру.

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

• Запустите через context.pool.start(..., lane='primary') ровно main_workers воркеров, читающих context.main_queue, и через context.pool.start(..., lane='reserve') ровно reserve_workers воркеров, читающих context.reserve_queue.
• Непрерывно считывайте кадры через await context.receive(). Для каждого кадра попытайтесь немедленно положить его в основную очередь через context.main_queue.put_nowait(frame).
• Если основная очередь заполнена (возникает asyncio.QueueFull), сразу поместите этот же кадр в context.reserve_queue. Не ждите освобождения места в основной очереди, не сбрасывайте кадры и не заменяйте их.
• Воркеры каждого контура извлекают кадры из своей очереди. Для каждого извлечённого кадра воркер выполняет await context.send(frame, route), удерживая общий семафор context.sem.
• После успешной отправки воркер захватывает общий замок context.lock и вызывает context.stats.record(route).
• Для каждого извлечённого из очереди кадра должен быть гарантированно вызван queue.task_done() соответствующей очереди.
• Когда метод receive() возвращает TelemetryContext.END, новые кадры больше не поступают. Диспетчер должен дождаться полной обработки всех кадров через await context.main_queue.join() и await context.reserve_queue.join(), после чего вызвать await context.pool.stop().
• Итоговые значения в возвращаемом словаре должны совпадать со значениями в context.stats.

Ввод и вывод

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

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

import asyncio
items = iter(range(1))
async def receive():
    return next(items, TelemetryContext.END)
async def send(frame, route):
    await asyncio.sleep(0)
context = TelemetryContext(receive, send, 2, 2)
print(asyncio.run(route_overflow(context, 2, 1)))

Вывод:

{'primary': 1, 'reserve': 0}