Как корректно читать строки из канала в Python: EOF и readline

Разбираем, как читать строки из канала в Python и правильно завершать цикл: поведение readline при EOF, почему не нужен select, надёжный шаблон чтения.

Чтение строк из канала и определение момента, когда записывающая сторона завершила работу, кажется простой задачей, но её легко переусложнить. Распространённая ловушка — совмещать низкоуровневые проверки готовности с высокоуровневым файловым вводом-выводом и в итоге упускать чистый сигнал окончания потока. Хорошая новость: файловый интерфейс уже даёт ответ — и всё проще, чем выглядит.

Постановка задачи

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

from datetime import datetime
from itertools import batched
import os
from select import select
from threading import Thread
from time import sleep

# Каждая надстрочная цифра кодируется в 3 байта.
sample_txt = '⁰¹²\n³\n⁴\n⁵⁶\n⁷⁸⁹⁰¹²\n³'
sample_buf = bytes(sample_txt, 'utf8')

r_fd, w_fd = os.pipe()

bin_out = open(w_fd, 'wb', buffering=0)
txt_in = open(r_fd, 'r')

collected = []

def pump_reader():
    t_prev = datetime.now()
    while True:
        sleep(1)
        t_now = datetime.now()
        print('A', (t_now - t_prev).total_seconds())
        t_prev = t_now

        r_ready, w_ready, e_ready = select([txt_in], [txt_in], [txt_in], 0)
        if txt_in.closed:
            break
        if e_ready:
            break
        if not r_ready:
            continue

        piece = txt_in.readline()
        print('B', (t_now - t_prev).total_seconds())
        t_prev = t_now

        if piece:
            print('got chunk', repr(piece))
            collected.append(piece)

worker = Thread(target=pump_reader)
worker.start()

for segment in batched(sample_buf, 4):
    payload = bytes(segment)
    sleep(1.6)
    bin_out.write(payload)

bin_out.close()

worker.join()

print(repr(collected))

Наблюдаемое поведение: все ожидаемые строки приходят, включая последнюю без завершающего переноса строки, но цикл чтения не выходит, и поток остаётся жив.

Что происходит на самом деле

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

Чёткий сигнал уже есть на уровне файлового интерфейса. В этом сценарии нет нужды комбинировать select с проверками closed или списками исключений. Главное — трактовать пустую строку от readline как конец потока.

Решение

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

def pump_reader():
    while line := txt_in.readline():
        print('got chunk', repr(line))
        collected.append(line)

Такой вариант делает семантику EOF явной и надёжной. Равнозначный подход — итерироваться по самому файловому объекту: он будет выдавать строки до конца потока.

Почему это важно

При работе с каналами подпроцессов или любыми конвейерами «производитель–потребитель» корректный путь завершения так же важен, как и производительность в штатном режиме. Опора на поведение EOF в файловом API упрощает код и предотвращает скрытые зависания, когда потоки ждут бесконечно, а join блокируется до принудительного прерывания. Если вы уже читаете текст через readline, позвольте ему сообщить об окончании потока.

Выводы

Не спорьте с файловым интерфейсом. Если читаете через readline, воспринимайте пустую строку как конец — и цикл завершится без вспомогательных проверок или таймаутов. Это снижает сложность, убирает крайние случаи с неполными строками и делает жизненный цикл потоков предсказуемым.