Как вы узнаете, будет ли sys.stdin.readline() блокироваться?
Как я могу узнать, собирается ли блокировать вызов sys.stdin.readline() (или, в более общем смысле, readline() на любом файловом объекте на основе файлового дескриптора)?
Это происходит, когда я пишу текстовую фильтрацию на основе строки в python; то есть программа повторно считывает строку текста с ввода, может быть, преобразует ее, а затем записывает на выход.
Я бы хотел реализовать разумную стратегию буферизации вывода. Мои критерии:
- Он должен быть эффективен при обработке миллионов строк в bulk--, в основном, буферизации вывода, со случайными сбросами.
- Он не должен блокироваться при вводе при буферизированном выходе.
Таким образом, небуферизованный вывод не является хорошим, потому что он нарушает (1) (слишком много записей в ОС). И вывод строки с буфером не годится, потому что он по-прежнему нарушает (1) (не имеет смысла очищать вывод в ОС на каждой из миллионов строк навалом). И вывод по умолчанию с буферизацией не является хорошим, потому что он нарушает (2) (он будет ненадлежащим образом выводить вывод, если вывод - в файл или канал).
Я думаю, что хорошим решением для большинства случаев было бы: "flush sys.stdout всякий раз, когда (его буфер заполнен или) sys.stdin.readline() собирается блокировать". Может ли это быть реализовано?
(Обратите внимание: я не утверждаю, что эта стратегия идеально подходит для всех случаев. Например, она, вероятно, не идеальна в тех случаях, когда программа привязана к процессору, в этом случае может быть разумнее очищаться чаще, чтобы избежать удержания вывода во время делая длинные вычисления.)
Для определенности, скажем, я реализую программу unix "cat -n" в python.
(На самом деле "cat -n" умнее, чем строка в то время, то есть он знает, как читать и писать часть строки до того, как прочитана полная строка, но для этого примера я в любом случае, реализовать его в любом случае.)
Линейная буферизация
(хорошо себя ведет, но нарушает критерий (1), т.е. он необоснованно медленный, поскольку он слишком сильно смывается):
#!/usr/bin/python
# cat-n.linebuffered.py
import sys
num_lines_read = 0
while True:
line = sys.stdin.readline()
if line == '': break
num_lines_read += 1
print("%d: %s" % (num_lines_read, line))
sys.stdout.flush()
Реализация по умолчанию
(быстро, но нарушает критерий (2), т.е. недружественный выход)
#!/usr/bin/python
# cat-n.defaultbuffered.py
import sys
num_lines_read = 0
while True:
line = sys.stdin.readline()
if line == '': break
num_lines_read += 1
print("%d: %s" % (num_lines_read, line))
Желаемая реализация:
#!/usr/bin/python
num_lines_read = 0
while True:
if sys_stdin_readline_is_about_to_block(): # <--- How do I implement this??
sys.stdout.flush()
line = sys.stdin.readline()
if line == '': break
num_lines_read += 1
print("%d: %s" % (num_lines_read, line))
Поэтому возникает вопрос: можно ли реализовать sys_stdin_readline_is_about_to_block()?
Мне нужен ответ, который работает как в python2, так и в python3. Я изучил каждый из следующих методов, но пока ничего не вышло.
-
Используйте select([sys.stdin],[],[],0) чтобы узнать, будет ли чтение из sys.stdin блокироваться. (Это не работает, если sys.stdin является буферизованным файловым объектом по крайней мере одной и, возможно, двум причинам: (1) он ошибочно скажет "не будет блокировать", если частичная строка готова к чтению из базового входного канала, (2) он ошибочно скажет, что "будет блокировать", если буфер sys.stdin содержит полную строку ввода, но основной канал не готов к дополнительному чтению... я думаю).
-
Неблокирующий io, используя os.fdopen(sys.stdin.fileno(), 'r') и fcntl с O_NONBLOCK (я не мог заставить это работать с readline() в любой версии python: в python2.7 он теряет ввод, когда приходит частичная строка, в python3, кажется, невозможно отличить "блокировать" и "конец ввода".)
-
asyncio (Мне непонятно, что из этого доступно в python2, и я не думаю, что он работает с sys.stdin, однако мне все равно будет интересен ответ, который работал только при чтении из канала, возвращаемого из подпроцесса.Popen()).
-
Создайте поток для выполнения цикла readline() и передайте каждую строку в основную программу через queue.Queue; то основная программа может опросить очередь перед чтением каждой строки из нее, и всякий раз, когда она увидит, что она собирается блокировать, сначала запустите stdout. (Я попробовал это и фактически получил его работу, см. Ниже, но он ужасно медленный, намного медленнее, чем буферизация строк.)
Резьбовая реализация:
Обратите внимание, что это строго не отвечает на вопрос "как определить, собирается ли sys.stdin.readline() блокировать", но в любом случае ему удается реализовать желаемую стратегию буферизации. Тем не менее, это слишком медленно.
#!/usr/bin/python
# cat-n.threaded.py
import queue
import sys
import threading
def iter_with_abouttoblock_cb(callable, sentinel, abouttoblock_cb, qsize=100):
# child will send each item through q to parent.
q = queue.Queue(qsize)
def child_fun():
for item in iter(callable, sentinel):
q.put(item)
q.put(sentinel)
child = threading.Thread(target=child_fun)
# The child thread normally runs until it sees the sentinel,
# but we mark it daemon so that it won't prevent the parent
# from exiting prematurely if it wants.
child.daemon = True
child.start()
while True:
try:
item = q.get(block=False)
except queue.Empty:
# q is empty; call abouttoblock_cb before blocking
abouttoblock_cb()
item = q.get(block=True)
if item == sentinel:
break # do *not* yield sentinel
yield item
child.join()
num_lines_read = 0
for line in iter_with_abouttoblock_cb(sys.stdin.readline,
sentinel='',
abouttoblock_cb=sys.stdout.flush):
num_lines_read += 1
sys.stdout.write("%d: %s" % (num_lines_read, line))
Проверка поведения буферизации:
Следующие команды (в bash on linux) показывают ожидаемое поведение буферизации: слишком буферизированные по умолчанию буферы, тогда как "linebuffered" и "threaded" buffer в порядке.
(Обратите внимание, что | cat в конце конвейера должен сделать буферный буфер python вместо строкового буфера по умолчанию.)
for which in defaultbuffered linebuffered threaded; do
for python in python2.7 python3.5; do
echo "$python cat-n.$which.py:"
(echo z; echo -n a; sleep 1; echo b; sleep 1; echo -n c; sleep 1; echo d; echo x; echo y; echo z; sleep 1; echo -n e; sleep 1; echo f) | $python cat-n.$which.py | cat
done
done
Выход:
python2.7 cat-n.defaultbuffered.py:
[... pauses 5 seconds here. Bad! ...]
1: z
2: ab
3: cd
4: x
5: y
6: z
7: ef
python3.5 cat-n.defaultbuffered.py:
[same]
python2.7 cat-n.linebuffered.py:
1: z
[... pauses 1 second here, as expected ...]
2: ab
[... pauses 2 seconds here, as expected ...]
3: cd
4: x
5: y
6: z
[... pauses 2 seconds here, as expected ...]
6: ef
python3.5 cat-n.linebuffered.py:
[same]
python2.7 cat-n.threaded.py:
[same]
python3.5 cat-n.threaded.py:
[same]
Тайминги:
(в bash на linux):
for which in defaultbuffered linebuffered threaded; do
for python in python2.7 python3.5; do
echo -n "$python cat-n.$which.py: "
timings=$(time (yes 01234567890123456789012345678901234567890123456789012345678901234567890123456789 | head -1000000 | $python cat-n.$which.py >| /tmp/REMOVE_ME) 2>&1)
echo $timings
done
done
/bin/rm /tmp/REMOVE_ME
Выход:
python2.7 cat-n.defaultbuffered.py: real 0m1.490s user 0m1.191s sys 0m0.386s
python3.5 cat-n.defaultbuffered.py: real 0m1.633s user 0m1.007s sys 0m0.311s
python2.7 cat-n.linebuffered.py: real 0m5.248s user 0m2.198s sys 0m2.704s
python3.5 cat-n.linebuffered.py: real 0m6.462s user 0m3.038s sys 0m3.224s
python2.7 cat-n.threaded.py: real 0m25.097s user 0m18.392s sys 0m16.483s
python3.5 cat-n.threaded.py: real 0m12.655s user 0m11.722s sys 0m1.540s
Повторяю, я бы хотел, чтобы решение, которое никогда не блокируется при сохранении буферизованного вывода (как "linebuffered", так и "threaded" хороши в этом отношении), и это также быстро: то есть сопоставимо по скорости с "defaultbuffered".
Ответы
Ответ 1
Вы, безусловно, можете использовать select: это то, за что его и его производительность хороши для небольшого количества файловых дескрипторов. Вы должны реализовать буферизацию/ломание линии, чтобы вы могли определить, будет ли более доступный вход после буферизации (что окажется) частичной линией.
Вы можете сделать всю буферизацию самостоятельно (что разумно, так как select работает на уровне файловых дескрипторов), или вы можете установить stdin для неблокирования и использовать file.read() или BufferedReader.read() (в зависимости от вашего Python), чтобы потреблять все, что доступно. Вы должны использовать неблокирующий вход независимо от буферизации, если ваш вход может быть интернет-сокетом, поскольку общие реализации select могут ложно указывать читаемые данные из сокета. (Версия Python 2 поднимает IOError с EAGAIN в этом случае, версия Python 3 не возвращает None.)
(os.fdopen здесь не помогает, поскольку он не создает новый файловый дескриптор для использования fcntl. В некоторых системах вы можете открыть /dev/stdin с помощью O_NONBLOCK.)
Реализация Python 2 на основе стандартного (буферизованного) file.read():
import sys,os,select,fcntl,errno
fcntl.fcntl(sys.stdin.fileno(),fcntl.F_SETFL,os.O_NONBLOCK)
rfs=[sys.stdin.fileno()]
xfs=rfs+[sys.stdout.fileno()]
buf=""
lnum=0
timeout=None
rd=True
while rd:
rl,_,xl=select.select(rfs,(),xfs,timeout)
if xl: raise IOError # "exception" occurred (TCP OOB data?)
if rl:
try: rd=sys.stdin.read() # read whatever we have
except IOError as e: # spurious readiness?
if e.errno!=errno.EAGAIN: raise # die on other errors
else: buf+=rd
nl0=0 # previous newline
while True:
nl=buf.find('\n',nl0)
if nl<0:
buf=buf[nl0:] # hold partial line for "processing"
break
lnum+=1
print "%d: %s"%(lnum,buf[nl0:nl])
timeout=0
nl0=nl+1
else: # no input yet
sys.stdout.flush()
timeout=None
if buf: sys.stdout.write("%d: %s"%(lnum+1,buf)) # write any partial last line
Для просто cat -n мы могли бы выписать частичные строки, как только получим их, но это позволяет им представлять обработку всей строки сразу.
На моей (не впечатляющей) машине ваш тест yes принимает "реальный 0m2.454s пользователя 0m2.144s sys 0m0.504s".
