Watch for process termination too

Makefile

@@ -8,7 +8,10 @@ else
 	@echo "Try 'make install'"
 endif
 
-test: test_trainola
+test: test_pipewatch
+
+test_pipewatch:
+	nilmtools/pipewatch.py -t 3 "seq 10 20" "seq 20 30"
 
 test_trainola:
 	-nilmtool -u http://bucket/nilmdb remove -s min -e max \

nilmtools/pipewatch.py

@@ -0,0 +1,155 @@
+#!/usr/bin/python
+
+import nilmdb.client
+from nilmdb.utils.printf import *
+import nilmdb.utils.lock
+import nilmtools
+
+import time
+import sys
+import os
+import argparse
+import subprocess
+import tempfile
+import threading
+import select
+import signal
+import Queue
+
+def parse_args(argv = None):
+    parser = argparse.ArgumentParser(
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter,
+        version = nilmtools.__version__,
+        description = """\
+    Pipe data from 'generator' to 'consumer'.  This is intended to be
+    executed frequently from cron, and will exit if another copy is
+    already running.  If 'generator' or 'consumer' returns an error,
+    or if 'generator' stops sending data for a while, it will exit.
+
+    Intended for use with ethstream (generator) and nilm-insert
+    (consumer).  Commands are executed through the shell.
+    """)
+    parser.add_argument("-l", "--lock", metavar="FILENAME", action="store",
+                        default=tempfile.gettempdir() +
+                        "/nilm-pipewatch.lock",
+                        help="Lock file for detecting running instance")
+    parser.add_argument("-t", "--timeout", metavar="SECONDS", action="store",
+                        type=float, default=30,
+                        help="Restart if no output from " +
+                        "generator for this long")
+    group = parser.add_argument_group("commands to execute")
+    group.add_argument("generator", action="store",
+                       help="Data generator (e.g. \"ethstream -r 8000\")")
+    group.add_argument("consumer", action="store",
+                       help="Data consumer (e.g. \"nilm-insert /foo/bar\")")
+    args = parser.parse_args(argv)
+
+    return args
+
+def reader_thread(queue, fd):
+    # Read from a file descriptor, write to queue.
+    try:
+        while True:
+            (r, w, x) = select.select([fd], [], [fd], 0.25)
+            if x:
+                raise Exception # generator died?
+            if not r:
+                # short timeout -- just try again.  This is to catch the
+                # fd being closed elsewhere, which is only detected
+                # when select restarts.
+                continue
+            data = os.read(fd, 65536)
+            if data == "": # generator EOF
+                raise Exception
+            queue.put(data)
+    except Exception:
+        queue.put(None)
+
+def watcher_thread(queue, procs):
+    # Put None in the queue if either process dies
+    while True:
+        for p in procs:
+            if p.poll() is not None:
+                queue.put(None)
+                return
+        time.sleep(0.25)
+
+def main(argv = None):
+    args = parse_args(argv)
+
+    with open(args.lock, "w") as lockfile:
+        if not nilmdb.utils.lock.exclusive_lock(lockfile):
+            printf("pipewatch process already running (according to %s)\n",
+                   args.lock)
+            sys.exit(0)
+
+
+        with open(os.devnull, "r") as devnull:
+            generator = subprocess.Popen(args.generator, shell = True,
+                                         bufsize = -1, close_fds = True,
+                                         stdin = devnull,
+                                         stdout = subprocess.PIPE,
+                                         stderr = None)
+            consumer = subprocess.Popen(args.consumer, shell = True,
+                                        bufsize = -11, close_fds = True,
+                                        stdin = subprocess.PIPE,
+                                        stdout = None, stderr = None)
+
+            queue = Queue.Queue(maxsize = 32)
+            reader = threading.Thread(target = reader_thread,
+                                      args = (queue, generator.stdout.fileno()))
+            reader.start()
+            watcher = threading.Thread(target = watcher_thread,
+                                       args = (queue, [generator, consumer]))
+            watcher.start()
+            try:
+                while True:
+                    try:
+                        data = queue.get(True, args.timeout)
+                        if data is None:
+                            break
+                        consumer.stdin.write(data)
+                    except Queue.Empty:
+                        # Timeout: kill the generator
+                        fprintf(sys.stderr, "pipewatch: timeout\n")
+                        generator.terminate()
+                        break
+
+                generator.stdout.close()
+                consumer.stdin.close()
+            except IOError:
+                fprintf(sys.stderr, "pipewatch: I/O error\n")
+
+            def kill(proc):
+                # Wait for a process to end, or kill it
+                def poll_timeout(proc, timeout):
+                    for x in range(1+int(timeout / 0.1)):
+                        if proc.poll() is not None:
+                            break
+                        time.sleep(0.1)
+                    return proc.poll()
+                try:
+                    if poll_timeout(proc, 0.5) is None:
+                        proc.terminate()
+                        if poll_timeout(proc, 0.5) is None:
+                            proc.kill()
+                except OSError:
+                    pass
+                return poll_timeout(proc, 0.5)
+
+            # Wait for them to die, or kill them
+            gret = kill(generator)
+            cret = kill(consumer)
+
+            fprintf(sys.stderr, "pipewatch: generator returned %d, " +
+                    "consumer returned %d\n", gret, cret)
+            if gret == 0 and cret == 0:
+                sys.exit(0)
+            sys.exit(1)
+    try:
+        os.unlink(args.lock)
+    except OSError:
+        pass
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -80,6 +80,7 @@ setup(name='nilmtools',
               'nilm-cleanup = nilmtools.cleanup:main',
               'nilm-median = nilmtools.median:main',
               'nilm-trainola = nilmtools.trainola:main',
+              'nilm-pipewatch = nilmtools.pipewatch:main',
               ],
           },
       zip_safe = False,