pybootchartgui: add the original code

This is from:
http://pybootchartgui.googlecode.com/files/pybootchartgui-r124.tar.gz

Will modify it to make the build profiling in pictures.

Remove the examples since they would not work any more, and they cost
much disk space.

[YOCTO #2403]

(From OE-Core rev: 1f0791109e1aed715f02945834d6d7fdb9a411b4)

Signed-off-by: Robert Yang <liezhi.yang@windriver.com>
Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
This commit is contained in:
Robert Yang
2012-06-06 13:52:43 +08:00
committed by Richard Purdie
parent bc19f8bc9c
commit 3d78bc19c5
14 changed files with 2207 additions and 0 deletions

View File

@@ -0,0 +1,93 @@
import sys, os, re, struct, operator, math
from collections import defaultdict
import unittest
sys.path.insert(0, os.getcwd())
import parsing
debug = False
def floatEq(f1, f2):
return math.fabs(f1-f2) < 0.00001
class TestBCParser(unittest.TestCase):
def setUp(self):
self.name = "My first unittest"
self.rootdir = '../examples/1'
def mk_fname(self,f):
return os.path.join(self.rootdir, f)
def testParseHeader(self):
state = parsing.parse_file(parsing.ParserState(), self.mk_fname('header'))
self.assertEqual(6, len(state.headers))
self.assertEqual(2, parsing.get_num_cpus(state.headers))
def test_parseTimedBlocks(self):
state = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_diskstats.log'))
self.assertEqual(141, len(state.disk_stats))
def testParseProcPsLog(self):
state = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_ps.log'))
samples = state.ps_stats
processes = samples.process_list
sorted_processes = sorted(processes, key=lambda p: p.pid )
for index, line in enumerate(open(self.mk_fname('extract2.proc_ps.log'))):
tokens = line.split();
process = sorted_processes[index]
if debug:
print tokens[0:4]
print process.pid, process.cmd, process.ppid, len(process.samples)
print '-------------------'
self.assertEqual(tokens[0], str(process.pid))
self.assertEqual(tokens[1], str(process.cmd))
self.assertEqual(tokens[2], str(process.ppid))
self.assertEqual(tokens[3], str(len(process.samples)))
def testparseProcDiskStatLog(self):
state_with_headers = parsing.parse_file(parsing.ParserState(), self.mk_fname('header'))
state_with_headers.headers['system.cpu'] = 'xxx (2)'
samples = parsing.parse_file(state_with_headers, self.mk_fname('proc_diskstats.log')).disk_stats
self.assertEqual(141, len(samples))
for index, line in enumerate(open(self.mk_fname('extract.proc_diskstats.log'))):
tokens = line.split('\t')
sample = samples[index]
if debug:
print line.rstrip(),
print sample
print '-------------------'
self.assertEqual(tokens[0], str(sample.time))
self.assert_(floatEq(float(tokens[1]), sample.read))
self.assert_(floatEq(float(tokens[2]), sample.write))
self.assert_(floatEq(float(tokens[3]), sample.util))
def testparseProcStatLog(self):
samples = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_stat.log')).cpu_stats
self.assertEqual(141, len(samples))
for index, line in enumerate(open(self.mk_fname('extract.proc_stat.log'))):
tokens = line.split('\t')
sample = samples[index]
if debug:
print line.rstrip()
print sample
print '-------------------'
self.assert_(floatEq(float(tokens[0]), sample.time))
self.assert_(floatEq(float(tokens[1]), sample.user))
self.assert_(floatEq(float(tokens[2]), sample.sys))
self.assert_(floatEq(float(tokens[3]), sample.io))
def testParseLogDir(self):
res = parsing.parse([self.rootdir], False)
self.assertEqual(4, len(res))
if __name__ == '__main__':
unittest.main()

View File

@@ -0,0 +1,78 @@
import sys
import os
import unittest
sys.path.insert(0, os.getcwd())
import parsing
import process_tree
class TestProcessTree(unittest.TestCase):
def setUp(self):
self.name = "Process tree unittest"
self.rootdir = '../examples/1'
self.ps_stats = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_ps.log')).ps_stats
self.processtree = process_tree.ProcessTree(self.ps_stats, None, False, for_testing = True)
def mk_fname(self,f):
return os.path.join(self.rootdir, f)
def flatten(self, process_tree):
flattened = []
for p in process_tree:
flattened.append(p)
flattened.extend(self.flatten(p.child_list))
return flattened
def checkAgainstJavaExtract(self, filename, process_tree):
for expected, actual in zip(open(filename), self.flatten(process_tree)):
tokens = expected.split('\t')
self.assertEqual(int(tokens[0]), actual.pid)
self.assertEqual(tokens[1], actual.cmd)
self.assertEqual(long(tokens[2]), 10 * actual.start_time)
self.assert_(long(tokens[3]) - 10 * actual.duration < 5, "duration")
self.assertEqual(int(tokens[4]), len(actual.child_list))
self.assertEqual(int(tokens[5]), len(actual.samples))
def testBuild(self):
process_tree = self.processtree.process_tree
self.checkAgainstJavaExtract(self.mk_fname('extract.processtree.1.log'), process_tree)
def testMergeLogger(self):
self.processtree.merge_logger(self.processtree.process_tree, 'bootchartd', None, False)
process_tree = self.processtree.process_tree
self.checkAgainstJavaExtract(self.mk_fname('extract.processtree.2.log'), process_tree)
def testPrune(self):
self.processtree.merge_logger(self.processtree.process_tree, 'bootchartd', None, False)
self.processtree.prune(self.processtree.process_tree, None)
process_tree = self.processtree.process_tree
self.checkAgainstJavaExtract(self.mk_fname('extract.processtree.3b.log'), process_tree)
def testMergeExploders(self):
self.processtree.merge_logger(self.processtree.process_tree, 'bootchartd', None, False)
self.processtree.prune(self.processtree.process_tree, None)
self.processtree.merge_exploders(self.processtree.process_tree, set(['hwup']))
process_tree = self.processtree.process_tree
self.checkAgainstJavaExtract(self.mk_fname('extract.processtree.3c.log'), process_tree)
def testMergeSiblings(self):
self.processtree.merge_logger(self.processtree.process_tree, 'bootchartd', None, False)
self.processtree.prune(self.processtree.process_tree, None)
self.processtree.merge_exploders(self.processtree.process_tree, set(['hwup']))
self.processtree.merge_siblings(self.processtree.process_tree)
process_tree = self.processtree.process_tree
self.checkAgainstJavaExtract(self.mk_fname('extract.processtree.3d.log'), process_tree)
def testMergeRuns(self):
self.processtree.merge_logger(self.processtree.process_tree, 'bootchartd', None, False)
self.processtree.prune(self.processtree.process_tree, None)
self.processtree.merge_exploders(self.processtree.process_tree, set(['hwup']))
self.processtree.merge_siblings(self.processtree.process_tree)
self.processtree.merge_runs(self.processtree.process_tree)
process_tree = self.processtree.process_tree
self.checkAgainstJavaExtract(self.mk_fname('extract.processtree.3e.log'), process_tree)
if __name__ == '__main__':
unittest.main()