climate_ref_core.resources

Measurement of the resources a block of work consumes.

A block can be measured via a context manager, :func:measure_resources, and results are returned as a frozen dataclass, :class:ResourceUsage.

.. code-block:: python

with measure_resources() as recorder:
    run_diagnostic()

usage = recorder.usage
print(usage.wall_seconds, usage.peak_memory_bytes, usage.memory_source)

Measurement never raises. Any probe that fails degrades a single field to None, or degrades :attr:ResourceUsage.memory_source to "unavailable".

Peak memory comes from a summed sweep of the process tree by default. proc_tree observes this block's processes and nothing else. A cgroup reading covers every process in the container, so it only describes this block when the caller declares, via cgroup_exclusive, that nothing else shares the cgroup. The fallbacks, in order, are a cgroup reading and then getrusage. :attr:ResourceUsage.memory_source records which one won, because a getrusage figure must never be silently compared against a cgroup figure.

Both sampled peaks are always recorded, in :attr:ResourceUsage.cgroup_peak_bytes and :attr:ResourceUsage.proc_tree_peak_bytes, whichever of them was reported. A large divergence between the two is itself the evidence that the cgroup was shared.

CGROUP_V2_MOUNT = Path('/sys/fs/cgroup') module-attribute

Mount point of the cgroup v2 unified hierarchy.

MemorySource = Literal['cgroup', 'proc_tree', 'rusage', 'unavailable'] module-attribute

Provenance of a peak memory measurement.

ResourceRecorder

Handle yielded by :func:measure_resources.

:attr:usage is None while the block runs and holds a :class:ResourceUsage once it exits.

The priority for the method of determining memory usage is:

cgroup (exclusive) > proc_tree > cgroup > rusage

In practice this generally means proc_tree for most use cases (LocalExecutor or Celery under MacOS/Linux).

Source code in packages/climate-ref-core/src/climate_ref_core/resources.py

class ResourceRecorder:
    """
    Handle yielded by :func:`measure_resources`.

    :attr:`usage` is None while the block runs and holds a :class:`ResourceUsage` once it exits.

    The priority for the method of determining memory usage is:

    cgroup (exclusive) > proc_tree > cgroup > rusage

    In practice this generally means ``proc_tree`` for most use cases
    (LocalExecutor or Celery under MacOS/Linux).
    """

    def __init__(self, interval: float, cgroup_exclusive: bool = False) -> None:
        self.usage: ResourceUsage | None = None
        self._interval = interval
        self._cgroup_exclusive = cgroup_exclusive
        self._exclusive = cgroup_exclusive
        self._cgroup: Path | None = None
        self._sampler: _PeakSampler | None = None
        self._wall_start = 0.0
        self._cpu_start: float | None = None
        self._cgroup_peak_at_entry: int | None = None

    def _start(self) -> None:
        """Begin measuring."""
        with _registry_lock:
            for other in _in_flight:
                other._exclusive = False
            self._exclusive = self._cgroup_exclusive and not _in_flight
            _in_flight.append(self)

        self._wall_start = time.monotonic()
        try:
            self._start_probes()
        except Exception:
            return

    def _start_probes(self) -> None:
        """Take the entry readings and start the sampler."""
        self._cgroup = _cgroup_directory()
        if self._cgroup is not None:
            self._cgroup_peak_at_entry = _read_cgroup_int(self._cgroup / "memory.peak")

        self._cpu_start = _cpu_seconds()

        if self._cgroup is not None or _psutil is not None:
            self._sampler = _PeakSampler(self._cgroup, self._interval)
            self._sampler.start()

    def _finish(self) -> None:
        """Stop measuring and populate :attr:`usage`."""
        wall_seconds = max(0.0, time.monotonic() - self._wall_start)
        try:
            self._stop_sampler()
        finally:
            self._deregister()

        self.usage = self._build_usage(wall_seconds)

    def _deregister(self) -> None:
        """
        Drop this recorder from the in-flight registry.

        Safe to call more than once.
        The registry is process-global state behind ``exclusive``,
        so a leaked entry would mark every later measurement as non-exclusive
        and quietly remove it from aggregation.
        """
        with _registry_lock:
            if self in _in_flight:
                _in_flight.remove(self)

    def _stop_sampler(self) -> None:
        """
        Ask the sampler to finish, waiting only for a bounded time.

        Safe to call more than once.
        A sampler left running would keep sweeping the process for the life of the process,
        charging its CPU cost to whatever runs next.
        """
        if self._sampler is not None:
            self._sampler.stop()
            self._sampler.join(timeout=_SAMPLER_JOIN_TIMEOUT)

    def _build_usage(self, wall_seconds: float) -> ResourceUsage:
        """Assemble the record from the exit readings."""
        cpu_seconds = _safe(self._elapsed_cpu, None)
        sampler = self._sampler
        rusage_peak = _safe(_rusage_peak_bytes, None)
        cgroup_peak = _safe(lambda: self._cgroup_peak(sampler), None)
        proc_tree_peak = sampler.proc_tree_peak if sampler is not None else None
        unmeasured: tuple[int | None, MemorySource] = (None, "unavailable")
        peak, source = _safe(lambda: self._resolve_peak(cgroup_peak, proc_tree_peak, rusage_peak), unmeasured)

        return ResourceUsage(
            wall_seconds=wall_seconds,
            cpu_seconds=cpu_seconds,
            peak_memory_bytes=peak,
            memory_source=source,
            cgroup_peak_bytes=cgroup_peak,
            proc_tree_peak_bytes=proc_tree_peak,
            memory_limit_bytes=_safe(self._memory_limit, None),
            cpu_limit=_safe(self._cpu_limit, None),
            exclusive=self._exclusive,
            context={
                "host": _safe(_hostname, None),
                "cpu_count": _safe(os.cpu_count, None),
                "sample_interval": self._interval,
                "samples": sampler.samples if sampler is not None else 0,
                "cgroup": str(self._cgroup) if self._cgroup is not None else None,
                "cgroup_peak_at_entry": self._cgroup_peak_at_entry,
                "cgroup_exclusive_declared": self._cgroup_exclusive,
                "cgroup_peak_bytes": cgroup_peak,
                "proc_tree_peak_bytes": proc_tree_peak,
                "psutil_available": _psutil is not None,
                "rusage_peak_bytes": rusage_peak,
                "rusage_is_process_lifetime_peak": True,
            },
        )

    def _elapsed_cpu(self) -> float | None:
        """CPU seconds used inside the block, or None when the counters are unreadable."""
        end = _cpu_seconds()
        if self._cpu_start is None or end is None:
            return None
        return max(0.0, end - self._cpu_start)

    def _cgroup_peak(self, sampler: _PeakSampler | None) -> int | None:
        """
        Best cgroup figure for this block, or None when the group could not be read.

        The group high-water mark when this block raised it,
        otherwise the largest sampled ``memory.current``.
        """
        if self._cgroup is not None:
            peak = _read_cgroup_int(self._cgroup / "memory.peak")
            entry = self._cgroup_peak_at_entry
            # memory.peak is a high-water mark for the whole group,
            # so it only describes this block when the block pushed it higher.
            # Without the entry reading there is nothing to subtract the group's history from,
            # and the sampled series below is a measurement of this block rather than a guess.
            if peak is not None and entry is not None and peak > entry:
                return peak

        return sampler.cgroup_peak if sampler is not None else None

    def _resolve_peak(
        self, cgroup_peak: int | None, proc_tree_peak: int | None, rusage_peak: int | None
    ) -> tuple[int | None, MemorySource]:
        """
        Pick the peak to report, and name the source it came from.

        The cgroup wins only when this block had the cgroup to itself,
        because otherwise it measures the container rather than the block.
        The process tree is the default because it sweeps this process and its descendants and nothing else.

        A shared cgroup reading is still preferred over ``getrusage``,
        which cannot be attributed to a block at all.
        """
        if self._exclusive and cgroup_peak is not None:
            return cgroup_peak, "cgroup"

        if proc_tree_peak is not None:
            return proc_tree_peak, "proc_tree"

        if cgroup_peak is not None:
            # Names the container, not this block, which ``exclusive`` being False records.
            return cgroup_peak, "cgroup"

        if rusage_peak is not None:
            # A lifetime high-water mark rather than a measurement of this block,
            # which is what "rusage" in memory_source warns the reader about.
            return rusage_peak, "rusage"

        return None, "unavailable"

    def _memory_limit(self) -> int | None:
        """Read the cgroup ``memory.max`` limit in bytes, or None when unlimited or unavailable."""
        if self._cgroup is None:
            return None
        return _read_cgroup_int(self._cgroup / "memory.max")

    def _cpu_limit(self) -> float | None:
        """Read the cgroup ``cpu.max`` quota in cores, or None when unlimited or unavailable."""
        if self._cgroup is None:
            return None
        return _read_cgroup_cpu_limit(self._cgroup / "cpu.max")

ResourceUsage

What one block of work cost.

Every field except :attr:wall_seconds and :attr:exclusive is nullable, because each of them comes from a probe that a given host may not answer.

Source code in packages/climate-ref-core/src/climate_ref_core/resources.py

@frozen
class ResourceUsage:
    """
    What one block of work cost.

    Every field except :attr:`wall_seconds` and :attr:`exclusive` is nullable,
    because each of them comes from a probe that a given host may not answer.
    """

    wall_seconds: float
    """Elapsed monotonic time."""

    cpu_seconds: float | None
    """CPU time used, self plus children, user plus system."""

    peak_memory_bytes: int | None
    """Peak memory, measured by whichever source :attr:`memory_source` names."""

    memory_source: MemorySource
    """Provenance of :attr:`peak_memory_bytes`.

    Two numbers are only comparable when they share a source.
    """

    cgroup_peak_bytes: int | None
    """Peak memory of the whole cgroup, whether or not it was the source that won.

    None when this is not a cgroup v2 host or the control files could not be read.
    """

    proc_tree_peak_bytes: int | None
    """Peak summed resident memory of this process and its descendants, whether or not it won.

    None when psutil is missing or every sweep failed.
    """

    memory_limit_bytes: int | None
    """cgroup ``memory.max`` at run time, or None when the group is unlimited."""

    cpu_limit: float | None
    """cgroup ``cpu.max`` quota expressed in cores, or None when the group is unlimited."""

    exclusive: bool
    """Whether the cgroup readings are attributable to this block alone.

    True only when the caller declared the cgroup exclusive
    *and* no other measured block overlapped this one in this process.
    Sibling worker processes saturating the same container are invisible from here,
    so without it a cgroup figure describes the container rather than this block.
    """

    context: dict[str, Any]
    """Host and sampler detail, JSON serialisable."""

cgroup_peak_bytes instance-attribute

Peak memory of the whole cgroup, whether or not it was the source that won.

None when this is not a cgroup v2 host or the control files could not be read.

context instance-attribute

Host and sampler detail, JSON serialisable.

cpu_limit instance-attribute

cgroup cpu.max quota expressed in cores, or None when the group is unlimited.

cpu_seconds instance-attribute

CPU time used, self plus children, user plus system.

exclusive instance-attribute

Whether the cgroup readings are attributable to this block alone.

True only when the caller declared the cgroup exclusive and no other measured block overlapped this one in this process. Sibling worker processes saturating the same container are invisible from here, so without it a cgroup figure describes the container rather than this block.

memory_limit_bytes instance-attribute

cgroup memory.max at run time, or None when the group is unlimited.

memory_source instance-attribute

Provenance of :attr:peak_memory_bytes.

Two numbers are only comparable when they share a source.

peak_memory_bytes instance-attribute

Peak memory, measured by whichever source :attr:memory_source names.

proc_tree_peak_bytes instance-attribute

Peak summed resident memory of this process and its descendants, whether or not it won.

None when psutil is missing or every sweep failed.

wall_seconds instance-attribute

Elapsed monotonic time.

measure_resources(*, interval=0.5, enabled=True, cgroup_exclusive=False)

Measure wall time, CPU time and peak memory of everything done in the block.

The yielded recorder exposes a single attribute, usage, which is None inside the block and a :class:ResourceUsage after it exits.

A sampling failure degrades individual fields to None rather than failing the execution. An exception raised inside the block still propagates, with usage populated.

Parameters:

Name	Type	Description	Default
interval	float	Seconds between memory samples.	0.5
enabled	bool	Whether to measure at all. When False the block runs untouched and usage stays None, which every consumer already reads as unmeasured. No sampler thread is started and no cgroup file is read.	True
cgroup_exclusive	bool	Whether the caller can promise that nothing else shares this process's cgroup while the block runs. Only a caller that owns the concurrency knows this. It defaults to False, under which a cgroup figure is reported only when the process tree cannot be swept, and is marked as non-exclusive so aggregation excludes it.	False

Yields:

Type	Description
ResourceRecorder	The recorder holding the result.

Source code in packages/climate-ref-core/src/climate_ref_core/resources.py

@contextmanager
def measure_resources(
    *, interval: float = 0.5, enabled: bool = True, cgroup_exclusive: bool = False
) -> Iterator[ResourceRecorder]:
    """
    Measure wall time, CPU time and peak memory of everything done in the block.

    The yielded recorder exposes a single attribute,
    ``usage``, which is None inside the block and a :class:`ResourceUsage` after it exits.

    A sampling failure degrades individual fields to None rather than failing the execution.
    An exception raised inside the block still propagates, with ``usage`` populated.

    Parameters
    ----------
    interval
        Seconds between memory samples.
    enabled
        Whether to measure at all.

        When False the block runs untouched and ``usage`` stays None,
        which every consumer already reads as unmeasured.
        No sampler thread is started and no cgroup file is read.
    cgroup_exclusive
        Whether the caller can promise that nothing else shares this process's cgroup while the block runs.

        Only a caller that owns the concurrency knows this.
        It defaults to False,
        under which a cgroup figure is reported only when the process tree cannot be swept,
        and is marked as non-exclusive so aggregation excludes it.

    Yields
    ------
    :
        The recorder holding the result.
    """
    recorder = ResourceRecorder(interval, cgroup_exclusive)
    if not enabled:
        yield recorder
        return

    recorder._start()
    try:
        yield recorder
    finally:
        try:
            recorder._finish()
        except Exception:
            recorder.usage = None
        finally:
            _safe(recorder._stop_sampler, None)
            recorder._deregister()