代码之家 › 专栏 › 技术社区 › Don Hatch

如何判断sys.stdin.readline()是否要阻塞?

python

Don Hatch · 技术社区 · 8 年前

如何确定对sys.stdin.readline()的调用(或者更一般地说,对任何基于文件描述符的文件对象的readline()调用)是否将被阻止?

当我用python编写一个基于行的文本过滤程序时,就会出现这种情况;

我想实现一个合理的输出缓冲策略。

当处理数百万个
当保持缓冲输出时,它不应该阻塞输入。

因此,无缓冲输出是不好的,因为它违反了(1)(对操作系统的写入太多)。 (在一百万行中的每一行批量刷新输出到操作系统是没有意义的)。默认的缓冲输出是不好的,因为它违反了(2)(如果输出是到文件或管道,它将不适当地保留输出)。

“每当(其缓冲区已满或)sys.stdin.readline()即将阻塞时,刷新sys.stdout。”。这能实现吗?

(请注意,我并不认为这种策略是完美的案例。例如, 更频繁地刷新,以避免在执行长时间计算时保留输出。)

(实际上,“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)如果部分行准备好从底层输入管道读取,它会错误地说“will not block”,(2)如果sys.stdin的缓冲区包含一个完整的输入行,但是底层管道没有准备好进行额外的读取,那么它会错误地说“will block”。。。我想是的)。
os.fdopen(sys.stdin.fileno(), 'r') 和 fcntl 具有 O_NONBLOCK (我无法在任何python版本中使用readline(): 在python2.7中,每当部分行进入时,它就会丢失输入; 在python3中,似乎无法区分“will block”
asyncio
创建一个线程来执行 readline() 循环并将每条线传递到主管道程序通过一个队列。队列;然后主程序可以在从中读取每一行,当它看到它要阻塞时,首先刷新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))

下面的命令(在linux上的bash中)显示了预期的缓冲行为:“defaultbuffered”缓冲区过于激进,而“linebuffered”和“threaded”缓冲区正好相反。

| 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]

时间安排:

(在linux上的bash中):

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 回复 | 直到 8 年前

Davis Herring 8 年前

你当然可以用 select :这就是它的用途,对于少量的文件描述符,它的性能很好。您必须自己实现行缓冲/中断,这样才能检测缓冲(结果是)部分行之后是否有更多的输入可用。

你能做到的缓冲你自己(这是合理的,因为 选择 在文件描述符级别操作),也可以设置 stdin 无阻塞使用 file.read() 或 BufferedReader.read() (取决于您的Python版本)来使用任何可用的东西。你呢必须 选择 IOError 具有 EAGAIN 在这种情况下;Python3版本返回 None .)

( os.fdopen fcntl 使用。在某些系统上,你可以打开 /dev/stdin 具有 O_NONBLOCK .)

基于默认值(缓冲)的Python2实现 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 测试采用“real 0m2.454s user 0m2.144s sys 0m0.504s”。

minitchoup 6 年前

# -*- coding: utf-8 -*-
import os
import sys
import select
import fcntl
import threading


class StdInput:
    def __init__(self):
        self.close_evt = threading.Event()

        fcntl.fcntl(sys.stdin.fileno(), fcntl.F_SETFL, fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK);
        self.input = (sys.stdin.original_stdin if hasattr(sys.stdin, "original_stdin") else sys.stdin)
        self.epoll = select.epoll()
        self.epoll.register(sys.stdin.fileno(), select.EPOLLIN | select.EPOLLPRI | select.EPOLLERR | select.EPOLLHUP | select.EPOLLRDBAND)

    def read(self):
        while not self.close_evt.is_set():
            input_line = self.input.readline()
            # If the object is in non-blocking mode and no bytes are available, None is returned.
            if input_line is not None and len(input_line) > 0:
                break           
            print("Nothing yet...")
            evt_lst = self.epoll.poll(1.0)  # Timeout 1s
            print("Poll exited: event list size={}".format(len(evt_lst)))
            if len(evt_lst) > 0:
                assert len(evt_lst) == 1
                if (evt_lst[0][1] & (select.EPOLLERR | select.EPOLLHUP)) > 0:
                    raise Exception("Ooops!!!")
        return input_line


if __name__ == "__main__":
    i = StdInput()

    def alm_handle():
        i.close_evt.set()
    threading.Timer(4, alm_handle).start()

    print("Reading...")
    input_line = i.read()
    print("Read='{}'".format(input_line))