Polling/awaiting HUP on Popen.stdin

Viewed 201

I'm writing a Zenity-based status monitor for GNU dd(1) in Python. Due to the constraints of the system I'm targeting, the wrapper must run on Python 2, and cannot pull in external libraries.

Among the requirements are for Zenity's "Cancel" button to terminate dd if it's not yet finished.

I must do the following without delay (i.e. triggered/driven/immediately); if multiple of the below conditions are met/triggered simultaneously, they are to be executed in the listed order:

  • when Zenity exits, terminate dd
  • when dd writes to its stderr, munge+forward that data to Zenity's stdin
  • when dd exits, if its return code was not 0, terminate Zenity

However, the epoll object seems to only be triggering on dd's output; it never triggers on Zenity exiting, despite my registering EPOLLHUP on Zenity's stdin.

How should/can this be done? I understand that epoll is the only primitive that can be used to correctly trigger on dd's output (via EPOLLIN); I also understand that it is an unweildy primitive which may be unsuitable for triggering on Zenity's exiting. (If required, I can implement more logic into this file; doing so is infinitely preferable to pulling in any 3rd-party library, however small or "common". I reiterate that I understand that epoll is unwieldy to work with and may require a large amount of glue-logic.)

Alternatively: if epoll is not the correct primitive for monitoring a subprocess for its exit, then what is the correct way to monitor a subprocess for exit while monitoring a subprocess for output in a Python 2 compatible way?

(I do not inherently need multithreaded capabilities; to take all actions sequentially would be fully in-spec; though, if multithreaded programming would be absolutely necessary in this case in order to avoid a busy-loop, then so be it.)


Below is my full code so far.

#!/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()
1 Answers

It turns out the answer was as trivially simple as: Use EPOLLERR instead of EPOLLHUP.

I have serious doubts that this the right solution*, but it certainly seems to work:

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()

*If this isn't the right solution, then I would, even now, very much like to discover what the right solution is.


(Here's the completed version of the script from the original question, if anyone cares.)

Related