仅读取一行时，命名管道内容将被丢弃

Question

我可能会误解一些东西，但命名管道应该像这样吗？

# consumer:
while True:
  queue: int = os.open('pipe', flags=os.O_RDONLY | os.O_NONBLOCK)
  with os.fdopen(queue, 'rb') as stream:
    readers, _, _ = select([stream], [], [])
    if readers:
      reader = readers.pop()
      contents: bytes = reader.readline().strip()

      if b'quit' == contents:
        break

      print(contents)

# producer:
fd = os.open(pipe, os.O_WRONLY | os.O_NONBLOCK)
ps = open(fd, 'wb')
for i in range(10):
  ps.write(str(i).encode())
  ps.write(os.linesep.encode())
ps.close()

我可以看到正在写入管道的所有数据，一旦文件关闭，消费者中的选择就会将其拾取并开始读取......这是输出：

b'0'

管道的所有其余部分都被丢弃，就像它从未存在过一样。这是预期的行为吗？我的期望是打印：

b'0'
b'1'
...
b'9'

我想使用命名管道进行进程间通信。脚本 A 正在向独立脚本 B 发送命令，然后可能会发送另外三个。 B 应该把那些命令捡起来，一个接一个地执行。因此上面的代码。但是，只有第一个被执行，其余的都消失了。命令与上面的示例不同。

• cmd1+cmd2
• 10 秒后
• cmd3
• 5 秒后
• cmd4+cmd5+cmd6
• 20 秒后
• 退出

我怎样才能做到这一点？

出于某种神奇的原因，第一次选择挂起。

write(hello)
write(newline)
write(world)
flush()
print(hello) <- select hangs
... one eternity later ...
write(how)
write(newline)
write(are)
write(newline)
write(ya)
flush()
print(world) <- This should have been printed also without the new writes...
print(how) <- there were no new writes yet select didn't block as there was new data available for reading
print(are)
print(ya)

通过向 select 语句添加 1 的超时，我不需要虚拟写入来读取管道的其余部分。不确定这是否是一个选择限制，但确实看起来很可疑，尤其是从第二次开始按预期工作。

Answer 1

更新 #1

问题是每个选择之间的消费者关闭并重新打开管道。 它只消耗了一行，并且在 close 后，管道中等待读取的所有其他数据都消失了。

因此，您应该用fdopen 置换while True。此外，您应该在消费者中正确处理 EOF（这是第一个额外的 while True 的来源）。这是固定代码：

消费者

import os, sys
import select

while True:

    print("Opening pipe")
    queue: int = os.open('pipe', flags=os.O_RDONLY)
    with os.fdopen(queue, 'rb') as stream:

        while True:
            readers, _, _ = select.select([stream], [], [])

            if readers:
                reader = readers.pop()

                contents: bytes = reader.readline()
                if contents is b'':
                    print("EOF detected")
                    break

                contents: bytes = contents.strip()
                if b'quit' == contents:
                    sys.exit(0)

                print(contents)

制片人

生产者缺少 flush 调用：

import os

fd = os.open('pipe', os.O_WRONLY)
ps = open(fd, 'wb')
for i in range(10):
    ps.write(str(i).encode())
    ps.write(os.linesep.encode())
    ps.flush()

ps.close()

在 strace 下运行消费者对分析您的代码有很大帮助，它显示了所有系统调用，我可以注意到 selects 之间的 closes。

---

过时的答案

行内：

reader = readers.pop()

您从管道中获得的内容不仅仅是第一个字节。 但是你只打印出你得到的第一个字节。

看到当您在消费者代码的末尾添加另一个阅读器读取时：

contents2: bytes = reader.readline().strip()
print(contents2)

你会得到：

b'0'
b'1'

当然完整的代码应该测试它从管道中得到了多少。打印它，然后等待更多数据出现在管道中。

这是我更新的消费者代码：

import os, sys
import select

while True:
    queue: int = os.open('pipe', flags=os.O_RDONLY | os.O_NONBLOCK)
    with os.fdopen(queue, 'rb') as stream:
        readers, _, _ = select.select([stream], [], [])
        if readers:
            reader = readers.pop()
            while True:
                contents: bytes = reader.readline().strip()
                if contents is b'':
                    break
                if b'quit' == contents:
                    sys.exit(0)

                print(contents)