root/pyconquer.py

#!/usr/bin/python
"""PyConquer, a concurrent-thread analyzer."""

usage = """PyConquer tracing tool
    
    Usage:
        pyconquer.py [-h] [-e=<event list>] [-f=path] [-n] [-r=.*] MODULE.py [ARG1 ARG2 ...]
    
    -e --events=<event list>: Trace the given events (separate by commas).
        Available events:
            'c_events' = ('c_call', 'c_return', 'c_exception')
            'non_c_events' = ('call', 'return', 'exception', 'line')
            'all_events' = c_events + non_c_events
            'all_call_events' = ('call', 'return', 'c_call', 'c_return')
    
    -f --filename=<path>: the file where trace output will be written.
        If omitted, output goes to stdout.
    
    -n --notimes: do not emit call times in the output.
    
    -r --regex=<regular expression>: limit traced modules to filenames
        matching the given regular expression. If omitted, all modules
        are traced. If the given regex begins with "!", filenames which
        do NOT match will be traced.
    """

import os
import re

try:
    set
except NameError:
    from sets import Set as set

import sys
import thread
import threading
import time
import traceback


non_c_events = ('call', 'return', 'exception', 'line')
c_events = ('c_call', 'c_return', 'c_exception')
all_events = c_events + non_c_events
all_call_events = ('call', 'return', 'c_call', 'c_return')

def scope_name(frame):
    path, name = os.path.split(frame.f_code.co_filename)
    # Strip off any file extension
    name = os.path.splitext(name)[0]
    if name == "__init__":
        return "/".join((os.path.split(path)[1], name))
    else:
        return name

def is_thread_start(frame, event):
    """Return True if current frame/event is threading.Thread.start()."""
    if event == 'call' and frame.f_code.co_name == 'start':
        localself = frame.f_locals.get('self')
        if localself and isinstance(localself, threading.Thread):
            return True
    return False

def is_thread_stop(frame, event):
    """Return True if current frame/event is threading.Thread.__stop()."""
    if event == 'return' and frame.f_code.co_name == '__stop':
        # Process threading.Thread.__stop()
        localself = frame.f_locals.get('self')
        if localself and isinstance(localself, threading.Thread):
            return True
    return False


TERMINATED = 0
RUNNING = 1

class Swimlane:
    """A thread (since thread id's get re-used, this differentiates them)."""
    
    def __init__(self, index):
        self.indent = 0
        self.index = index
        self.state = RUNNING
        self.calltimes = []
        self.ccalltimes = []


class Tracer:
    
    def __init__(self, fileregex=".*", events=None):
        self.active = False
        
        self.swimlanes = {}
        self.numlanes = 0
        self.tabsize = 8
        
        self.fileregex = fileregex
        if events is None:
            events = ['call', 'return', 'exception']
        self.events = set(events)
    
    def start(self, caller=None):
        self._using_profile_hook = False
        for e in c_events:
            if e in self.events:
                # Must also use the profile hook,
                # since tracing does not support C events.
                threading.setprofile(self.profhook)
                sys.setprofile(self.profhook)
                self._using_profile_hook = True
                break
        
        # Set the trace hook for all events other than C events
        # (it's easier to clean up nicely).
        threading.settrace(self.hook)
        sys.settrace(self.hook)
        
        # Set the calling frame's local trace func to our hook.
        if caller is None:
            caller = sys._getframe().f_back
        caller.f_trace = self.hook
        
        # Set '_caller_lineno' to help with writing tests.
        self._caller_lineno = caller.f_lineno
        
        self.active = True
    
    def stop(self, caller=None):
        self.active = False
        
        if threading._trace_hook == self.hook:
            threading.settrace(None)
        if threading._profile_hook == self.profhook:
            threading.setprofile(None)
        
        # Ugly hack to get around no sys.gettrace().
        if caller is None:
            caller = sys._getframe().f_back
        if caller.f_trace == self.hook:
            sys.settrace(None)
        if caller.f_back and caller.f_back.f_trace == self.hook:
            sys.settrace(None)
        
        # Ugly hack to get around no sys.getprofile().
        if self._using_profile_hook:
            sys.setprofile(None)
    
    def profhook(self, frame, event, arg):
        if not event.startswith("c_"):
            return
        self.hook(frame, event, arg)
    
    def hook(self, frame, event, arg):
        # Base method; must be overridden.
        pass



#                                 Logging                                 #


class WatchedValue:
    
    def __init__(self, obj, attr, default=None, name=None):
        self.obj = obj
        self.attr = attr
        self.default = default
        
        if name is None:
            try:
                name = ("%s.%s" % (obj.__class__.__module__, obj.__class__.__name__))
            except AttributeError:
                name = repr(obj)
            name = "%s.%s" % (name, attr)
        self.name = name
        
        self.oldvalue = self.getvalue()
    
    def getvalue(self):
        return getattr(self.obj, self.attr, self.default)


class Logger(Tracer):
    
    def __init__(self, fileregex=".*", events=None):
        Tracer.__init__(self, fileregex, events)
        self.currentlane = Swimlane(-1)
        self.watches = {}
        self.out = sys.stdout
        self.time_calls = True
        
        # Don't use an RLock here! hook() will hang because Thread.__delete
        # will acquire the _active_limbo_lock, and then RLock.acquire()
        # will call currentThread (which creates a _DummyThread, which will
        # try in its __init__ method to acquire the _active_limbo_lock).
        self._mutex = threading.Lock()
    
    def start(self):
        self.out.write("\n\n________ pyconquer tracing started ________\n")
        self.write_watches()
        Tracer.start(self, sys._getframe().f_back)
    
    def stop(self):
        self.currentlane = Swimlane(-1)
        Tracer.stop(self, sys._getframe().f_back)
        self.out.write("\n\n________ pyconquer tracing stopped ________\n")
    
    def hook(self, frame, event, arg):
        if not self.active:
            return
        
        self._mutex.acquire(True)
        try:
            filename = frame.f_code.co_filename
            
            # Deny tracing this module itself.
            if r"\pyconquer.py" in filename:
                return self.hook
            
            # Handle this event if requested.
            if event in self.events:
                if self.fileregex.startswith("!"):
                    process = not re.search(self.fileregex[1:], filename)
                else:
                    process = re.search(self.fileregex, filename)
                if not process:
                    if is_thread_start(frame, event):
                        process = True
                        event = 'threadstart'
                    elif is_thread_stop(frame, event):
                        process = True
                        event = 'threadstop'
                
                if process:
                    self.process_event(frame, event, arg)
            
            # See if any of our watched attributes have changed.
            self.write_watches(changes_only=True)
            
            return self.hook
        finally:
            self._mutex.release()
    
    def write_lane(self, msg, indent=None):
        lane = self.currentlane
        tab = " " * (self.tabsize * lane.index)
        if indent is None:
            indent = lane.indent * 2
        self.out.write("\n%s%s%s" % (tab, "-" * indent, msg))
    
    
    #                              Processors                              #
    
    def process_event(self, frame, event, arg):
        # Locate the current lane.
        ct = thread.get_ident()
        lane = self.swimlanes.get(ct)
        if lane is None:
            # Since we're logging only (and not verifying), re-use
            # the lane index to reduce right-shift.
            for k, v in self.swimlanes.items():
                if v.state == TERMINATED:
                    index = v.index
                    break
            else:
                index = self.numlanes
                self.numlanes += 1
            lane = self.swimlanes[ct] = Swimlane(index)
        elif lane.state == TERMINATED:
            # Terminated threads may have their ID's re-used;
            # but use the same lane to reduce right-shift
            lane = self.swimlanes[ct] = Swimlane(lane.index)
        
        if event == 'threadstop':
            lane.state = TERMINATED
        
        if not (lane is self.currentlane):
            self.currentlane = lane
            
            # Write a header row with the current thread's name.
            self.out.write("\n")
            self.write_lane("[ %s ]" % threading.currentThread().getName())
        
        getattr(self, "process_" + event)(frame, arg)
    
    def process_call(self, frame, arg):
        self.currentlane.indent += 1
        name = frame.f_code.co_name
        if name == "?":
            self.write_lane("> import %s" % frame.f_code.co_filename)
        else:
            self.write_lane("> %s (%s:%s)" %
                            (name, scope_name(frame), frame.f_lineno))
        if self.time_calls:
            self.currentlane.calltimes.append(time.clock())
    
    def process_return(self, frame, arg):
        if self.time_calls:
            end = time.clock()
            start = self.currentlane.calltimes.pop()
            diff = " %.3fms" % ((end - start) * 1000)
        else:
            diff = ""
        
        retval = ""
        if arg is not None:
            retval = ": %s" % repr(arg)
        
        name = frame.f_code.co_name
        if name == "?":
            self.write_lane("< import %s (%s lines)" %
                            (frame.f_code.co_filename, frame.f_lineno))
        else:
            self.write_lane("< %s (%s:%s)%s%s" %
                            (name, scope_name(frame), frame.f_lineno,
                             retval, diff))
        self.currentlane.indent -= 1
    
    def process_c_call(self, frame, arg):
        self.currentlane.indent += 1
        
        # arg should be the called C function
        name = type(arg.__self__).__name__
        if name == 'NoneType':
            name = repr(arg)
        else:
            name = name + "." + arg.__name__
        
        self.write_lane("[ %s (%s:%s)" % (name, scope_name(frame),
                                          frame.f_lineno))
        if self.time_calls:
            self.currentlane.ccalltimes.append(time.clock())
    
    def process_c_return(self, frame, arg):
        if self.time_calls:
            end = time.clock()
            start = self.currentlane.ccalltimes.pop()
            diff = " %.3fms" % ((end - start) * 1000)
        else:
            diff = ""
        
        # arg should be the called C function
        name = type(arg.__self__).__name__
        if name == 'NoneType':
            name = repr(arg)
        else:
            name = name + "." + arg.__name__
        
        self.write_lane("] %s (%s:%s) %s" % (name, scope_name(frame),
                                             frame.f_lineno, diff))
        self.currentlane.indent -= 1
    
    def process_exception(self, frame, arg):
        exc_class = arg[0]
        if issubclass(exc_class, basestring):
            retval = repr(exc_class)
        else:
            retval = exc_class.__name__
        self.write_lane("E %s (%s:%s): %s" %
                        (frame.f_code.co_name, scope_name(frame),
                         frame.f_lineno, retval))
    
    def process_c_exception(self, frame, arg):
        # Note that Python exceptions will also (eventually)
        # call process_return, but C exceptions won't.
        # So we figure time and dedent here.
        if self.time_calls:
            end = time.clock()
            start = self.currentlane.ccalltimes.pop()
            diff = " %.3fms" % ((end - start) * 1000)
        else:
            diff = ""
        
        # arg should be the called C function
        self.write_lane("e %s (%s:%s) %s: %s" %
                        (frame.f_code.co_name, scope_name(frame),
                         frame.f_lineno, diff, arg))
        self.currentlane.indent -= 1
    
    def process_line(self, frame, arg):
        self.write_lane(". %s (%s:%s)" % (frame.f_code.co_name,
                                          scope_name(frame), frame.f_lineno))
    
    def process_threadstart(self, frame, arg):
        threadname = frame.f_locals['self'].getName()
        frame = frame.f_back
        self.write_lane("* %s (%s:%s) New Thread: %s"
                       % (frame.f_code.co_name, scope_name(frame),
                          frame.f_lineno, threadname))
    
    def process_threadstop(self, frame, arg):
        threadname = frame.f_locals['self'].getName()
        self.write_lane("X Thread terminated: %s" % threadname)
    
    #                               Watches                               #
    
    def watch(self, obj, attr, default=None, name=None):
        """Log the state of obj.attr each time it changes."""
        self.watches[(obj, attr)] = WatchedValue(obj, attr, default, name)
    
    def write_watches(self, changes_only=False):
        for w in self.watches.itervalues():
            newvalue = w.getvalue()
            if (not changes_only) or newvalue != w.oldvalue:
                w.oldvalue = newvalue
                self.write_lane("= %s: %s" % (w.name, repr(newvalue)))


def log(callback, filename, fileregex=".*", events=None):
    tr = Logger(fileregex, events=events)
    tr.out = open(filename, "wb")
    try:
        tr.start()
        callback()
    finally:
        tr.stop()
        tr.out.close()

def log_times(globpath, pattern=".*"):
    """Return a dict of matching (float(time in ms), line) pairs for each file."""
    import glob
    
    results = {}
    for fname in glob.glob(globpath):
        f = open(fname, 'rb')
        lines = []
        for line in f.readlines():
            line = line.rstrip()
            if line.endswith("ms") and re.search(pattern, line):
                line = line.rsplit(" ", 1)
                lines.append((float(line[1][:-2]), line[0]))
        results[fname] = lines
        f.close()
    return results

#                                Verifier                                 #

SLEEPING = 2

class Node:
    """A node on an execution tree, with references to children."""
    
    def __init__(self, data=None, exhausted=False, children=None):
        self.data = data
        self.exhausted = exhausted
        # The 'children' dict will eventually have one entry for each lane.
        self.children = children or {}
        self.result = None
    
    def __repr__(self):
        return "Node(%s, %s, %s, %s)" % (self.data, self.exhausted,
                                         self.children, self.result)


class Verifier(Tracer):
    """A tool for proving the thread-safety of closed, finite machines."""
    
    def __init__(self, fileregex="!(threading.py)", events=None):
        if events is None:
            events = non_c_events
        Tracer.__init__(self, fileregex, events)
        
        self.exhausted = False
        self.debug = False
    
    def start(self, caller=None):
        # Overridden because we don't need to set hooks on the main thread.
        self._using_profile_hook = False
        for e in c_events:
            if e in self.