k4bench.analysis.heap

k4bench.analysis.heap

The heap each component of a k4run job held: {label}_heap.json.

Written by the k4BenchAuditor in the k4run benchmark's memory pass, a run of the job of its own after the timed one, whose allocation counter records which component allocated every heap block. A block is charged to the component whose audited call allocated it, at any depth (a service initialized from within an algorithm's call is a call of its own), and stays charged to it until freed, wherever and whenever that happens. What no component allocated — the Python configuration, ROOT's start-up, the framework between calls — belongs to :data:NO_COMPONENT.

For every component, in bytes of heap (malloc's 8-byte header and rounding to 16 bytes included, as _components.json's allocation metrics count them):

• held before the first event, which is what initialize and start left, and the most it held at once until then;
• held at the start and the end of each event, and the most it held at once during the event (from its first top-level call to its last). The most it held beyond what it held at the start is its working set, the memory the event needed of it (:meth:ComponentHeap.working_set);
• held at the auditor's finalize: what survived the job's own finalize.

Beside them, at the same moments, the process's heap as malloc counts it (mallinfo2: the chunks in use, those it mapped on their own included, and the free ones it caches per thread) and its anonymous RSS. The components and :data:NO_COMPONENT add up to nearly malloc's count (to 0.06% in CLD reconstruction); the rest is the blocks allocated while the process was still loading, before the counter started, and the auditor's own memory. The anonymous RSS counts something else: it also holds what libraries map themselves, the free heap malloc has not returned and the counter's own table (table_mapped_bytes), but not heap that was allocated and never touched, which takes no pages. In CLD reconstruction that is about a sixth of the heap, so held bytes are memory allocated, not memory resident.

What the numbers do not say:

• who uses memory: a block is charged to the component that created it, so the event data the input reader creates stays the reader's until the output writer frees it, and the memory a library takes as it loads (its static constructors, ROOT dictionaries) is that of the component whose call loaded it;
• what lies outside malloc's heap: memory a library maps itself (pool allocators, CPython's arenas, large buffers) and file-backed memory (code, CVMFS), which _events.json's file-backed RSS shows;
• the timed run's values: the pass is another, shorter run of the same job. It reads the same input, so its events are the timed run's first ones, but recording owners slows every allocation, so its numbers are comparable in kind, not identical in value;
• what a service first used during an event keeps is held from that event on, not before the first: the service is initialized then.

In a job running calls concurrently (threads > 1 in _components.json), attribution still follows each thread's running component, but a snapshot taken while other threads run catches them mid-call. A job that had several events in flight at once has no per-event peaks: the owners' peaks are the process's.

A missing value is NaN: before the first event of a job that ran none, the per-event peaks of a job that ran events at once, and an RSS that could not be read.

HeapTotals dataclass

HeapTotals(before_first_mb: float, per_event_mb: Series, typical_mb: float, largest_mb: float, largest_event: int | None, growth_mb_per_event: float, after_finalize_mb: float)

The heap of the whole job, every component's and no component's together (:meth:ComponentHeap.totals), in MB.

ComponentHeap dataclass

ComponentHeap(execute: dict[str, DataFrame], before_first_event: DataFrame, at_end: DataFrame, process: DataFrame, process_before_first_event: Series, process_at_end: Series, table_peak_entries: int, table_mapped_bytes: int)

One label's _heap.json, as frames.

Components are :data:NO_COMPONENT first, then the auditor's components in its order, by name, as in the timed run's :class:~k4bench.analysis.components.ComponentTiming, which they join on.

execute maps each of :data:EVENT_HEAP_METRICS to an events × components frame indexed by event number: what each component held as the event started and at its end, and the most it held during it. before_first_event is components × :data:HEAP_METRICS, at_end components × held_bytes. process is events × :data:PROCESS_METRICS, and process_before_first_event and process_at_end the same at those two moments. table_peak_entries is the most blocks the counter's table held at once, table_mapped_bytes the memory it mapped for them.

from_json classmethod

from_json(raw: Mapping, *, source: str | Path | None = None) -> ComponentHeap

Build from a parsed _heap.json, refusing an inconsistent one.

The schema version is checked by the caller (:func:~k4bench.plugin.schema.validate_heap_schema).

Source code in k4bench/analysis/heap.py

@classmethod
def from_json(cls, raw: Mapping, *, source: str | Path | None = None) -> ComponentHeap:
    """Build from a parsed ``_heap.json``, refusing an inconsistent one.

    The schema version is checked by the caller
    (:func:`~k4bench.plugin.schema.validate_heap_schema`).
    """
    where = f"{source}: " if source is not None else ""
    missing = [
        key
        for key in ("components", "before_first_event", "event_numbers", "execute", "at_end", "table")
        if key not in raw
    ]
    if missing:
        raise ValueError(f"{where}missing keys: {missing}")

    names = list(raw["components"])
    if not names or names[0] != NO_COMPONENT:
        raise ValueError(f"{where}components must start with {NO_COMPONENT!r}")
    if len(set(names)) != len(names):
        raise ValueError(f"{where}component names are not unique")
    components = pd.Index(names, name="component")
    n = len(names)

    events = pd.Index(raw["event_numbers"], name="event_number")
    if events.has_duplicates:
        raise ValueError(f"{where}event_numbers contains duplicates")
    execute = raw["execute"]
    if absent := [key for key in (*EVENT_HEAP_METRICS, *PROCESS_METRICS) if key not in execute]:
        raise ValueError(f"{where}execute is missing {absent}")
    frames = {}
    for metric in EVENT_HEAP_METRICS:
        rows = execute[metric]
        if len(rows) != len(events) or any(len(row) != n for row in rows):
            raise ValueError(f"{where}execute.{metric} is not {len(events)} events × {n} components")
        frames[metric] = pd.DataFrame(rows, index=events, columns=components, dtype=float)
    for metric in PROCESS_METRICS:
        if len(execute[metric]) != len(events):
            raise ValueError(f"{where}execute.{metric} does not have one value per event")
    process = pd.DataFrame(
        {metric: execute[metric] for metric in PROCESS_METRICS}, index=events, dtype=float
    )
    # The auditor writes an RSS it could not read as negative.
    process = process.mask(process < 0)

    def moment(key: str, metrics: tuple[str, ...]) -> tuple[pd.DataFrame, pd.Series]:
        """The components × *metrics* and process values at moment *key*,
        all missing when the file has none."""
        snapshot = raw[key]
        if snapshot is None:
            return (
                pd.DataFrame(float("nan"), index=components, columns=list(metrics)),
                pd.Series(float("nan"), index=list(PROCESS_METRICS)),
            )
        if absent := [name for name in (*metrics, *PROCESS_METRICS) if name not in snapshot]:
            raise ValueError(f"{where}{key} is missing {absent}")
        if any(len(snapshot[metric]) != n for metric in metrics):
            raise ValueError(f"{where}{key} does not have one value per component")
        values = pd.Series({metric: snapshot[metric] for metric in PROCESS_METRICS}, dtype=float)
        return (
            pd.DataFrame({metric: snapshot[metric] for metric in metrics}, index=components, dtype=float),
            values.mask(values < 0),
        )

    before_first_event, process_before = moment("before_first_event", HEAP_METRICS)
    at_end, process_at_end = moment("at_end", ("held_bytes",))

    table = raw["table"]
    return cls(
        execute=frames,
        before_first_event=before_first_event,
        at_end=at_end,
        process=process,
        process_before_first_event=process_before,
        process_at_end=process_at_end,
        table_peak_entries=int(table["peak_entries"]),
        table_mapped_bytes=int(table["mapped_bytes"]),
    )

working_set

working_set() -> DataFrame

Events × components: the most each component held during each event beyond what it held as the event started, the heap the event needed of it. Its persistent state is left out, and so is the data others created that it read.

Source code in k4bench/analysis/heap.py

def working_set(self) -> pd.DataFrame:
    """Events × components: the most each component held during each event
    beyond what it held as the event started, the heap the event needed of
    it. Its persistent state is left out, and so is the data others created
    that it read."""
    return self.execute["peak_held_bytes"] - self.execute["start_held_bytes"]

holders

holders() -> Index

The components that held heap at some recorded moment: before the first event, at any event's start, peak or end (the warm-up event's included) or after finalize, in the file's order.

Source code in k4bench/analysis/heap.py

def holders(self) -> pd.Index:
    """The components that held heap at some recorded moment: before the
    first event, at any event's start, peak or end (the warm-up event's
    included) or after finalize, in the file's order."""
    moments = [self.before_first_event[metric] for metric in HEAP_METRICS]
    moments += [frame.max() for frame in self.execute.values()]
    moments.append(self.at_end["held_bytes"])
    held = pd.concat(moments, axis=1).fillna(0.0).gt(0).any(axis=1)
    return held.index[held]

steps

steps(*, warm_up: int | None = 0) -> DataFrame

Every rise of :data:STEP_MIN_MB or more in what a component held from one event's end to the next's, with columns component, event_number (the event it rose in), step_mb and kept (whether at every event's end from it to the last the component still held at least :data:STEP_KEPT_SHARE of it beyond what it held before it), components in the file's order and each one's events in order. The first event has no event before it, and the warm_up event's rise is not one, but its end is what the next event's rise is measured from.

Source code in k4bench/analysis/heap.py

def steps(self, *, warm_up: int | None = 0) -> pd.DataFrame:
    """Every rise of :data:`STEP_MIN_MB` or more in what a component held
    from one event's end to the next's, with columns ``component``,
    ``event_number`` (the event it rose in), ``step_mb`` and ``kept``
    (whether at every event's end from it to the last the component still
    held at least :data:`STEP_KEPT_SHARE` of it beyond what it held before
    it), components in the file's order and each one's events in order.
    The first event has no event before it, and the *warm_up* event's rise
    is not one, but its end is what the next event's rise is measured
    from."""
    held = self.execute["held_bytes"].sort_index() / MB
    rise = held.diff()
    # The least each component holds from each event's end to the last's.
    lowest = held[::-1].cummin()[::-1]
    if warm_up is not None:
        rise = rise[rise.index != warm_up]
    # Components × events, stacked: component order first, then events.
    long = rise.T.stack()
    long = long[long >= STEP_MIN_MB]
    names = long.index.get_level_values(0)
    events = long.index.get_level_values(1)
    step = long.to_numpy(dtype=float)
    kept = [
        lowest.at[e, name] - (held.at[e, name] - mb) >= STEP_KEPT_SHARE * mb
        for name, e, mb in zip(names, events, step)
    ]
    return pd.DataFrame({
        "component": names,
        "event_number": events.astype(int),
        "step_mb": step,
        "kept": np.asarray(kept, dtype=bool),
    })

summary

summary(*, warm_up: int | None = 0) -> DataFrame

Each component's heap over the job in MB, indexed by component in the file's order, with the :data:SUMMARY_COLUMNS:

• before_first_mb: what it held before the first event;
• typical_mb, largest_mb: the median and the largest of what it held at the events' ends;
• working_set_median_mb, working_set_p95_mb, working_set_max_mb: of its :meth:working_set, and working_set_max_event the event of its largest (NaN without one);
• growth_mb_per_event: the Theil–Sen slope of what it held at the events' ends against their numbers, NaN with too few events;
• after_finalize_mb: what it held after the job's own finalize;
• retained_mb: how much more than at the first event's end that is, NaN without events;
• leak: whether it is suspected of leaking: it grew by :data:LEAK_MIN_MB or more over the events, in both halves of them (each at :data:LEAK_STEADY_SHARE of the whole's rate or more), and after finalize still holds :data:LEAK_MIN_MB or more beyond what it held at the first event, and at least :data:LEAK_KEPT_SHARE of what it gained from the first event to the last. With fewer than :data:LEAK_MIN_EVENTS events none is;
• stepped_mb: what its kept :meth:steps added up to, 0 without any;
• stepped: whether it kept any of its steps.

The warm_up event is left out of every statistic over events.

Source code in k4bench/analysis/heap.py

def summary(self, *, warm_up: int | None = 0) -> pd.DataFrame:
    """Each component's heap over the job in MB, indexed by component in
    the file's order, with the :data:`SUMMARY_COLUMNS`:

    - ``before_first_mb``: what it held before the first event;
    - ``typical_mb``, ``largest_mb``: the median and the largest of what it
      held at the events' ends;
    - ``working_set_median_mb``, ``working_set_p95_mb``,
      ``working_set_max_mb``: of its :meth:`working_set`, and
      ``working_set_max_event`` the event of its largest (``NaN`` without
      one);
    - ``growth_mb_per_event``: the Theil–Sen slope of what it held at the
      events' ends against their numbers, ``NaN`` with too few events;
    - ``after_finalize_mb``: what it held after the job's own finalize;
    - ``retained_mb``: how much more than at the first event's end that
      is, ``NaN`` without events;
    - ``leak``: whether it is suspected of leaking: it grew by
      :data:`LEAK_MIN_MB` or more over the events, in both halves of them
      (each at :data:`LEAK_STEADY_SHARE` of the whole's rate or more), and
      after finalize still holds :data:`LEAK_MIN_MB` or more beyond what
      it held at the first event, and at least :data:`LEAK_KEPT_SHARE` of
      what it gained from the first event to the last. With fewer than
      :data:`LEAK_MIN_EVENTS` events none is;
    - ``stepped_mb``: what its kept :meth:`steps` added up to, ``0``
      without any;
    - ``stepped``: whether it kept any of its steps.

    The *warm_up* event is left out of every statistic over events.
    """
    held = _without(self.execute["held_bytes"], warm_up) / MB
    working = _without(self.working_set(), warm_up) / MB
    growth = pd.Series({name: _growth(held[name]) for name in held.columns}, dtype=float)
    span = float(held.index.max() - held.index.min()) if len(held) else 0.0
    nothing = pd.Series(np.nan, index=held.columns)
    first, last = (held.iloc[0], held.iloc[-1]) if len(held) else (nothing, nothing)
    after = self.at_end["held_bytes"] / MB
    gained, retained = last - first, after - first
    steps = self.steps(warm_up=warm_up)
    steps = steps[steps["kept"]]
    stepped = steps.groupby("component")["step_mb"].sum().reindex(held.columns, fill_value=0.0)
    kept = (
        (growth * span >= LEAK_MIN_MB) & (gained > 0) & (retained >= LEAK_MIN_MB)
        & (retained >= LEAK_KEPT_SHARE * gained)
    )
    return pd.DataFrame({
        "before_first_mb": self.before_first_event["held_bytes"] / MB,
        "typical_mb": held.median(),
        "largest_mb": held.max(),
        "working_set_median_mb": working.median(),
        "working_set_p95_mb": working.quantile(0.95),
        "working_set_max_mb": working.max(),
        "working_set_max_event": pd.Series({
            name: working[name].idxmax() if working[name].notna().any() else np.nan
            for name in working.columns
        }, dtype=float),
        "growth_mb_per_event": growth,
        "after_finalize_mb": after,
        "retained_mb": retained,
        "leak": kept & _steady(held[kept.index[kept]], growth),
        "stepped_mb": stepped,
        "stepped": stepped > 0,
    }, index=held.columns)

totals

totals(*, warm_up: int | None = 0) -> HeapTotals

The whole job's heap (:class:HeapTotals): what every component and no component held together, the warm_up event left out of the statistics over events.

Source code in k4bench/analysis/heap.py

def totals(self, *, warm_up: int | None = 0) -> HeapTotals:
    """The whole job's heap (:class:`HeapTotals`): what every component and
    no component held together, the *warm_up* event left out of the
    statistics over events."""
    per_event = self.execute["held_bytes"].sum(axis=1).sort_index() / MB
    events = _without(per_event, warm_up)
    return HeapTotals(
        before_first_mb=float(self.before_first_event["held_bytes"].sum(min_count=1) / MB),
        per_event_mb=per_event,
        typical_mb=float(events.median()) if len(events) else float("nan"),
        largest_mb=float(events.max()) if len(events) else float("nan"),
        largest_event=int(events.idxmax()) if len(events) else None,
        growth_mb_per_event=_growth(events),
        after_finalize_mb=float(self.at_end["held_bytes"].sum(min_count=1) / MB),
    )