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​

cpu_timer_name​

def cpu_timer_name(name: str) ‑> str:

Return the name of the CPU-time companion timer for phase name.

flush_all_timers​

def flush_all_timers() ‑> None:

Report the trailing window of every timer registered so far.

For a caller that owns the end of a run rather than the phases within it: it need not know which phases ran, and a phase that never ran contributes nothing. A timer whose window is empty is left alone, so this is safe to call at any boundary; call it at one a run passes through once, since every call cuts the windows of every timer short.

get_timer​

def get_timer(    name: str,    interval: int = 1000,    metric_interval: int | None = None,    report_after_seconds: float | None = None,    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.
  • metric_interval: Samples per window reported to Datadog. Applied only when the timer is first created.
  • report_after_seconds: Wall-clock floor on how long a phase may stay silent. Applied only when the timer is first created.
  • clock: Time source. Applied only when the timer is first created.

Returns The timer for name.

time_wall_and_cpu​

def time_wall_and_cpu(    name: str, interval: int = 1000,) ‑> collections.abc.Generator[None, None, None]:

Time one call on both the wall clock and the process CPU clock.

Elapsed time alone cannot say whether a phase is burning CPU or waiting on something - a decode and a slow network read look identical from the outside - and that distinction decides whether parallelising the phase can help at all. Recording both answers it: CPU time close to wall time means the work is compute, far below it means the phase is mostly waiting.

The wall figure lands in name and the CPU figure in name.cpu, so the two read as an ordinary pair of timers and need no special handling downstream.

time.process_time counts CPU across the whole process, so the CPU figure is only attributable to this phase while it is the only thing running - true of a phase timed on the main thread of a worker, not of one timed while a pool is busy. Compare the pair; do not read the CPU figure alone.

Arguments

  • name: Identifies the phase in the log output.
  • interval: Samples per reported window, for both timers.

timed_iter​

def timed_iter(    iterable: Iterable[_T],    name: str,    interval: int = 1000,    metric_interval: int | None = None,    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.
  • metric_interval: Samples per window reported to Datadog. Defaults to interval.
  • clock: Time source, in seconds. Injectable for testing.

Classes​

PeriodicTimingLogger​

class PeriodicTimingLogger(    name: str,    interval: int = 1000,    metric_interval: int | None = None,    report_after_seconds: float | None = None,    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.
  • metric_interval: Samples per window reported to Datadog. Defaults to interval. Set it higher for a phase logged once per call, where a metric per call would be a point per call with nothing aggregated.
  • report_after_seconds: Report a partial window once this long has passed since the last report, whichever comes first. A sample count alone cannot bound how long a phase stays silent: at seconds per call a window of fifty is minutes, and at minutes per call it is hours - which is how a step that is working looks like a step that has hung. A clock makes the reporting rate a function of elapsed time rather than of how much work there is, so the same setting behaves the same on a datasource of two hundred files and one of two hundred thousand. None leaves the timer purely count-driven and does not read the clock at all on the sample path, so a caller that injects a clock sees exactly the calls it saw before this existed.
  • 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.

  • total_samples : int - How many calls have been timed, across every window.

    A phase's sample count is often the interesting figure on its own: a timer that only fires on a fallback path measures how much of the population took that path, whatever the durations say.

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 each 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.