File size: 1,433 Bytes
aac350d | 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 | """
Execution timing collector — tracks end-to-end job durations.
"""
from __future__ import annotations
import threading
from collections import defaultdict
from typing import Dict, List
class TimingCollector:
"""Tracks durations grouped by operation label."""
MAX_SAMPLES = 500
def __init__(self) -> None:
self._lock = threading.RLock()
self._samples: Dict[str, List[float]] = defaultdict(list)
def record(self, label: str, duration_ms: float) -> None:
with self._lock:
self._samples[label].append(duration_ms)
if len(self._samples[label]) > self.MAX_SAMPLES:
self._samples[label] = self._samples[label][-self.MAX_SAMPLES:]
def snapshot(self) -> dict:
with self._lock:
out = {}
for label, samples in self._samples.items():
if not samples:
continue
s = sorted(samples)
out[label] = {
"count": len(samples),
"avg_ms": round(sum(samples) / len(samples), 3),
"p50_ms": round(s[len(s) // 2], 3),
"p95_ms": round(s[int(len(s) * 0.95)], 3),
"min_ms": round(s[0], 3),
"max_ms": round(s[-1], 3),
}
return out
def reset(self) -> None:
with self._lock:
self._samples.clear()
|