实时从管道获取基于行的输出

Question

我想在（接近）实时读取tcpdump子进程的逐行输出，但我需要选项来评估管道是否为空（因此是队列）。线程等待0.5秒，得到所有排队的输出行，处理它（例如，超过0.5秒的平均分组分配）并返回一些东西。

最小的非工作示例：

millis = lambda: int(round(time.time() * 1000))
def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        print(millis())
        print(line)
        queue.put(line)
    out.close()

def infiniteloop1():
    p = Popen( [ 'sudo', 'tcpdump', '-i', 'wlan0', '-nn', '-s0', '-q', '-l', '-p', '-S' ], stdout=subprocess.PIPE, stderr=STDOUT)
    q = Queue()
    t = Thread(target=enqueue_output, args=(p.stdout, q))
    t.daemon = True # thread dies with the program
    t.start()

    while True:
        while True:
            # read line without blocking
            try: 
                row = q.get_nowait() # or q.get(timeout=.1)
            except Empty:
                print('empty')
                break
            else:
                pass
        time.sleep(0.5)
thread1 = threading.Thread(target=infiniteloop1)
thread1.daemon = True
thread1.start()

捕获连续的包流时的输出：

[...]
1552905183422
10:33:03.334167 IP 192.168.1.2.36189 > a.b.c.d.443: tcp 437
1552905183422
10:33:03.357215 IP a.b.c.d.443 > 192.168.1.2.36189: tcp 0
1552905183423
10:33:03.385145 IP 192.168.1.2.36189 > a.b.c.d.443: tcp 437
empty
empty
1552905184438
10:33:03.408408 IP a.b.c.d.443 > 192.168.1.2.36189: tcp 0
1552905184439
10:33:03.428045 IP 192.168.1.2.36189 > a.b.c.d.443: tcp 437
1552905184439
10:33:03.451235 IP a.b.c.d.443 > 192.168.1.2.36189: tcp 0
[...]

注意两个连续的“空”。第一个“空”之前的最后一个数据包由tcpdump以10：33：03.385145捕获，并在1552905183423处传送到队列，耗时38毫秒。在两个“空”之间没有数据包被传递到队列中。第二个“空”之后的第一个包在10：33：03.408408被捕获并且被传送1552905184438，它在前一个数据包之后被传送1秒但是在“空”之间被捕获。为什么不在“空”之间交付？这种情况并不罕见，但每次弹出队列都会导致没有交付包，为什么会这样？

Answer 1

第二个“空”之后的第一个包在10：33：03.408408被捕获并且被传送1552905184438，它在前一个数据包之后被传送1秒但是在“空”之间被捕获。

给定您的代码，只有在for line in iter(out.readline, b'')中的迭代器返回一个新项时才会计算和打印系统时间戳，因此这就是延迟似乎来自的地方。

我怀疑stdio缓冲是罪魁祸首。在Linux（即libc / glibc）上，如果STDOUT描述符引用TTY，则启用行缓冲。如果它引用其他东西（例如管道），则STDOUT描述符被完全缓冲;在调用write系统调用之前，您的进程需要填充4096个字节（在Linux上是默认的）。 非常粗略地计算，根据您在此处显示的输出，您的子进程似乎每~0.025秒生成~65个字节。给定4kB缓冲区，填充它并触发刷新/写入需要大约1.625秒。

从subprocess.PIPE读取并将输出发送到主进程'stdout需要少得多，因此你会看到tcpdump输出的突发，即在几微秒内打印（从stdout迭代器接收）约25ms，以及你的程序随后等到下一个4kB被刷新。

如果您有可能安装第三方软件包（并使用Python> = 2.7），您可能需要查看pexpect。该包的子节点连接到PTY，使系统将它们视为交互式程序，因此它们的stdout描述符是行缓冲的。