Use NilmDB serializer for ProcessManager

README.txt

@@ -6,10 +6,8 @@ Prerequisites:
   # Runtime and build environments
   sudo apt-get install python2.7 python-setuptools
 
-  # Base dependencies
+  # Plus nilmdb and its dependencies
-  sudo apt-get install python-numpy python-scipy
+  nilmdb (1.8.3+)
-
-  nilmdb (1.8.0+)
 
 Install:
 

nilmrun/processmanager.py

@@ -43,9 +43,8 @@ class LogReceiver(object):
 
 class Process(object):
     """Spawn and manage a subprocess, and capture its output."""
-    def __init__(self, name, argv, tempfile = None):
+    def __init__(self, argv, tempfile = None):
         self.start_time = None
-        self.name = name
 
         # Use a pipe for communicating log data
         (rpipe, wpipe) = os.pipe()
@@ -82,38 +81,41 @@ class Process(object):
     def terminate(self, timeout = 1.0):
         """Terminate a process, and all of its children that are in the same
         process group."""
-        # First give it some time to die on its own
+        try:
-        if self._join(timeout):
+            # First give it some time to die on its own
+            if self._join(timeout):
+                return True
+
+            def getpgid(pid):
+                try:
+                    return os.getpgid(pid)
+                except OSError: # pragma: no cover
+                    return None
+
+            # Find all children
+            group = getpgid(self._process.pid)
+            main = psutil.Process(self._process.pid)
+            allproc = [ main ] + main.get_children(recursive = True)
+
+            # Kill with SIGTERM, if they're still in this process group
+            for proc in allproc:
+                if getpgid(proc.pid) == group:
+                    os.kill(proc.pid, signal.SIGTERM)
+
+            # Wait for it to die again
+            if self._join(timeout):
+                return True
+
+            # One more try with SIGKILL
+            for proc in allproc:
+                if getpgid(proc.pid) == group:
+                    os.kill(proc.pid, signal.SIGKILL)
+
+            # See if it worked
+            return self._join(timeout)
+        except psutil.Error: # pragma: no cover (race condition)
             return True
 
-        def getpgid(pid):
-            try:
-                return os.getpgid(pid)
-            except OSError: # pragma: no cover
-                return None
-
-        # Find all children
-        group = getpgid(self._process.pid)
-        main = psutil.Process(self._process.pid)
-        allproc = [ main ] + main.get_children(recursive = True)
-
-        # Kill with SIGTERM, if they're still in this process group
-        for proc in allproc:
-            if getpgid(proc.pid) == group:
-                os.kill(proc.pid, signal.SIGTERM)
-
-        # Wait for it to die again
-        if self._join(timeout):
-            return True
-
-        # One more try with SIGKILL
-        for proc in allproc:
-            if getpgid(proc.pid) == group:
-                os.kill(proc.pid, signal.SIGKILL)
-
-        # See if it worked
-        return self._join(timeout)
-
     def clear_log(self):
         self._log.clear()
 
@@ -129,6 +131,49 @@ class Process(object):
     def exitcode(self):
         return self._process.returncode
 
+    def get_info_prepare(self):
+        """Prepare the process list and measurement for .get_info.
+        Call .get_info() about a second later."""
+        try:
+            main = psutil.Process(self._process.pid)
+            self._process_list = [ main ] + main.get_children(recursive = True)
+            for proc in self._process_list:
+                proc.get_cpu_percent(0)
+        except psutil.Error: # pragma: no cover (race condition)
+            self._process_list = [ ]
+
+    @staticmethod
+    def get_empty_info():
+        return { "cpu_percent": 0,
+                 "cpu_user": 0,
+                 "cpu_sys": 0,
+                 "mem_phys": 0,
+                 "mem_virt": 0,
+                 "io_read": 0,
+                 "io_write": 0,
+                 "procs": 0 }
+
+    def get_info(self):
+        """Return a dictionary with info about the process CPU and memory
+        usage.  Call .get_info_prepare() about a second before this."""
+        d = self.get_empty_info()
+        for proc in self._process_list:
+            try:
+                d["cpu_percent"] += proc.get_cpu_percent(0)
+                cpuinfo = proc.get_cpu_times()
+                d["cpu_user"] += cpuinfo.user
+                d["cpu_sys"] += cpuinfo.system
+                meminfo = proc.get_memory_info()
+                d["mem_phys"] += meminfo.rss
+                d["mem_virt"] += meminfo.vms
+                ioinfo = proc.get_io_counters()
+                d["io_read"] += ioinfo.read_bytes
+                d["io_write"] += ioinfo.write_bytes
+                d["procs"] += 1
+            except psutil.Error:
+                pass
+        return d
+
 class ProcessManager(object):
     """Track and manage a collection of Process objects"""
     def __init__(self):
@@ -143,7 +188,7 @@ class ProcessManager(object):
     def __getitem__(self, key):
         return self.processes[key]
 
-    def run_code(self, procname, code, args):
+    def run_code(self, code, args):
         """Evaluate 'code' as if it were placed into a Python file and
         executed.  The arguments, which must be strings, will be
         accessible in the code as sys.argv[1:]."""
@@ -154,13 +199,13 @@ class ProcessManager(object):
         with os.fdopen(fd, 'w') as f:
             f.write(code)
         argv = [ sys.executable, "-B", "-s", "-u", path ] + args
-        pid = self.run_command(procname, argv)
+        pid = self.run_command(argv)
         self.tmpfiles[pid] = path
         return pid
 
-    def run_command(self, procname, argv):
+    def run_command(self, argv):
         """Execute a command line program"""
-        new = Process(procname, argv)
+        new = Process(argv)
         self.processes[new.pid] = new
         return new.pid
 
@@ -175,3 +220,37 @@ class ProcessManager(object):
                 pass
             del self.tmpfiles[pid]
         del self.processes[pid]
+
+    def get_info(self):
+        """Get info about all running PIDs"""
+        info = { "total" : Process.get_empty_info(),
+                 "pids" : {},
+                 "system" : {}
+                 }
+
+        # Trigger CPU usage collection
+        for pid in self:
+            self[pid].get_info_prepare()
+        psutil.cpu_percent(0, percpu = True)
+
+        # Give it some time
+        time.sleep(1)
+
+        # Retrieve info for system
+        info["system"]["cpu_percent"] = sum(psutil.cpu_percent(0, percpu=True))
+        info["system"]["cpu_max"] = 100.0 * psutil.NUM_CPUS
+        info["system"]["procs"] = len(psutil.get_pid_list())
+        # psutil > 0.6.0's psutil.virtual_memory() would be better here,
+        # but this should give the same info.
+        meminfo = psutil.phymem_usage()
+        info["system"]["mem_total"] = meminfo.total
+        info["system"]["mem_used"] = int(meminfo.total * meminfo.percent / 100)
+
+        # Retrieve info for each PID
+        for pid in self:
+            info["pids"][pid] = self[pid].get_info()
+            # Update totals
+            for key in info["total"]:
+                info["total"][key] += info["pids"][pid][key]
+
+        return info

nilmrun/server.py

@@ -5,10 +5,7 @@ import sys
 import os
 import socket
 import simplejson as json
-import decorator
-import psutil
 import traceback
-import argparse
 import time
 
 import nilmdb
@@ -23,7 +20,9 @@ from nilmdb.server.serverutil import (
         json_error_page,
         cherrypy_start,
         cherrypy_stop,
+        bool_param,
         )
+from nilmdb.utils import serializer_proxy
 import nilmrun
 import nilmrun.testfilter
 
@@ -62,13 +61,16 @@ class AppProcess(object):
         self.manager = manager
 
     def process_status(self, pid):
+        # We need to convert the log (which is bytes) to Unicode
+        # characters, in order to send it via JSON.  Treat it as UTF-8
+        # but replace invalid characters with markers.
+        log = self.manager[pid].log.decode('utf-8', errors='replace')
         return {
             "pid": pid,
             "alive": self.manager[pid].alive,
             "exitcode": self.manager[pid].exitcode,
-            "name": self.manager[pid].name,
             "start_time": self.manager[pid].start_time,
-            "log": self.manager[pid].log,
+            "log": log
             }
 
     # /process/status
@@ -77,6 +79,7 @@ class AppProcess(object):
     def status(self, pid, clear = False):
         """Return status about a process.  If clear = True, also clear
         the log."""
+        clear = bool_param(clear)
         if pid not in self.manager:
             raise cherrypy.HTTPError("404 Not Found", "No such PID")
         status = self.process_status(pid)
@@ -91,6 +94,14 @@ class AppProcess(object):
         """Return a list of processes in the manager."""
         return list(self.manager)
 
+    # /process/info
+    @cherrypy.expose
+    @cherrypy.tools.json_out()
+    def info(self):
+        """Return detailed CPU and memory info about the system and
+        all processes"""
+        return self.manager.get_info()
+
     # /process/remove
     @cherrypy.expose
     @cherrypy.tools.json_in()
@@ -124,7 +135,7 @@ class AppRun(object):
         if not isinstance(argv, list):
             raise cherrypy.HTTPError("400 Bad Request",
                                      "argv must be a list of strings")
-        return self.manager.run_command("command", argv)
+        return self.manager.run_command(argv)
 
     # /run/code
     @cherrypy.expose
@@ -132,16 +143,18 @@ class AppRun(object):
     @cherrypy.tools.json_out()
     @exception_to_httperror(nilmrun.processmanager.ProcessError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
-    def code(self, code, args):
+    def code(self, code, args = None):
         """Execute arbitrary Python code.  'code' is a formatted string.
         It will be run as if it were written into a Python file and
         executed.  'args' is a list of strings, and they are passed
         on the command line as additional arguments (i.e., they end up
         in sys.argv[1:])"""
+        if args is None:
+            args = []
         if not isinstance(args, list):
             raise cherrypy.HTTPError("400 Bad Request",
                                      "args must be a list of strings")
-        return self.manager.run_code("usercode", code, args)
+        return self.manager.run_code(code, args)
 
 class Server(object):
     def __init__(self, host = '127.0.0.1', port = 8080,
@@ -192,8 +205,11 @@ class Server(object):
         # error messages.
         cherrypy._cperror._ie_friendly_error_sizes = {}
 
+        # The manager maintains internal state and isn't necessarily
+        # thread-safe, so wrap it in the serializer.
+        manager = serializer_proxy(nilmrun.processmanager.ProcessManager)()
+
         # Build up the application and mount it
-        manager = nilmrun.processmanager.ProcessManager()
         root = App()
         root.process = AppProcess(manager)
         root.run = AppRun(manager)

setup.py

@@ -61,10 +61,10 @@ setup(name='nilmrun',
       long_description = "NILM Database Filter Runner",
       license = "Proprietary",
       author_email = 'jim@jtan.com',
-      install_requires = [ 'nilmdb >= 1.8.0',
+      install_requires = [ 'nilmdb >= 1.8.3',
-                           'nilmtools >= 1.2.2',
+                           'psutil >= 0.3.0',
-                           'numpy',
+                           'cherrypy >= 3.2',
-                           'scipy',
+                           'simplejson',
                            ],
       packages = [ 'nilmrun',
                    'nilmrun.scripts',
@@ -75,7 +75,6 @@ setup(name='nilmrun',
       entry_points = {
           'console_scripts': [
               'nilmrun-server = nilmrun.scripts.nilmrun_server:main',
-              'nilm-trainola = nilmrun.trainola:main',
               ],
           },
       zip_safe = False,

tests/test_nilmrun.py

@@ -65,6 +65,7 @@ class TestClient(object):
             status = client.get("process/status", { "pid": pid })
             if status["alive"] == False:
                 break
+            time.sleep(0.1)
         else:
             raise AssertionError("process " + str(pid) + " didn't die in " +
                                  str(timeout) + " seconds: " + repr(status))
@@ -145,7 +146,7 @@ class TestClient(object):
 
         # Verify that status looks OK
         status = client.get("process/status", { "pid": pid, "clear": True })
-        for x in [ "pid", "alive", "exitcode", "name", "start_time", "log" ]:
+        for x in [ "pid", "alive", "exitcode", "start_time", "log" ]:
             in_(x, status)
         in_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
         eq_(status["alive"], True)
@@ -248,6 +249,7 @@ class TestClient(object):
             sys.stdout.flush()
             if status["alive"] == False:
                 break
+            time.sleep(0.1)
         status = client.post("process/remove", { "pid": pid })
         os._exit(int(status["exitcode"]))
 
@@ -256,7 +258,10 @@ class TestClient(object):
         eq_(client.get("process/list"), [])
 
         def do(code, args, kill):
-            pid = client.post("run/code", { "code": code, "args": args } )
+            if args is not None:
+                pid = client.post("run/code", { "code": code, "args": args } )
+            else:
+                pid = client.post("run/code", { "code": code } )
             eq_(client.get("process/list"), [pid])
             if kill:
                 return self.wait_kill(client, pid)
@@ -321,6 +326,15 @@ class TestClient(object):
         eq_(status["log"], "hello\n")
         ne_(status["exitcode"], 0)
 
+        # default arguments are empty
+        code = textwrap.dedent("""
+        import sys
+        print 'args:', len(sys.argv[1:])
+        """)
+        status = do(code, None, False)
+        eq_(status["log"], "args: 0\n")
+        eq_(status["exitcode"], 0)
+
     def test_client_08_bad_types(self):
         client = HTTPClient(baseurl = testurl, post_json = True)
 
@@ -331,3 +345,48 @@ class TestClient(object):
         with assert_raises(ClientError) as e:
             client.post("run/command", { "argv": "asdf" })
         in_("must be a list", str(e.exception))
+
+    def test_client_09_info(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        # start some processes
+        a = client.post("run/command", { "argv": ["sleep","60"] } )
+        b = client.post("run/command", { "argv": ["sh","-c","sleep 2;true"] } )
+        c = client.post("run/command", { "argv": ["sh","-c","burnP5;true"] } )
+        d = client.post("run/command", { "argv": ["burnP5" ] } )
+
+        info = client.get("process/info")
+        eq_(info["pids"][a]["procs"], 1)
+        eq_(info["pids"][b]["procs"], 2)
+        eq_(info["pids"][c]["procs"], 2)
+        eq_(info["pids"][d]["procs"], 1)
+        eq_(info["total"]["procs"], 6)
+        lt_(info["pids"][a]["cpu_percent"], 50)
+        lt_(20, info["pids"][c]["cpu_percent"])
+        lt_(80, info["system"]["cpu_percent"])
+
+        time.sleep(2)
+        info = client.get("process/info")
+        eq_(info["pids"][b]["procs"], 0)
+
+        # kill all processes
+        for pid in client.get("process/list"):
+            client.post("process/remove", { "pid": pid })
+
+    def test_client_10_unicode(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        eq_(client.get("process/list"), [])
+        def verify(cmd, result):
+            pid = client.post("run/command", { "argv": [ "sh", "-c", cmd ] })
+            eq_(client.get("process/list"), [pid])
+            status = self.wait_end(client, pid)
+            eq_(result, status["log"])
+
+        # Unicode should work
+        verify("echo -n hello", "hello")
+        verify(u"echo -n ☠", u"☠")
+        verify("echo -ne \\\\xe2\\\\x98\\\\xa0", u"☠")
+
+        # Programs that spit out invalid UTF-8 should get replacement
+        # markers
+        verify("echo -ne \\\\xae", u"\ufffd")