Asyncio Queue: вытеснение старейшей заявки без блокировки поставщика

Вытесняющий буфер котировок

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

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

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

Что уже дано

В среде исполнения уже доступны вспомогательные инструменты make_quote_probe и observe_quote_buffer, а также объект терминала terminal с асинхронным методом apply(quote). Вам предстоит реализовать класс DropOldestQuoteBuffer с методами __init__, publish и consume_one.

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

Реализуйте класс DropOldestQuoteBuffer:

• __init__(self, queue, lock, terminal) — сохраняет переданные разделяемые ресурсы:
• queue — разделяемый экземпляр asyncio.Queue с фиксированной вместимостью maxsize (от 1 до 4);
• lock — разделяемый экземпляр asyncio.Lock для защиты терминала;
• terminal — объект торгового терминала, метод apply(quote) которого обновляет отображение цен.
• publish(self, quote) -> None — синхронный неблокирующий метод добавления котировки:
• Котировка представляет собой словарь (или подкласс словаря) с ключами "symbol" (ASCII-строка длиной от 1 до 20 символов) и "price" (целое число от 0 до 1 000 000). Допускаются повторные значения и одинаковые объекты котировок.
• Если очередь заполнена до предела, метод обязан сначала немедленно извлечь самый старый ожидающий в ней элемент (FIFO) и сразу подтвердить его вызовом queue.task_done(), после чего поместить новую котировку в освободившееся место.
• Метод возвращает None, не ожидает консьюмеров и не создает отложенных задач на добавление.
• Запрещено вытеснять или подтверждать элементы, которые уже были извлечены консьюмерами и находятся в обработке, а также запрещено объединять котировки с одинаковым символом.
• async consume_one(self) -> None — асинхронный метод консьюмера:
• Дожидается извлечения одной котировки из очереди.
• Вызывает await terminal.apply(quote) строго под защитой предоставленного lock (например, через async with lock:).
• Ровно один раз подтверждает завершение обработки этой котировки вызовом queue.task_done() и возвращает None.

Учёт задач в очереди должен корректно удерживать queue.join() в состоянии ожидания, пока в очереди есть необработанные или находящиеся в обработке элементы, и позволять join() завершиться, когда все оставшиеся котировки обработаны.

Ввод и вывод

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

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

probe = make_quote_probe({
    'capacity': 1,
    'quotes': [['A', 1], ['A', 2], ['B', 3]],
    'offers': [0, 1, 2],
})
buffer = DropOldestQuoteBuffer(probe.queue, probe.lock, probe.terminal)
observation = observe_quote_buffer(probe, buffer)
print(observation['applied'])
print(observation['prices'])
print(observation['unfinished_at_end'])

Вывод:

[2]
{'B': 3}
0