Stop monitoring and return collected metrics.
(self)
| 165 | self.initial_memory_trace = tracemalloc.get_traced_memory()[0] |
| 166 | |
| 167 | def stop_monitoring(self) -> BenchmarkMetrics: |
| 168 | """Stop monitoring and return collected metrics.""" |
| 169 | end_time = time.perf_counter() |
| 170 | end_memory = self.process.memory_info().rss / 1024 / 1024 # MB |
| 171 | end_cpu_times = self.process.cpu_times() |
| 172 | |
| 173 | current_memory, peak_memory = tracemalloc.get_traced_memory() |
| 174 | tracemalloc.stop() |
| 175 | |
| 176 | # Calculate metrics |
| 177 | execution_time_ms = (end_time - self.start_time) * 1000 |
| 178 | memory_usage_mb = end_memory - self.start_memory |
| 179 | |
| 180 | # Calculate CPU usage (approximate) |
| 181 | cpu_time_used = (end_cpu_times.user - self.start_cpu_times.user) + (end_cpu_times.system - self.start_cpu_times.system) |
| 182 | wall_time = end_time - self.start_time |
| 183 | cpu_usage_percent = (cpu_time_used / wall_time) * 100 if wall_time > 0 else 0 |
| 184 | |
| 185 | return BenchmarkMetrics( |
| 186 | execution_time_ms=execution_time_ms, |
| 187 | memory_usage_mb=memory_usage_mb, |
| 188 | cpu_usage_percent=cpu_usage_percent, |
| 189 | latency_ms=execution_time_ms, # Same as execution time for single tasks |
| 190 | metadata={ |
| 191 | "peak_memory_mb": peak_memory / 1024 / 1024, |
| 192 | "memory_delta_mb": (current_memory - self.initial_memory_trace) / 1024 / 1024, |
| 193 | }, |
| 194 | ) |
| 195 | |
| 196 | |
| 197 | class BaseBenchmark(ABC): |
no test coverage detected