Skip to main content

timing_utils

Utilities for measuring how long frequently-repeated work takes.

Aimed at hot loops - iterating files, reading a cache, writing rows - where timing an individual call tells you nothing but the distribution over many calls tells you a lot, and where logging every call would drown the log.

Module​

Functions​

get_timer​

def get_timer(    name: str,    interval: int = 1000,    clock: Callable[[], float] = <built-in function perf_counter>,) ‑> PeriodicTimingLogger:

Return the timer registered under name, creating it on first use.

Works like logging.getLogger: callers in different scopes that name the same phase share one timer, so a phase measured across several functions does not need a timer threaded through them. Samples for one name are therefore pooled across every caller and every object using it - name the phase accordingly if that is not wanted.

Arguments

  • name: Identifies the phase, and keys the registry.
  • interval: Samples per reported window. Applied only when the timer is first created; ignored afterwards.
  • clock: Time source. Applied only when the timer is first created.

Returns The timer for name.

timed_iter​

def timed_iter(    iterable: Iterable[_T],    name: str,    interval: int = 1000,    clock: Callable[[], float] = <built-in function perf_counter>,) ‑> collections.abc.Generator[~_T, None, None]:

Yield from iterable, timing how long each item takes to arrive.

Measures retrieval rather than processing, which is what matters for a lazily produced sequence - enumerating a network share, streaming query results - where the consumer's own work would otherwise mask the cost of producing the next item.

Owns its timer and flushes it when iteration ends, including when the consumer abandons the iterator early, so callers have nothing to remember.

Arguments

  • iterable: The sequence to consume.
  • name: Identifies the phase in the log output.
  • interval: Samples per reported window.
  • clock: Time source, in seconds. Injectable for testing.

Classes​

PeriodicTimingLogger​

class PeriodicTimingLogger(    name: str,    interval: int = 1000,    clock: Callable[[], float] = <built-in function perf_counter>,):

Times repeated calls, logging a summary of each window of samples.

Emits one INFO line per interval samples giving the count, mean, standard deviation, minimum and maximum for that window, plus a running total. Statistics are per-window rather than cumulative so that a phase which degrades partway through a run - a network share slowing down, a cache growing - shows up as changing numbers rather than being averaged away.

Time individual calls with time(), and call flush() when the work is finished so that a trailing partial window is not discarded:

timer = PeriodicTimingLogger("cache.read") for batch in batches: with timer.time(): ... timer.flush()

Forgetting flush() loses only the final partial window, never a reported one. For iterators, prefer timed_iter, which owns the timer and flushes for you.

Thread-safe: the cost of the lock is negligible next to the work being measured.

Arguments

  • name: Identifies the phase in the log output.
  • interval: Samples per reported window.
  • clock: Time source, in seconds. Injectable for testing.

Variables​

  • cumulative_stats : tuple[float, float] | None - Mean and standard deviation, in seconds, over every sample so far.

    None until the first sample. Complements last_window_stats: the window shows how the phase is behaving now, this shows how it has behaved overall.

  • last_window_stats : tuple[float, float] | None - Mean and standard deviation, in seconds, of the last reported window.

    None until a window has been reported. Exposed so that callers can flag individual outliers against a recent baseline without keeping their own accumulators.

Methods​

flush​

def flush(self) ‑> None:

Report any samples not yet included in a window summary.

record​

def record(self, elapsed: float) ‑> None:

Add a sample, reporting and resetting the window when it is full.

time​

def time(self) ‑> collections.abc.Generator[None, None, None]:

Time one call, reporting a summary once a full window is collected.

A call that raises is still recorded - a phase that fails slowly is exactly what this is for - and the exception propagates unchanged.