performance.py

#!/usr/bin/env python

# ----------------------------------------------------------------------
# Be sure to add the python path that points to the LLDB shared library.
# On MacOSX csh, tcsh:
# setenv PYTHONPATH /Applications/Xcode.app/Contents/SharedFrameworks/LLDB.framework/Resources/Python
# On MacOSX sh, bash:
# export PYTHONPATH=/Applications/Xcode.app/Contents/SharedFrameworks/LLDB.framework/Resources/Python
# ----------------------------------------------------------------------

importoptparse
importos
importplatform
importre
importresource
importsys
importsubprocess
importtime
importtypes

# ----------------------------------------------------------------------
# Code that auto imports LLDB
# ----------------------------------------------------------------------
try:
# Just try for LLDB in case PYTHONPATH is already correctly setup
importlldb
exceptImportError:
lldb_python_dirs=list()
# lldb is not in the PYTHONPATH, try some defaults for the current platform
platform_system=platform.system()
ifplatform_system=="Darwin":
# On Darwin, try the currently selected Xcode directory
xcode_dir=subprocess.check_output("xcode-select --print-path", shell=True)
ifxcode_dir:
lldb_python_dirs.append(
os.path.realpath(
xcode_dir+"/../SharedFrameworks/LLDB.framework/Resources/Python"
 )
 )
lldb_python_dirs.append(
xcode_dir+"/Library/PrivateFrameworks/LLDB.framework/Resources/Python"
 )
lldb_python_dirs.append(
"/System/Library/PrivateFrameworks/LLDB.framework/Resources/Python"
 )
success=False
forlldb_python_dirinlldb_python_dirs:
ifos.path.exists(lldb_python_dir):
ifnot (sys.path.__contains__(lldb_python_dir)):
sys.path.append(lldb_python_dir)
try:
importlldb
exceptImportError:
pass
else:
print('imported lldb from: "%s"'% (lldb_python_dir))
success=True
break
ifnotsuccess:
print(
"error: couldn't locate the 'lldb' module, please set PYTHONPATH correctly"
 )
sys.exit(1)


classTimer:
def__enter__(self):
self.start=time.clock()
returnself

def__exit__(self, *args):
self.end=time.clock()
self.interval=self.end-self.start


classAction(object):
"""Class that encapsulates actions to take when a thread stops for a reason."""

def__init__(self, callback=None, callback_owner=None):
self.callback=callback
self.callback_owner=callback_owner

defThreadStopped(self, thread):
assert (
False
 ), "performance.Action.ThreadStopped(self, thread) must be overridden in a subclass"


classPlanCompleteAction(Action):
def__init__(self, callback=None, callback_owner=None):
Action.__init__(self, callback, callback_owner)

defThreadStopped(self, thread):
ifthread.GetStopReason() ==lldb.eStopReasonPlanComplete:
ifself.callback:
ifself.callback_owner:
self.callback(self.callback_owner, thread)
else:
self.callback(thread)
returnTrue
returnFalse


classBreakpointAction(Action):
def__init__(
self,
callback=None,
callback_owner=None,
name=None,
module=None,
file=None,
line=None,
breakpoint=None,
 ):
Action.__init__(self, callback, callback_owner)
self.modules=lldb.SBFileSpecList()
self.files=lldb.SBFileSpecList()
self.breakpoints=list()
# "module" can be a list or a string
ifbreakpoint:
self.breakpoints.append(breakpoint)
else:
ifmodule:
ifisinstance(module, types.ListType):
formodule_pathinmodule:
self.modules.Append(lldb.SBFileSpec(module_path, False))
elifisinstance(module, types.StringTypes):
self.modules.Append(lldb.SBFileSpec(module, False))
ifname:
# "file" can be a list or a string
iffile:
ifisinstance(file, types.ListType):
self.files=lldb.SBFileSpecList()
forfinfile:
self.files.Append(lldb.SBFileSpec(f, False))
elifisinstance(file, types.StringTypes):
self.files.Append(lldb.SBFileSpec(file, False))
self.breakpoints.append(
self.target.BreakpointCreateByName(name, self.modules, self.files)
 )
eliffileandline:
self.breakpoints.append(
self.target.BreakpointCreateByLocation(file, line)
 )

defThreadStopped(self, thread):
ifthread.GetStopReason() ==lldb.eStopReasonBreakpoint:
forbpinself.breakpoints:
ifbp.GetID() ==thread.GetStopReasonDataAtIndex(0):
ifself.callback:
ifself.callback_owner:
self.callback(self.callback_owner, thread)
else:
self.callback(thread)
returnTrue
returnFalse


classTestCase:
"""Class that aids in running performance tests."""

def__init__(self):
self.verbose=False
self.debugger=lldb.SBDebugger.Create()
self.target=None
self.process=None
self.thread=None
self.launch_info=None
self.done=False
self.listener=self.debugger.GetListener()
self.user_actions=list()
self.builtin_actions=list()
self.bp_id_to_dict=dict()

defSetup(self, args):
self.launch_info=lldb.SBLaunchInfo(args)

defRun(self, args):
assertFalse, "performance.TestCase.Run(self, args) must be subclassed"

defLaunch(self):
ifself.target:
error=lldb.SBError()
self.process=self.target.Launch(self.launch_info, error)
ifnoterror.Success():
print("error: %s"%error.GetCString())
ifself.process:
self.process.GetBroadcaster().AddListener(
self.listener,
lldb.SBProcess.eBroadcastBitStateChanged
|lldb.SBProcess.eBroadcastBitInterrupt,
 )
returnTrue
returnFalse

defWaitForNextProcessEvent(self):
event=None
ifself.process:
whileeventisNone:
process_event=lldb.SBEvent()
ifself.listener.WaitForEvent(lldb.UINT32_MAX, process_event):
state=lldb.SBProcess.GetStateFromEvent(process_event)
ifself.verbose:
print("event = %s"% (lldb.SBDebugger.StateAsCString(state)))
iflldb.SBProcess.GetRestartedFromEvent(process_event):
continue
if (
state==lldb.eStateInvalid
orstate==lldb.eStateDetached
orstate==lldb.eStateCrashed
orstate==lldb.eStateUnloaded
orstate==lldb.eStateExited
 ):
event=process_event
self.done=True
elif (
state==lldb.eStateConnected
orstate==lldb.eStateAttaching
orstate==lldb.eStateLaunching
orstate==lldb.eStateRunning
orstate==lldb.eStateStepping
orstate==lldb.eStateSuspended
 ):
continue
elifstate==lldb.eStateStopped:
event=process_event
call_test_step=True
fatal=False
selected_thread=False
forthreadinself.process:
frame=thread.GetFrameAtIndex(0)
select_thread=False

stop_reason=thread.GetStopReason()
ifself.verbose:
print(
"tid = %#x pc = %#x "
% (thread.GetThreadID(), frame.GetPC()),
end=" ",
 )
ifstop_reason==lldb.eStopReasonNone:
ifself.verbose:
print("none")
elifstop_reason==lldb.eStopReasonTrace:
select_thread=True
ifself.verbose:
print("trace")
elifstop_reason==lldb.eStopReasonPlanComplete:
select_thread=True
ifself.verbose:
print("plan complete")
elifstop_reason==lldb.eStopReasonThreadExiting:
ifself.verbose:
print("thread exiting")
elifstop_reason==lldb.eStopReasonExec:
ifself.verbose:
print("exec")
elifstop_reason==lldb.eStopReasonInvalid:
ifself.verbose:
print("invalid")
elifstop_reason==lldb.eStopReasonException:
select_thread=True
ifself.verbose:
print("exception")
fatal=True
elifstop_reason==lldb.eStopReasonBreakpoint:
select_thread=True
bp_id=thread.GetStopReasonDataAtIndex(0)
bp_loc_id=thread.GetStopReasonDataAtIndex(1)
ifself.verbose:
print("breakpoint id = %d.%d"% (bp_id, bp_loc_id))
elifstop_reason==lldb.eStopReasonWatchpoint:
select_thread=True
ifself.verbose:
print(
"watchpoint id = %d"
% (thread.GetStopReasonDataAtIndex(0))
 )
elifstop_reason==lldb.eStopReasonSignal:
select_thread=True
ifself.verbose:
print(
"signal %d"
% (thread.GetStopReasonDataAtIndex(0))
 )
elifstop_reason==lldb.eStopReasonFork:
ifself.verbose:
print(
"fork pid = %d"
% (thread.GetStopReasonDataAtIndex(0))
 )
elifstop_reason==lldb.eStopReasonVFork:
ifself.verbose:
print(
"vfork pid = %d"
% (thread.GetStopReasonDataAtIndex(0))
 )
elifstop_reason==lldb.eStopReasonVForkDone:
ifself.verbose:
print("vfork done")

ifselect_threadandnotselected_thread:
self.thread=thread
selected_thread=self.process.SetSelectedThread(thread)

foractioninself.user_actions:
action.ThreadStopped(thread)

iffatal:
# if self.verbose:
# Xcode.RunCommand(self.debugger,"bt all",true)
sys.exit(1)
returnevent


classMeasurement:
"""A class that encapsulates a measurement"""

def__init__(self):
object.__init__(self)

defMeasure(self):
assertFalse, "performance.Measurement.Measure() must be subclassed"


classMemoryMeasurement(Measurement):
"""A class that can measure memory statistics for a process."""

def__init__(self, pid):
Measurement.__init__(self)
self.pid=pid
self.stats= [
"rprvt",
"rshrd",
"rsize",
"vsize",
"vprvt",
"kprvt",
"kshrd",
"faults",
"cow",
"pageins",
 ]
self.command="top -l 1 -pid %u -stats %s"% (self.pid, ",".join(self.stats))
self.value=dict()

defMeasure(self):
output=subprocess.getoutput(self.command).split("\n")[-1]
values=re.split(r"[-+\s]+", output)
foridx, statinenumerate(values):
multiplier=1
ifstat:
ifstat[-1] =="K":
multiplier=1024
stat=stat[:-1]
elifstat[-1] =="M":
multiplier=1024*1024
stat=stat[:-1]
elifstat[-1] =="G":
multiplier=1024*1024*1024
elifstat[-1] =="T":
multiplier=1024*1024*1024*1024
stat=stat[:-1]
self.value[self.stats[idx]] =int(stat) *multiplier

def__str__(self):
"""Dump the MemoryMeasurement current value"""
s=""
forkeyinself.value.keys():
ifs:
s+="\n"
s+="%8s = %s"% (key, self.value[key])
returns


classTesterTestCase(TestCase):
def__init__(self):
TestCase.__init__(self)
self.verbose=True
self.num_steps=5

defBreakpointHit(self, thread):
bp_id=thread.GetStopReasonDataAtIndex(0)
loc_id=thread.GetStopReasonDataAtIndex(1)
print(
"Breakpoint %i.%i hit: %s"
% (bp_id, loc_id, thread.process.target.FindBreakpointByID(bp_id))
 )
thread.StepOver()

defPlanComplete(self, thread):
ifself.num_steps>0:
thread.StepOver()
self.num_steps=self.num_steps-1
else:
thread.process.Kill()

defRun(self, args):
self.Setup(args)
withTimer() astotal_time:
self.target=self.debugger.CreateTarget(args[0])
ifself.target:
withTimer() asbreakpoint_timer:
bp=self.target.BreakpointCreateByName("main")
print("Breakpoint time = %.03f sec."%breakpoint_timer.interval)

self.user_actions.append(
BreakpointAction(
breakpoint=bp,
callback=TesterTestCase.BreakpointHit,
callback_owner=self,
 )
 )
self.user_actions.append(
PlanCompleteAction(
callback=TesterTestCase.PlanComplete, callback_owner=self
 )
 )

ifself.Launch():
whilenotself.done:
self.WaitForNextProcessEvent()
else:
print("error: failed to launch process")
else:
print("error: failed to create target with '%s'"% (args[0]))
print("Total time = %.03f sec."%total_time.interval)


if__name__=="__main__":
lldb.SBDebugger.Initialize()
test=TesterTestCase()
test.Run(sys.argv[1:])
mem=MemoryMeasurement(os.getpid())
mem.Measure()
print(str(mem))
lldb.SBDebugger.Terminate()
# print "sleeeping for 100 seconds"
# time.sleep(100)