1from omegaml.backends.tracking.simple import OmegaSimpleTracker
2
3
4class BackgroundProfiler:
5 """Profile CPU, Memory and Disk use in a background thread"""
6
7 def __init__(self, interval=10, callback=print):
8 self._stop = False
9 self._interval = interval
10 self._callback = callback
11 self._metrics = ['cpu', 'memory', 'disk']
12
13 def profile(self):
14 """
15 returns memory and cpu data as a dict
16 memory_load (float): percent of total memory used
17 memory_total (float): total memory in bytes
18 cpu_load (list): cpu load in percent, per cpu
19 cpu_count (int): logical cpus,
20 cpu_freq (float): MHz of current cpu frequency
21 cpu_avg (list): list of cpu load over last 1, 5, 15 minutes
22 disk_use (float): percent of disk used
23 disk_total (int): total size of disk, bytes
24 """
25 from datetime import datetime as dt
26
27 import psutil
28
29 p = psutil
30 disk = p.disk_usage('/')
31 cpu_count = p.cpu_count()
32 data = {'profile_dt': dt.utcnow()}
33 if 'memory' in self._metrics:
34 data.update(memory_load=p.virtual_memory().percent, memory_total=p.virtual_memory().total)
35 if 'cpu' in self._metrics:
36 data.update(
37 cpu_load=p.cpu_percent(percpu=True),
38 cpu_count=cpu_count,
39 cpu_freq=[f.current for f in p.cpu_freq(percpu=True)],
40 cpu_avg=[x / cpu_count for x in p.getloadavg()],
41 )
42 if 'disk' in self._metrics:
43 data.update(disk_use=disk.percent, disk_total=disk.total)
44 return data
45
46 @property
47 def interval(self):
48 return self._interval
49
50 @interval.setter
51 def interval(self, interval):
52 self._interval = interval
53 self.stop()
54 self.start()
55
56 @property
57 def metrics(self):
58 return self._metrics
59
60 @metrics.setter
61 def metrics(self, metrics):
62 self._metrics = metrics
63
64 def start(self):
65 """runs a background thread that reports stats every interval seconds
66
67 Every interval, calls callback(data), where data is the output of BackgroundProfiler.profile()
68 Stop by BackgroundProfiler.stop()
69 """
70 import atexit
71 from threading import Thread
72 from time import sleep
73
74 def runner():
75 cb = self._callback
76 try:
77 while not self._stop:
78 cb(self.profile())
79 sleep(self._interval)
80 except (KeyboardInterrupt, SystemExit):
81 pass
82
83 # handle exits by stopping the profiler
84 atexit.register(self.stop)
85 # start the profiler
86 self._stop = False
87 t = Thread(target=runner)
88 t.start()
89 # clean up
90 atexit.unregister(self.stop)
91
92 def stop(self):
93 self._stop = True
94
95
[docs]
96class OmegaProfilingTracker(OmegaSimpleTracker):
97 """A metric tracker that runs a system profiler while the experiment is active
98
99 Will record ``profile`` events that contain cpu, memory and disk profilings.
100 See BackgroundProfiler.profile() for details of the profiling metrics collected.
101
102 Usage:
103
104 To log metrics and system performance data::
105
106 with om.runtime.experiment('myexp', provider='profiling') as exp:
107 ...
108
109 data = exp.data(event='profile')
110
111 Properties::
112
113 exp.profiler.interval = n.m # interval of n.m seconds to profile, defaults to 3 seconds
114 exp.profiler.metrics = ['cpu', 'memory', 'disk'] # all or subset of metrics to collect
115 exp.max_buffer = n # number of items in buffer before tracking
116
117 Notes:
118 - the profiling data is buffered to reduce the number of database writes, by
119 default the data is written on every 6 profiling events (default: 6 * 10 = every 60 seconds)
120 - the step reported in the tracker counts the profiling event since the start, it is not
121 related to the step (epoch) reported by e.g. tensorflow
122 - For every step there is a ``event=profile``, ``key=profile_dt`` entry which you can use
123 to relate profiling events to a specific wall-clock time.
124 - It usually sufficient to report system metrics in intervals > 10 seconds since machine
125 learning algorithms tend to use CPU and memory over longer periods of time.
126 """
127
128 def __init__(self, *args, **kwargs):
129 super().__init__(*args, **kwargs)
130 self.profile_logs = []
131 self.max_buffer = 10
132
[docs]
133 def log_profile(self, data):
134 """the callback for BackgroundProfiler"""
135 self.profile_logs.append(data)
136 if len(self.profile_logs) >= (self.max_buffer or 1):
137 self.flush()
138
139 def flush(self):
140 super().flush()
141
142 def log_items():
143 for step, data in enumerate(self.profile_logs):
144 # record the actual time instead of logging time (avoid buffering delays)
145 dt = data.get('profile_dt')
146 for k, v in data.items():
147 item = self._common_log_data('profile', k, v, step=step, dt=dt)
148 yield item
149
150 if self.profile_logs:
151 # passing list of list, as_many=True => collection.insert_many() for speed
152 self._store.put(
153 [item for item in log_items()], self._data_name, index=['event'], as_many=True, noversion=True
154 )
155 self.profile_logs = []
156
157 def start_runtime(self):
158 self.profiler = BackgroundProfiler(callback=self.log_profile)
159 self.profiler.start()
160 super().start_runtime()
161
162 def stop_runtime(self):
163 self.profiler.stop()
164 self.flush()
165 super().stop_runtime()