Python：检查命名管道是否有数据答案

【问题标题】：Python: Check if named pipe has dataPython：检查命名管道是否有数据
【发布时间】：2016-09-23 14:44:03
【问题描述】：

我的 Unix 系统上有一个 Python3 进程一直在运行，我希望能够通过命名管道从仅偶尔运行的其他进程随机向它发送数据。如果命名管道没有数据，我希望我的进程继续做其他事情，所以我需要检查它是否有数据没有阻塞。

我无法弄清楚如何在不打开它的情况下进行检查，但除非我设置了非阻塞标志，否则会打开块。如果我设置了标志，如果我碰巧在读取之前或期间写入管道，它会崩溃。

这是我能做到的最好的：

import os

fifo = "pipe_test.fifo"
done = False
fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)
while not done:
    try:
        s = os.read(fd, 1024) # buffer size may need tweaking
        print(s)
        done = True
    except BlockingIOError as e:
        pass
os.close(fd)

如果管道中没有数据，我会得到b""，然后它会退出。如果管道中有数据，它会得到一次异常，重试，然后获取数据。好像我做错了什么，可能会遇到奇怪的比赛条件。有更好的方法吗？

【问题讨论】：

你能用第二个线程等待阻塞管道吗？
@Jasper 我可以做到。这也不理想，但我承认它更好。此外，我可以使用 Python 的文件库而不是低级的 os 东西，这样我就可以阅读到换行符等。谢谢，除非有办法在没有其他线程的情况下进行检查，否则我会将我的代码更改为那个。

标签： python named-pipes

【解决方案1】：

如果您可以更改客户端的代码，我不会使用命名管道，而是使用UNIX domain sockets，因为它们支持数据报：

import errno, fcntl, os, socket

服务器：

# bind socket
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
sock.bind('pipe_test.fifo')
# set socket non-blocking
fcntl.fcntl(sock.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)

# get a datagram
try:
    datagram = sock.recv(1024)
except (OSError, socket.error) as ex:
    if ex.errno not in (errno.EINTR, errno.EAGAIN):
        raise
else:
    print('Datagram: %r' % datagram)

客户：

sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
sock.sendto('Hello!', 'pipe_test.fifo')

但您可能想查看multithreading 而不是使用非阻塞套接字。

【讨论】：

我不知道数据报。这很有用。但看起来你像我一样抛出和捕捉错误。那安全吗？
@sudo，是的。如果没有等待接收的数据报，则抛出 EAGAIN。如果信号中断了接收，则抛出 EINTR。对于非阻塞套接字来说，后一种情况应该不是不可能的，但是为了安全也没有什么坏处。可能的错误是：读取缓冲区太少（但您知道您的数据），并且客户端无法发送数据，因为您读取速度太慢。一个socket的缓冲区默认是200kB或者512个包，stackoverflow.com/a/22007520/416224。

【解决方案2】：

这不是真正的答案，但如果它对任何人都有用，这就是我在另一个线程中的做法。

class QueryThread(threading.Thread):

    def __init__(self, args=(), kwargs=None):
        threading.Thread.__init__(self, args=(), kwargs=None)
        self.daemon = True
        self.buf = []
        if not general.f_exists("pipe"):
            os.mkfifo("pipe")

    def run(self):
        f = open("pipe")
        while True:
            try:
                query = next(f).replace("\n", "")
                if query != "":
                    self.buf.append(query)
                    print("Read in new query from pipe: {}, buf = {}".format(query, self.buf))
            except StopIteration: # not a pipe error, just means no data is left, so time to re-open
                f.close()
                f = open("pipe")
        f.close()

    def get_query(self):
        if len(self.buf) == 0: return ""
        query = self.buf[0]
        self.buf.__delitem__(0)
        return query

它将换行符分隔的消息保存在缓冲区中。您可以从另一个线程调用get_query 方法并获取最后收到的消息。

【讨论】：