Fix process termination in nilm-pipewatch

extras/sample-cron-scripts/process.sh

@@ -13,16 +13,20 @@ if ! flock -n -x 99 ; then
 fi
 trap 'rm -f "$LOCKFILE"' 0
 
-# sinefit on phase A voltage
+# redirect stdout/stderr to log, but keep it on the console too
+exec >  >(tee /home/nilm/data/process.log)
+exec 2> >(tee -a /home/nilm/data/process.log >&2)
+
+echo "sinefit on phase A voltage"
 nilm-sinefit -c 5 /sharon/raw /sharon/sinefit
 
-# prep on A, B, C with appropriate rotations
+echo "prep on A, B, C with appropriate rotations"
 nilm-prep -c 1 -r 0 /sharon/raw /sharon/sinefit /sharon/prep-a
 nilm-prep -c 2 -r 120 /sharon/raw /sharon/sinefit /sharon/prep-b
 nilm-prep -c 3 -r 240 /sharon/raw /sharon/sinefit /sharon/prep-c
 
-# decimate raw and prep data
+echo "decimate raw and prep data"
 nilm-decimate-auto /sharon/raw /sharon/prep*
 
-# run cleanup
+echo "run cleanup"
 nilm-cleanup --yes /home/nilm/data/cleanup.cfg

nilmtools/pipewatch.py

@@ -84,13 +84,16 @@ def pipewatch(args):
                                      bufsize = -1, close_fds = True,
                                      stdin = devnull,
                                      stdout = subprocess.PIPE,
-                                     stderr = None)
+                                     stderr = None,
+                                     preexec_fn = os.setpgrp)
         consumer = subprocess.Popen(args.consumer, shell = True,
                                     bufsize = -11, close_fds = True,
                                     stdin = subprocess.PIPE,
-                                    stdout = None, stderr = None)
+                                    stdout = None,
+                                    stderr = None,
+                                    preexec_fn = os.setpgrp)
 
-        queue = Queue.Queue(maxsize = 32)
+        queue = Queue.Queue(maxsize = 4)
         reader = threading.Thread(target = reader_thread,
                                   args = (queue, generator.stdout.fileno()))
         reader.start()
@@ -125,16 +128,21 @@ def pipewatch(args):
                 return proc.poll()
             try:
                 if poll_timeout(proc, 0.5) is None:
-                    proc.terminate()
+                    os.killpg(proc.pid, signal.SIGTERM)
                     if poll_timeout(proc, 0.5) is None:
-                        proc.kill()
+                        os.killpg(proc.pid, signal.SIGKILL)
             except OSError:
                 pass
             return poll_timeout(proc, 0.5)
 
         # Wait for them to die, or kill them
-        gret = kill(generator)
         cret = kill(consumer)
+        gret = kill(generator)
+
+        # Consume all remaining data in the queue until the reader
+        # and watcher threads are done
+        while reader.is_alive() or watcher.is_alive():
+            queue.get(True, 0.1)
 
         fprintf(sys.stderr, "pipewatch: generator returned %d, " +
                 "consumer returned %d\n", gret, cret)