Class Timer

python/ray/data/_internal/stats.py:163–279

Helper class for tracking accumulated time (in seconds). Every value passed to :meth:`add` is also fed into an internal :class:`DistributionTracker` (a KLL sketch with bounded memory) so :meth:`percentile` can return an approximate p-th percentile at any time. The sketch uses O(k lo

class Timer:
    """Helper class for tracking accumulated time (in seconds).

    Every value passed to :meth:`add` is also fed into an internal
    :class:`DistributionTracker` (a KLL sketch with bounded memory) so
    :meth:`percentile` can return an approximate p-th percentile at any
    time. The sketch uses O(k log(n/k)) memory (k=200 by default), so it
    stays a few kilobytes regardless of how many samples are added —
    safe for long-running production jobs.

    Percentile accuracy is the KLL guarantee — roughly 1.65% rank error
    at the default k=200. When the optional ``datasketches`` dependency
    is not installed, :meth:`percentile` returns 0 (the other stats are
    unaffected).
    """

    def __init__(self):
        self._total: float = 0
        self._min: float = float("inf")
        self._max: float = 0
        self._total_count: float = 0
        # Bounded-memory percentile backend. add() forwards every value
        # to ``add_sample`` and ``percentile`` reads from it.
        self._distribution: DistributionTracker = DistributionTracker()

    @contextmanager
    def timer(self) -> None:
        time_start = time.perf_counter()
        try:
            yield
        finally:
            self.add(time.perf_counter() - time_start)

    def add(self, value: float) -> None:
        self._total += value
        if value < self._min:
            self._min = value
        if value > self._max:
            self._max = value
        self._total_count += 1
        self._distribution.add_sample(value)

    def get(self) -> float:
        return self._total

    def min(self) -> float:
        return self._min

    def max(self) -> float:
        return self._max

    def avg(self) -> float:
        return self._total / self._total_count if self._total_count else float("inf")

    def percentile(self, p: float) -> float:
        """Approximate ``p``-th percentile in seconds.

        Backed by the internal :class:`DistributionTracker`&#x27;s KLL