Две ветки итератора Python: почему tee может удерживать весь поток

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

Отставание требует хранения

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

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

Размер примера выбран маленьким для обозримости. В реальном экспорте вместо числа может быть словарь с большим текстом. Буферизация удерживает ссылки на такие объекты, поэтому разница в миллионах элементов становится заметной даже без явного list в коде приложения. Особенно опасно запускать ветки последовательно и считать, что tee автоматически сделал обе потоковыми.

Увидеть чтения общего источника

from itertools import tee

reads = []

def source():
    for value in range(6):
        reads.append(value)
        yield value

fast, slow = tee(source())
total = sum(fast)
assert total == 15
assert reads == list(range(6))
before = len(reads)
restored = list(slow)
assert restored == list(range(6))
assert len(reads) == before
print("source reads:", len(reads))
print("slow replay:", restored)

paired = []
left, right = tee(range(4))
for a, b in zip(left, right):
    paired.append((a, b))
assert paired == [(0, 0), (1, 1), (2, 2), (3, 3)]
print("paired:", paired)

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

Управлять скоростью потребителей

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

Парный обход в конце примера показывает благоприятный сценарий: ветки идут почти рядом. Он объясняет, почему небольшое отставание может быть приемлемым. Однако переносить этот вывод на последовательные sum(fast), list(slow) нельзя. При ревью смотрите на расстояние между потребителями во времени и по числу элементов, а не только на присутствие слова iterator.

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

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

Продолжить практику

Продолжите с ленивыми вычислениями в курсе «Python для профессионалов»: программа включает итераторы, генераторы и работу с памятью. Откройте программу и сопоставьте время получения значений с временем освобождения ссылок — эта связь помогает выбирать потоковую архитектуру без скрытого накопления.