Генераторы Python: разделите поток между двумя потребителями

Синхронный диспетчер пакетов

Синхронный диспетчер принимает поток пакетов из внешнего источника и делит его между двумя потребителями с помощью парных генераторов: регулярная ветка забирает пакеты с меткой INFO, а приоритетная — с меткой CRIT.

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

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

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

Что уже дано

В среде уже объявлены вспомогательные функции make_packet_source и observe_packet_source для создания и отслеживания состояния источника, а вам нужно написать только саму функцию split_packets.

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

Напишите функцию split_packets(source).

Аргумент source — это однократный итератор, отдающий кортежи вида (tag, payload), где tag принимает значение 'INFO' или 'CRIT', а payload — целое число от -1 000 000 до 1 000 000. Всего в источнике содержится от 0 до 128 пакетов. Объект source имеет метод close(), который освобождает внешний ресурс и не выбрасывает исключений.

Функция должна возвращать кортеж из двух закрываемых однократных итераторов: (info_branch, crit_branch).

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

• При вызове split_packets(source) из источника не должно производиться ни одного чтения. Чтение выполняется строго по одному пакету и только по требованию активной ветки, у которой опустела собственная очередь.
• Пакеты, предназначенные для другой активной ветки, помещаются в её очередь с сохранением исходного порядка (FIFO).
• Закрытие любой ветки (вызовом метода close(), в том числе до первого обращения через next()) навсегда отключает потребителя. Пакеты с меткой закрытой ветки сразу безвозвратно отбрасываются.
• Источник освобождается вызовом source.close() ровно один раз, когда закрыты обе ветки либо когда входной поток source полностью исчерпан (StopIteration) и все накопленные в очередях пакеты выданы или сброшены.
• Если входной поток достиг конца, но у активной ветки ещё остались пакеты в очереди, source не закрывается преждевременно, пока эта очередь не будет прочитана или ветка не закроется.
• Повторные вызовы next() или close() на уже закрытой или исчерпанной ветке не должны повторно опрашивать или повторно закрывать source.

Ввод и вывод

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

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

source = make_packet_source([('INFO', 1), ('CRIT', 2), ('INFO', 3)])
info, crit = split_packets(source)
print(next(info))
info.close()
print(list(crit))
print(observe_packet_source(source)['close_calls'])

Вывод:

('INFO', 1)
[('CRIT', 2)]
1