在 Popen.stdin 上轮询/等待 HUP

Question

我正在用 Python 为 GNU dd(1) 编写基于 Zenity 的状态监视器。 由于我所针对的系统的限制，包装器必须在 Python 2 上运行，并且不能拉入外部库。

要求 Zenity 的“取消”按钮在 dd 尚未完成时终止它。

我必须立即进行以下操作（即触发/驱动/立即）；如果满足/触发以下多个条件同时，它们将按列出的顺序执行：

• Zenity 退出时，终止 dd
• 当 dd 写入它的stderr 时，munge+forward 那个数据到 Zenity 的stdin
• dd退出时，如果返回码不为0，则终止Zenity

然而，epoll 对象似乎只在 dd 的输出上触发；尽管我在 Zenity 的 stdin 上注册了 EPOLLHUP，但它永远不会在 Zenity 退出时触发。

应该/如何做到这一点？我知道 epoll 是唯一可以用来正确触发dd 输出的原语（通过EPOLLIN）；我也明白这是一个笨拙的原语，可能不适合在 Zenity 退出时触发。 （如果需要，我可以在这个文件中实现更多逻辑；这样做比引入任何 3rd 方库，无论多么小或“常见”都是无限可取的。我重申我理解 epoll 难以使用，可能需要大量的胶合逻辑。）

或者：如果 epoll 不是监视subprocess 退出的正确原语，那么在监视子进程的输出的同时监视子进程退出的正确方法是什么以兼容 Python 2 的方式?

（我本身并不需要多线程功能；按顺序执行所有操作将完全符合规范；但是，如果在这种情况下绝对需要多线程编程以避免繁忙-循环，那就这样吧。）

---

以下是我目前的完整代码。

#!/usr/bin/env python
from __future__ import division
import sys,os,stat,fcntl,select,subprocess,re

def main(args=sys.argv[1:]):
    fname = parseifname(args)
    n = sizeof(fname)

    dcmd = ['dd'] + args + ['status=progress']
    zcmd = ['zenity', '--progress', '--time-remaining']

    #Launch dd
    dd = subprocess.Popen(dcmd, stderr=subprocess.PIPE)
    set_nonblocking(dd.stderr)

    #Launch Zenity
    zenity = subprocess.Popen(zcmd, stdin=subprocess.PIPE)
    set_direct(zenity.stdin)#TODO: why doesn't this line work?*

    #set title/status
    zenity.stdin.write(('#%s\n' % ' '.join(dcmd)).encode())
    zenity.stdin.flush()#*i.e. instances of this line shouldn't be necessary...

    #We want to trigger on all of the following:
    toPoll = [
        (dd.stderr, select.EPOLLIN       #dd status update
                  | select.EPOLLHUP),    #dd exit
        (zenity.stdin, select.EPOLLHUP), #Zenity exit
    ]

    calcPercent = genCalcPercent(n)

    with ePoll(toPoll) as E:
        rBytes = re.compile(r'\r(\d+) bytes'.encode())
        while dd.poll() is None:
            evs = E.poll()#TODO: I'm not sure if this is blocking, or if I've induced a busy loop...
            for fn,ev in evs:
                if fn == dd.stderr.fileno():
                    if (ev & select.EPOLLIN):
                        #dd sent some output
                        line = dd.stderr.read()
                        m = rBytes.match(line)
                        #sys.stderr.buffer.write(line)
                        if m:
                            x = int(m.groups()[0])
                            zenity.stdin.write(('%f\n' % calcPercent(x)).encode())
                            zenity.stdin.flush()
                    if (ev & select.EPOLLHUP):
                        #dd exited
                        pass#The containing loop will handle this; don't need to take action
                if fn == zenity.stdin.fileno():
                    if (ev & select.EPOLLHUP):#TODO: WHY DOESN'T THIS ACTIVATE??
                        #Zenity exited
                        dd.terminate()
        if dd.returncode == 0:
            #dd exited successfully
            zenity.stdin.write('100\n'.encode())
            zenity.stdin.flush()
        else:
            zenity.terminate()

# Functions below here #

def parseifname(argv=sys.argv[:1], default='/dev/stdin'):
    '''Given dd's argument list, attempts to return the name of that file which dd would use as its input file'''
    M = re.compile(r'^if=(.*)$')
    ifname = default
    for x in argv:
        m = M.match(x)
        if m:
            ifname = m.groups()[0]
    return ifname

def sizeof(fname):
    '''Attempts to find the length, in bytes, of the given file or block device'''
    s = os.stat(fname)
    m = s.st_mode
    try:
        if stat.S_ISREG(m):
            #Regular File
            n = s.st_size
        elif stat.S_ISBLK(m):
            #Block Device
            n = int(subprocess.check_output(['lsblk', '-b', '-n', '-l', '-o', 'SIZE', '-d', fname]))
        else:
            raise ValueError("file is neither a standard nor block file")
    except:
        #Unidentifiable
        n = None
    return n

def genCalcPercent(n):
    '''Given n, returns a function which, given x, returns either x as a percentage of n, or some sane stand-in for such'''
    if n:
        #Input file size was identified
        return lambda x: 100 * x / n
    else:
        #Input file size was unidentifiable, zero, or otherwise falsy
        #we'll at least  try to visually show progress
        return lambda x: 99.99999 * (1 - 0.5 ** (x / 2**32))

def set_nonblocking(fd=sys.stdin):
    '''Appends os.O_NONBLOCK to the given file descriptor's flags.'''
    return fcntl.fcntl(
     fd,
     fcntl.F_SETFL,
     fcntl.fcntl(fd,fcntl.F_GETFL)
      | os.O_NONBLOCK
    )

def set_direct(fd=sys.stdout):
    '''Appends os.O_SYNC to the given file descriptor's flags.'''
    return fcntl.fcntl(
     fd,
     fcntl.F_SETFL,
     fcntl.fcntl(fd,fcntl.F_GETFL)
      | os.O_SYNC
    )

class ePoll:
    '''Thin contextlib wrapper around select.epoll; allows tersely watching multiple events'''
    def __init__(self, fdSpecs):
        self._E = select.epoll()
        self._fds = []
        for fd,opt in fdSpecs:
            self._E.register(fd,opt)
            self._fds.append(fd)
    def __enter__(self):
        return self._E
    def __exit__(self, exc_type, exc_value, traceback):
        for fd in self._fds:
            self._E.unregister(fd)
        self._E.close()

if __name__=='__main__':
    main()

Answer 1

事实证明，答案就像琐碎一样简单：使用EPOLLERR 而不是EPOLLHUP。

我非常怀疑这是正确解决方案*，但它确实似乎工作：

import select, subprocess, time

E = select.epoll()

p = subprocess.Popen(["sh", "-c", "sleep 3"], stdin=subprocess.PIPE)

#time.sleep(5) #Uncomment this line to convince yourself there is no race-condition here
E.register(p.stdin, select.EPOLLERR)

print("Polling...")
evs = E.poll()
print("Caught events!")

assert (p.stdin.fileno(), select.EPOLLERR) in evs

E.close()

*如果这不是正确的解决方案，那么即使是现在，我也非常想发现正确的解决方案是什么。

---

（如果有人关心，这里是原始问题中脚本的the completed version。）