1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
from aergia.aergia import Aergia
import yappi
import time
import unittest
class AergiaUnitTestCase(unittest.TestCase):
interval = 0.01
Aergia = Aergia(interval)
# yappi is a singleton
def setUp(self):
self.Aergia.clear()
yappi.stop()
yappi.clear_stats()
yappi.set_clock_type('wall')
yappi.set_context_id_callback(None)
yappi.set_context_name_callback(None)
yappi.set_tag_callback(None)
def assert_reasonable_delay(self, func_name, time_expected, samples):
'''Compares the results reported by Aergia for FUNC_NAME
with time_expected.'''
samples_actual = self.aergia_extract_values_by_func(samples, func_name)
time_actual = self.aergia_expected_time(samples_actual)
self.assert_roughly_equal(time_actual, time_expected)
def assert_similar_delay(self, func_name, yappi_stats, aergia_samples):
'''Compares the results reported by Aergia for FUNC_NAME
with yappi.'''
time_yappi = self.yappi_extract_values_by_func(yappi_stats, func_name)
samples_aergia = self.aergia_extract_values_by_func(
aergia_samples, func_name)
time_aergia = self.aergia_expected_time(samples_aergia)
self.assert_roughly_equal(time_aergia, time_yappi)
def assert_roughly_equal(self, v1, v2):
'''Throws an exception if values V1 and V2 are not within 0.035
seconds of each other. This number is influenced by instrumentation
overhead, sampling inconsistency, etc.'''
a = abs(v1 - v2)
self.assertTrue(a <= .035, f'{v1} (actual) not roughly {v2} (expected)')
def aergia_expected_time(self, total_samples):
'''Given TOTAL_SAMPLES, returns the total time, using the
sampling interval.'''
return total_samples * self.interval
def aergia_extract_values_by_func(self, samples, func_name):
'''Returns the total number of samples for function name, given an
Aergia SAMPLES object.'''
return sum(
value for key, value in samples.items()
if key.func == func_name
)
def yappi_extract_values_by_func(self, stats, func_name):
'''Returns the total number of samples for function name, given a
yappi stats object.'''
for s in stats:
if s.name == func_name:
return s.ttot
def burn_cpu(sec):
t0 = Aergia._gettime()
elapsed = 0
while (elapsed < sec):
for _ in range(1000):
pass
elapsed = Aergia._gettime() - t0
def burn_io(sec):
time.sleep(sec)
|
