-
Notifications
You must be signed in to change notification settings - Fork 19
/
stopwatch_global.py
61 lines (51 loc) · 2.27 KB
/
stopwatch_global.py
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
"""StopWatch - library for adding timers and tags in your code for performance monitoring
https://github.com/dropbox/stopwatch
This module provides an easy mechanism for using one-stopwatch-per-thread by storing
it at global scope.
For example:
```
with global_sw().timer('root'):
for i in range(50):
with global_sw().timer('inner_task'):
do_inner_task(i)
```
"""
import threading
from stopwatch import StopWatch
_GLOBAL_SW = None
class _GlobalSw(object):
"""A global store for thread-local stopwatches. Helps with the common case where
the caller only wants one stopwatch per thread.
"""
def __init__(self, time_func=None, export_aggregated_timers_func=None,
export_tracing_func=None, export_aggregated_timers_and_tracing_func=None):
self.threadlocal_sws = threading.local()
self.time_func = time_func
self.export_agg_timers_func = export_aggregated_timers_func
self.export_tracing_func = export_tracing_func
self.export_agg_timers_and_tracing_func = export_aggregated_timers_and_tracing_func
def global_sw(self):
"""Returns the thread local stopwatch (creating if it doesn't exists)"""
if not hasattr(self.threadlocal_sws, 'sw'):
self.threadlocal_sws.sw = StopWatch(
export_aggregated_timers_func=self.export_agg_timers_func,
time_func=self.time_func,
export_tracing_func=self.export_tracing_func,
export_aggregated_timers_and_tracing_func=self.export_agg_timers_and_tracing_func,
)
return self.threadlocal_sws.sw
def global_sw_init(*args, **kwargs):
"""Initialize global stopwatch with the completion callbacks"""
global _GLOBAL_SW
assert _GLOBAL_SW is None, "Cannot initialize global_sw twice"
_GLOBAL_SW = _GlobalSw(*args, **kwargs)
def global_sw_del():
"""Delete the global stopwatch. Typically not necessary, as stopwatch is reusable
but can useful for tests"""
global _GLOBAL_SW
assert _GLOBAL_SW is not None, "Cannot del global_sw since it was never initialized"
_GLOBAL_SW = None
def global_sw():
"""Return the stopwatch for the current thread"""
assert _GLOBAL_SW is not None, "Must initialize global_sw_init first"
return _GLOBAL_SW.global_sw()