python - 防止从空 FIFO 中读取数据阻塞

Question

在 Python 3 Web 应用程序中，我需要 shell out 到一个处理图像的命令行实用程序，将其输出写入命名管道 (fifo)，然后将该输出(管道的内容)解析为PIL/枕头图片。这是基本流程(工作代码很长而且没有错误!):

from os import mkfifo
from os import unlink
from PIL import Image
from subprocess import DEVNULL
from subprocess import PIPE
from subprocess import Popen

fifo_path = '/tmp/myfifo.bmp'
cmd = '/usr/bin/convert -resize 100 /path/to/some.tif ' + fifo_path
# make a named pipe
mkfifo(fifo_path)
# execute
proc = Popen(cmd, stdout=DEVNULL, stderr=PIPE, shell=True)
# parse the image
pillow_image = Image.open(fifo_path)
# finish the process:
proc_exit = proc.wait()
# remove the pipe:
unlink(fifo_path)
# just for proof:
pillow_image.show()

(在上面的示例中，我已经用 ImageMagick 替换了我实际必须使用的实用程序，只是因为您不太可能拥有它——它根本不会影响问题。)

这在大多数情况下都很好用，我可以处理大多数异常(为清楚起见，上面省略了)，但有一种情况我无法弄清楚如何处理，即如果出现问题该怎么办shellout，导致空管，例如如果图像不存在或由于某种原因损坏，例如:

fifo_path = '/tmp/myfifo.bmp'
cmd = '/usr/bin/convert -resize 100 /path/to/some/bad_or_missing.tif ' + fifo_path
# make a named pipe
mkfifo(fifo_path)
# execute
proc = Popen(cmd, stdout=DEVNULL, stderr=PIPE, shell=True)
# parse the image
pillow_image = Image.open(fifo_path) # STUCK
...

应用程序只是卡在这里，因为我无法到达 proc_exit = proc.wait() 我无法设置 timeout(例如 proc_exit = proc.wait(timeout=2))，这是我通常会做的。

我试过将整个业务包装在上下文管理器中，类似于 this answer ，但该配方不是线程安全的，这是一个问题，当我加入线程或进程(不是我的强项，但像这样):

from multiprocessing import Process
from os import mkfifo
from os import unlink
from PIL import Image
from subprocess import DEVNULL
from subprocess import PIPE
from subprocess import Popen

def do_it(cmd, fifo_path):
    mkfifo(fifo_path)
    # I hear you like subprocesses with your subprocesses...
    sub_proc = Popen(cmd, stdout=DEVNULL, stderr=PIPE, shell=True)
    pillow_image = Image.open(fifo_path)
    proc_exit = sub_proc.wait()
    unlink(fifo_path)

fifo_path = '/tmp/myfifo.bmp'
cmd = '/usr/bin/convert -resize 100 /path/to/some/bad_or_missing.tif ' + fifo_path
proc = Process(target=do_it, args=(cmd, fifo_path))
proc.daemon = True
proc.start()
proc.join(timeout=3) # I can set a timeout here
# Seems heavy anyway, and how do I get pillow_image back for further work?
pillow_image.show()

希望这些能说明我的问题和我尝试过的方法。提前致谢。

Answer 1

POSIX read(2) :

When attempting to read from an empty pipe or FIFO:

If no process has the pipe open for writing, read() shall return 0 to indicate end-of-file.

Image.open(fifo_path) 可能卡住当且仅当命令终止而在被阻止时打开 fifo_path 进行写入。

Normally, opening the FIFO blocks until the other end is opened also.

这是一个正常的序列:

1. cmd 在尝试打开 fifo_open 进行写入时阻塞
2. 您的 Python 代码块在尝试打开以供阅读时出现
3. 一旦 FIFO 被两个进程打开，数据流就开始了。除了名称外，FIFO 类似于管道——只有一个管道对象——内核在内部传递所有数据而不将其写入文件系统。 The pipe is not a seekable file and therefore Image.open() may read until EOF

4. cmd 关闭它的管道末端。您的代码收到 EOF，因为没有其他进程打开 FIFO 进行写入并且 Image.open(fifo_path) 返回。

   cmd 的管道末端由于成功完成或错误而关闭的原因并不重要，cmd 是突然终止还是不是:只要它的末端是封闭的。

   您的进程是否调用proc.wait() 并不重要。 proc.wait() 不会终止 cmd。 proc.wait() 不会阻止管道的另一端打开或关闭。 proc.wait() 唯一要做的就是等待子进程死亡和/或返回已死亡子进程的退出状态。

这是死锁案例:

1. 在调用 Image.open() 时，cmd 甚至不会出于任何原因尝试打开 fifo_open 进行写入，例如，没有/usr/bin/convert，错误的命令行参数，错误/无输入等
2. 您的 Python 代码块在尝试打开以供阅读时出现

fifo_open 未打开以进行写入，因此 Image.open(fifo_open) 永远卡在尝试打开它以进行读取.

---

您可以在后台线程中打开 FIFO 进行写入，并在父进程打开 FIFO 进行读取时关闭它:

#!/usr/bin/env python3
import contextlib
import os
import subprocess
import sys
import textwrap
import threading

fifo_path = "fifo"
with contextlib.ExitStack() as stack:
    os.mkfifo(fifo_path)
    stack.callback(os.remove, fifo_path)
    child = stack.enter_context(
        subprocess.Popen([
            sys.executable, '-c', textwrap.dedent('''
            import random
            import sys
            import time
            if random.random() < 0.5: # 50%
                open(sys.argv[1], 'w').write("ok")
            else:
                sys.exit("fifo is not opened for writing in the child")
            '''), fifo_path
        ]))
    stack.callback(child.kill)
    opened = threading.Event()  # set when the FIFO is opened for reading
    threading.Thread(target=open_for_writing, args=[fifo_path, opened, child],
                     daemon=True).start()
    pipe = stack.enter_context(open(fifo_path))  # open for reading
    opened.set()  # the background thread may close its end of the pipe now
    print(pipe.read()) # read data from the child or return in 3 seconds
sys.exit(child.returncode)

在 EOF 时， child 被杀死。

在 open_for_writing() 打开 FIFO 的地方，解锁 open(fifo_path) 进而可以关闭它。为避免 pipe.read() 返回得太快，它给 child 3 秒时间打开 FIFO 进行写入:

def open_for_writing(path, opened, child):
    with open(path, 'w'):
        opened.wait()  # don't close until opened for reading in the main thread
        try:
            child.wait(timeout=3)  # the child has 3 seconds to open for writing
        except subprocess.TimeoutExpired:
            pass

如果您确定子进程要么尝试打开 FIFO 要么最终退出(或者您可以在子进程运行时挂起 Python 进程，那么您可以放弃超时并使用 child.wait( ) 而不是 child.wait(timeout=3)。有了这个更改，就没有任意超时了，代码可以在任意慢的系统上运行(无论出于何种原因)。

该代码演示了为什么应该尽可能避免使用线程，或者为什么应该更喜欢已建立的模式(不太通用但可以保证正常工作)，例如通过通信进行同步。

答案中的代码应该适用于各种情况，但各个部分错综复杂。在非常具体的案例出现之前，即使是很小的变化也可能不会产生明显的影响。