dqm-mbProfile.py

Go to the documentation of this file.

#!/usr/bin/env python
 
import os
import collections
import logging
import resource
import time
import argparse
import subprocess
import signal
import json
import inspect
import shutil
 
LOG_FORMAT='%(asctime)s: %(name)-20s - %(levelname)-8s - %(message)s'
logging.basicConfig(format=LOG_FORMAT)
log = logging.getLogger("mbProfile")
log.setLevel(logging.INFO)
 
def read_procfs(ppath, only_ppid=True):
    def read(f):
        fp = os.path.join(ppath, f)
        with open(fp) as fd:
            return fd.read()
 
    def read_status():
        st = {}
 
        fp = os.path.join(ppath, "status")
        with open(fp) as fd:
            for line in fd.readlines():
                if not line: continue
 
                key, value = line.split(":", 1)
                st[key] = value.strip()
 
        return st
 
    try:
        dct = {}
 
        dct["statm"] = read("statm").strip()
        dct["stat"] = read("stat").strip()
        dct["cmdline"] = read("cmdline").strip().replace("\0", " ")
 
        status = read_status()
        dct["status"] = status
        dct["pid"] = int(status["Pid"])
        dct["parent_pid"] = int(status["PPid"])
 
        return dct
    except:
        log.warning("Exception in read_procfs.", exc_info=True)
        pass
 
def build_process_list():
    lst = os.listdir("/proc/")
    for f in lst:
        if not f.isdigit(): continue
 
        proc = read_procfs(os.path.join("/proc", f))
        if proc:
            yield proc
 
def get_children(ppid):
    """ Select all processes which are descendant from ppid (exclusive). """
 
    pid_dct = {}
    for proc in build_process_list():
        proc["_children"] = []
        pid_dct[proc["pid"]] = proc
 
    # fill in children array
    for pid in list(pid_dct.keys()):
        parent_pid = pid_dct[pid]["parent_pid"]
 
        if parent_pid in pid_dct:
            pid_dct[parent_pid]["_children"].append(pid)
 
    # now just walk down the tree
    if ppid is None or ppid not in pid_dct:
        # process has quit, we exit
        return []
 
    accepted = []
    to_accept = collections.deque([ppid, ])
    
    while to_accept:
        head = pid_dct[to_accept.popleft()]
 
        # do not include the monitoring pid
        if head["pid"] != ppid:
            accepted.append(head)
 
        to_accept.extend(head.get("_children", []))
        head["children"] = head["_children"]
        del head["_children"]
 
        # deleting children breaks infinite loops
        # but Dima, can a process tree contain a loop? yes - via race-condition in reading procfs
 
    return accepted
 
class Profile(object):
    def __init__(self, args):
        self.time = time.time()
        self.final = False
        self.pid = None 
        self.known_pids = {}
 
        self.ru = {}
        self.ru_diff = {}
 
        self._offset_ru = None
        self._args = args
 
        if self._args.file:
            self._file = open(self._args.file, "w")
        else:
            self._file = None
 
        self.update()
 
    def update_ru(self):
        fields_to_subtract = (
            "ru_utime", "ru_stime", "ru_maxrss", "ru_minflt", "ru_majflt", "ru_nswap",
            "ru_inblock", "ru_oublock", "ru_msgsnd", "ru_msgrcv", "ru_nsignals", "ru_nvcsw", "ru_nivcsw",
        )
 
        rusage = resource.getrusage(resource.RUSAGE_CHILDREN)
        self.ru = rusage
 
        if self._offset_ru is None:
            self._offset_ru = rusage
 
        for field in fields_to_subtract:
            current = getattr(self.ru, field)
            base = getattr(self._offset_ru, field)
 
            self.ru_diff[field] = current - base
 
    # this is taken from: http://github.com/pixelb/scripts/commits/master/scripts/ps_mem.py
    def read_smaps(self, proc_dict):
        Private, Shared, Pss = 0, 0, 0
 
        fp = os.path.join("/proc/%d" % proc_dict["pid"], "smaps")
        with open(fp) as fd:
            for line in fd.readlines():
                if line.startswith("Shared"):
                    Shared += int(line.split()[1])
                elif line.startswith("Private"):
                    Private += int(line.split()[1])
                elif line.startswith("Pss"):
                    Pss += int(line.split()[1])
    
        proc_dict["smaps_shared"] = Shared * 1024
        proc_dict["smaps_private"] = Private * 1024
        proc_dict["smaps_pss"] = Pss * 1024
 
    def update_proc(self):
        procs = get_children(os.getpid())
 
        # we can only do it here, permision-wise
        # ie only for owned processes
        for proc in procs:
            try:
                self.read_smaps(proc)
            except:
                log.warning("Exception in read_smaps.", exc_info=True)
 
        # we need to mark not-running ones as such
        stopped = set(self.known_pids.keys())
        for proc in procs:
            proc["running"] = True
 
            pid = proc["pid"]
            self.known_pids[pid] = proc
 
            if pid in stopped:
                stopped.remove(pid)
 
        for pid in stopped:
            self.known_pids[pid]["running"] = False
 
    def update(self):
        self.time = time.time()
 
        self.update_ru()
        self.update_proc()
 
        if self._file:
            json.dump(self.to_dict(), self._file)
            self._file.write("\n")
            self._file.flush()
 
        log.info("Written profile to: %s, took=%.03f", self._args.file, time.time() - self.time)
 
    def to_dict(self):
        dct = collections.OrderedDict()
        dct['time']         = self.time
        dct['pid']          = self.pid
        dct['final']        = self.final
        
        dct['ru_diff']      = dict(self.ru_diff)
        dct['ru']           = dict((k, v) for k, v in inspect.getmembers(self.ru) if k.startswith('ru_'))
        dct['known_pids']   = dict(self.known_pids)
        return dct
    
    def finish(self):
        self.final = True
        self.update()
 
        if self._file:
            self._file.close()
            self._file = None
        else:
            log.info("ru_diff: %s", self.ru_diff)
 
 
ALARM_TIMER = 1
ALARM_P_OBJECT = None
 
def handle_alarm(num, frame):
    if ALARM_P_OBJECT:
        ALARM_P_OBJECT.update()
 
    signal.alarm(ALARM_TIMER)
 
def run_and_monitor(args):
    profile = Profile(args)
 
    proc = subprocess.Popen(args.pargs)
    profile.pid = proc.pid
 
    global ALARM_P_OBJECT
    ALARM_P_OBJECT = profile
 
    signal.signal(signal.SIGALRM, handle_alarm)
    signal.alarm(ALARM_TIMER)
 
    proc.wait()
    profile.finish()
 
def find_and_write_html(p, args):
    # create the dir if necessary
    if p and not os.path.exists(p):
        os.makedirs(p)
 
    html_paths = [
        os.path.join(os.getenv("CMSSW_BASE"),"src/DQMServices/Components/data/html"),
        os.path.join(os.getenv("CMSSW_RELEASE_BASE"), "src/DQMServices/Components/data/html"),
    ]
 
    def find_file(f):
        fails = []
        for p in html_paths:
            x = os.path.join(p, f)
            if os.path.exists(x):
                return x
            else:
                fails.append(x)
 
        log.warning("Could not find html file: %s (%s)", f, fails)
 
    for f in ['mbGraph.js', 'mbGraph.html']:
        target_fn = os.path.join(p, f)
        source_fn = find_file(f)
        if source_fn:
            log.info("Copying %s to %s", source_fn, target_fn)
            shutil.copyfile(source_fn, target_fn)
 
    # create json file
    target_fn = os.path.join(p, "mbGraph.json")
    log.info("Creating %s", target_fn)
    with open(target_fn, "w") as fp:
        dct = {
            "file": os.path.basename(args.file),
            "interval": args.i,
            "env": {
                "CMSSW_GIT_HASH": os.getenv("CMSSW_GIT_HASH"),
                "CMSSW_RELEASE_BASE": os.getenv("CMSSW_RELEASE_BASE"),
                "SCRAM_ARCH": os.getenv("SCRAM_ARCH"),
            },
        }
 
        json.dump(dct, fp, indent=2)
 
 
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Profile child processes and produce data for rss and such graphs.")
    parser.add_argument("-f", "--file", type=str, default="performance.json", help="Filename to write.", metavar="performance.json")
    parser.add_argument("-i", type=int, help="Time interval between profiles.", default=15)
    parser.add_argument('-q', action='store_true', help="Reduce logging.")
    parser.add_argument('-w', action='store_true', help="Write html helper files for rendering the performance file.")
    parser.add_argument('pargs', nargs=argparse.REMAINDER)
 
    args = parser.parse_args()
 
    if not args.pargs:
        parser.print_help()
        sys.exit(-1)
    elif args.pargs[0] == "--":
        # compat with 2.6
        args.pargs = args.pargs[1:]
 
    ALARM_TIMER = args.i
 
    if args.q:
        log.setLevel(logging.WARNING)
 
    if args.w:
        p = os.path.dirname(args.file)
        find_and_write_html(p, args)
 
    
 
    run_and_monitor(args)