frame_timing.py

"""System-wide frame timing: one set of numbers for every presented frame.

Each scroller already logs its own stats line (ScrollHelper.log_frame_rate,
the Vegas coordinator's "Vegas FPS"), but in different formats, per source,
and Vegas only logs a healthy window at DEBUG. None of that answers the
question a release has to answer on each rig: *over a long run, how often did
a moving frame reach the panel late?*

Every frame reaches the panel through ``DisplayManager.update_display``, so it
is recorded there, once, whoever drew it. The render thread only appends a
tuple; a worker thread aggregates, and every ``flush_interval`` seconds writes
cumulative counters and histograms to a small JSON file -- in ``/dev/shm`` where
it exists, so a stats file refreshed all day costs no SD-card writes.
``scripts/frame_soak.py`` reads it twice and reports the difference.

What is counted
---------------
Only intervals between two consecutive *scrolling* frames count: a static
screen that changes once a second has no timing to get wrong, and the first
frame of a scroll has no predecessor worth measuring against.

"Scrolling" is the scroll state ``DisplayManager.update_display`` acted on for
the frame, sampled once before the blit and swap, and that state can go missing
in the middle of a scroll. It expires after 2s
without scroll activity, which a long enough stall outlasts, and any thread can
clear it: plugins call ``set_scrolling_state(False)`` from their own
``display()``, and Vegas captures some of those on the render thread between
two of its frames. The frame after that is recorded as static, and the interval
it ends -- the stall, or the capture -- would vanish from the report. So a
single static frame between two scrolling ones, with the scroll picking up
again within ``RESUME_SECONDS``, is treated as a frame of the scroll: both of
its intervals count. A second static frame in a row means the scroll really
ended. (On hdpi on 2026-09-24 the watchdog logged a 1.9s stall that the soak
report did not have; this is how.)

A frame held for ``hold`` refreshes should arrive ``hold`` refresh periods
after the one before it. One that arrives a whole refresh or more after that is
**late**: the panel showed the previous frame again, which on a moving strip is
a visible hitch. ``missed_refreshes`` sums how many refreshes late.

An interval of ``FREEZE_SECONDS`` or more is a **freeze** instead -- a
recompose, a plugin handover nobody tagged (see below), a blocking call on the
render thread. Those are counted separately, both because they are a different
fault and because folding a single 400ms handover into the late count as "40
missed refreshes" would drown the jitter the late count exists to measure.
``freeze_by`` splits them by length. Intervals of ``GAP_SECONDS`` or more are
ignored as not being frames of one scroll at all.

One kind of freeze is not a scroll stalling at all: the gap from one screen's
last frame to the next screen's first, while the next screen draws. The
display controller tags that frame ``handover`` (see "Operations") at the
start of every turn, the same mode's again included, and a tagged freeze is
counted in ``handover_freezes`` instead of ``freezes`` and ``freeze_by``.
Stats written before that field existed have handovers among their freezes,
so freeze counts from before and after it are not comparable.

A frame that arrives a whole refresh or more *early* means the swap did not
wait for the panel: the emulator, the fallback display, or a hold that was not
the one in effect. Those are counted as **early**, and a run with more than a
trace of them was not locked to the panel, so its late count means nothing.

The refresh period is estimated from the frames themselves: swaps that block
on vsync can only land on refresh boundaries, so the low end of
interval / hold is the period. It is the smallest per-window 10th percentile
seen so far, over windows with enough frames to trust -- except that a window
cutting it by more than ``MAX_REFRESH_DROP`` is ignored. A panel's refresh does
not jump like that; swaps that stopped blocking do, and adopting their period
would make every early frame look on time.

A caller that has measured the panel independently -- ``scripts/render_bench.py``
times bare swaps first with :func:`measure_refresh_hz` -- passes that rate in
as ``refresh_hz``. The estimate then starts from it instead of from the frames,
which is what catches a loop that never locked at all: one that free-runs
faster than the panel (every frame early) or sits at half its rate (every
frame late), both of which look self-consistent to an estimate taken from
their own intervals.

Operations
----------
The late count says how often, not which work did it. Render-thread work that
happens between two frames -- extending the Vegas strip, patching a live
element into it -- calls :meth:`FrameTimingRecorder.note_op` first, and the
next presented frame carries the tag: the interval that frame ends is the one
the work landed in. ``op_frames`` counts timed frames per kind,
``late_op_frames`` the late ones among them, ``op_freezes`` those that were a
freeze instead, and ``op_bytes`` what the work moved. A kind whose late rate
sits well above the overall one is the work to look at.

``handover`` (:data:`HANDOVER_OP`) is noted off the render thread: the display
controller notes it just before it starts a screen's first ``display()``,
which presents from a thread of its own, and drops the note again with
:meth:`FrameTimingRecorder.drop_op` once that call returns, so a first
``display()`` that drew nothing cannot leave the tag for an unrelated frame.

Garbage collection
------------------

Python's cyclic collector stops every thread while it runs. :class:`GcMonitor`
times each collection from ``gc.callbacks``; the display manager installs one
per process. A collection of ``GC_PAUSE_SECONDS`` or more tags the next
presented frame ``gc`` (:data:`GC_OP`), so it shows in ``op_frames``,
``late_op_frames`` and ``op_freezes`` like noted work, and the snapshot carries
a ``gc`` block of cumulative counters: collections and seconds per generation,
the longest, and the long ones. A stall dump says when a long collection ran
inside the stall. Diagnostic only: nothing tunes or freezes the collector.

Stall watchdog
--------------
Counting a freeze says that it happened, not why. ``StallWatchdog`` watches the
same frames from its own thread and, when a scroll's last frame is more than
``STALL_SECONDS`` old, logs the stack of the thread that presented it and the
top of every other thread's, so the log names what the render thread was
waiting on. It also measures how late its own wake-up was: if the watchdog was
held up as long as the render thread, the whole interpreter was blocked (C
code holding the GIL, or the process not scheduled), not one thread on a lock.
Set ``LEDMATRIX_STALL_WATCHDOG=0`` to turn it off, or
``LEDMATRIX_STALL_WATCHDOG_MS`` to dump at a lower threshold -- 30 catches
frames three refreshes late, which is where GIL contention shows. It polls
three times per threshold, so keep it to diagnostic runs, not soaks.
"""

from __future__ import annotations

import atexit
import copy
import gc
import json
import logging
import os
import queue
import sys
import tempfile
import threading
import time
import traceback
from typing import Any, Callable, Dict, List, Optional, Tuple, TypedDict

logger = logging.getLogger(__name__)

#: Bumped when a field changes meaning, so a reader can refuse stale files.
SCHEMA_VERSION = 1

#: Histogram resolution. 64ms of range covers any frame worth drawing a
#: distribution of; everything beyond lands in the last bucket.
BUCKET_MS = 0.25
BUCKET_COUNT = 256

#: See the module docstring.
FREEZE_SECONDS = 0.25

#: Intervals this long are not frames of one scroll. This used to be 1s,
#: which silently dropped every 1-2s stall inside a scroll. It is now only a
#: sanity bound.
GAP_SECONDS = 5.0

#: A frame recorded as static between two scrolling frames is a frame of the
#: scroll whose state went missing, if the scroll resumes within this long.
#: See "What is counted".
RESUME_SECONDS = 1.0

#: Buckets for freeze length, as cumulative counters a soak can difference.
FREEZE_BUCKETS = ((0.5, "<0.5s"), (1.0, "0.5-1s"), (2.0, "1-2s"),
                  (float("inf"), "2s+"))

#: The op the display controller notes before a screen's first frame. A
#: freeze it ends is a handover, counted apart from the freezes; see
#: "What is counted".
HANDOVER_OP = "handover"

#: The op a garbage collection of ``GC_PAUSE_SECONDS`` or more tags the next
#: frame with; see "Garbage collection".
GC_OP = "gc"
GC_PAUSE_SECONDS = 0.020

#: A window may lower the refresh-period estimate by at most this fraction.
MAX_REFRESH_DROP = 0.2

#: A window needs this many scrolling frames before its refresh estimate is
#: trusted -- about a second of scrolling.
MIN_FRAMES_FOR_REFRESH = 90

FLUSH_INTERVAL = 10.0

#: A scroll's last frame older than this is a stall worth a stack dump.
STALL_SECONDS = 0.25
#: How often the watchdog looks. Also the resolution of its starvation check.
WATCHDOG_POLL_SECONDS = 0.05
#: At most one stack dump per this many seconds: a stall that repeats every
#: extension would otherwise write the same stacks to the SD card all day.
STALL_LOG_INTERVAL = 30.0

#: Written by the display service, read by scripts/frame_soak.py and anything
#: else that wants the numbers. The web UI's viewer marker lives in /tmp; this
#: goes to RAM where there is some, since it is rewritten all day.
STATS_FILENAME = "ledmatrix_frame_stats.json"


def default_stats_path() -> str:
    # A fixed name in a shared directory is safe here: write() creates its
    # temp file with mkstemp and os.replace()s it over this path, which swaps
    # out whatever is there -- a planted symlink included -- without following it.
    base = "/dev/shm" if os.path.isdir("/dev/shm") else tempfile.gettempdir()  # nosec B108
    return os.path.join(base, STATS_FILENAME)


class GcMonitor:
    """Times every garbage collection, from ``gc.callbacks``.

    Python's cyclic collector stops every thread for as long as a collection
    takes, and a full one over a large heap (a season of game dicts) can take
    longer than a frame. Nothing measured that, so a stall it caused looked
    like any other. The callback runs inside the collection, with the GIL
    held, and collections never overlap, so these plain counters need no
    lock: the render thread and the stats writer only read them.

    Install it once per process with :func:`install_gc_monitor`.

    Collections still run while the interpreter shuts down, after module
    globals such as ``time`` may already be torn down to ``None``. The clock
    and ``sys.is_finalizing`` are bound here so the callback never looks a
    global up, it does nothing once finalization has begun, and
    :func:`install_gc_monitor` unregisters it at exit anyway.
    """

    def __init__(self, threshold: float = GC_PAUSE_SECONDS,
                 clock: Callable[[], float] = time.perf_counter):
        self.threshold = threshold
        self._clock = clock
        self._is_finalizing = sys.is_finalizing
        self._started: Optional[float] = None
        #: Per generation (0, 1, 2), since the monitor was installed.
        self.collections = [0, 0, 0]
        self.seconds = [0.0, 0.0, 0.0]
        self.max_seconds = 0.0
        #: Collections of ``threshold`` or more, and their total length. The
        #: recorder compares ``long_pauses`` with the count it last saw to tag
        #: the next frame.
        self.long_pauses = 0
        self.long_seconds = 0.0
        #: ``time.perf_counter()`` at the end of the last long collection,
        #: and its length, for the stall watchdog.
        self.last_long: Optional[Tuple[float, float]] = None

    def __call__(self, phase: str, info: Dict[str, Any]) -> None:
        if self._is_finalizing():
            return
        now = self._clock()
        if phase == "start":
            self._started = now
            return
        started, self._started = self._started, None
        if started is None:
            return
        took = now - started
        generation = min(max(int(info.get("generation", 0)), 0), 2)
        self.collections[generation] += 1
        self.seconds[generation] += took
        if took > self.max_seconds:
            self.max_seconds = took
        if took >= self.threshold:
            self.long_seconds += took
            self.last_long = (now, took)
            self.long_pauses += 1

    def snapshot(self) -> Dict[str, Any]:
        """Cumulative counters for the stats file (all since installation)."""
        return {
            "threshold_ms": round(self.threshold * 1000.0, 3),
            "collections": list(self.collections),
            "seconds": [round(x, 6) for x in self.seconds],
            "max_ms": round(self.max_seconds * 1000.0, 3),
            "long_pauses": self.long_pauses,
            "long_seconds": round(self.long_seconds, 6),
        }


_gc_monitor: Optional[GcMonitor] = None
_gc_monitor_lock = threading.Lock()


def install_gc_monitor() -> GcMonitor:
    """The process's GcMonitor, installed in ``gc.callbacks`` on first call.

    It is unregistered at exit (:func:`uninstall_gc_monitor`), before the
    interpreter tears module globals down.
    """
    global _gc_monitor
    with _gc_monitor_lock:
        if _gc_monitor is None:
            _gc_monitor = GcMonitor()
            gc.callbacks.append(_gc_monitor)
            atexit.register(uninstall_gc_monitor)
        return _gc_monitor


def uninstall_gc_monitor() -> None:
    """Take the process's GcMonitor out of ``gc.callbacks``; safe to repeat.

    A recorder that still holds the monitor keeps its counters; they just
    stop moving. The next :func:`install_gc_monitor` installs a fresh one.
    """
    global _gc_monitor
    with _gc_monitor_lock:
        monitor, _gc_monitor = _gc_monitor, None
        if monitor is None:
            return
        atexit.unregister(uninstall_gc_monitor)
        try:
            gc.callbacks.remove(monitor)
        except ValueError:
            pass


#: One presented frame's interval: (interval, blit, wait, hold, ops), where
#: ops is the work noted before it (kind -> bytes) or None.
_Frame = Tuple[float, float, float, int, Optional[Dict[str, int]]]


def _bucket(seconds: float) -> int:
    index = int(seconds * 1000.0 / BUCKET_MS)
    return min(max(index, 0), BUCKET_COUNT - 1)


def _bump(counter: Dict[str, int], key: str, by: int = 1) -> None:
    counter[key] = counter.get(key, 0) + by


def binding_releases_gil() -> Optional[bool]:
    """Whether the loaded rgbmatrix binding releases the GIL, or None.

    The stock binding blocks in SwapOnVSync holding the GIL, which starves
    every other thread for most of each frame (docs/SCROLL_PERFORMANCE.md).
    scripts/build_rgbmatrix_nogil.sh rebuilds it, and the rebuilt module links
    PyEval_SaveThread where the stock one never does -- a crude test, but the
    only one that needs neither a probe on the panel nor the source tree the
    module was built from. None when no hardware binding is loaded.
    """
    module = sys.modules.get("rgbmatrix.core")
    path = getattr(module, "__file__", None)
    if not path:
        return None
    try:
        with open(path, "rb") as handle:
            return b"PyEval_SaveThread" in handle.read()
    except OSError:
        return None


def _pi_model() -> Optional[str]:
    try:
        with open("/proc/device-tree/model", "rb") as handle:
            return handle.read().rstrip(b"\0").decode("ascii", "replace").strip()
    except OSError:
        return None


def measure_refresh_hz(matrix: Any, seconds: float = 4.0) -> float:
    """The panel's refresh rate with nothing else running, by timing bare swaps.

    ``SwapOnVSync`` blocks until the panel's next refresh, so a loop that does
    nothing else runs at exactly the panel's rate. ``limit_refresh_rate_hz`` is
    a *cap*, and a long chain, a high ``pwm_bits`` or an older Pi will sit well
    under it. Solving scroll speeds against a cap the panel cannot reach is
    what produces "3px every 4 refreshes" and the judder that comes with it.

    This is the idle rate. The panel refreshes a few percent slower while the
    Pi is also pushing frames into it (100.4Hz idle against 96.3Hz scrolling on
    a Pi 4 driving 512x64), which is why the recorder reads the rendering rate
    back from the frames rather than trusting this.

    Pass the matrix the display is already running on rather than opening a
    second one: the GPIO has a single owner, and the options in force change
    the answer.

    :returns: measured Hz, or 0.0 if the matrix cannot be swapped (no
        hardware, a stub, a mock).
    """
    try:
        canvas = matrix.CreateFrameCanvas()
        # Discard the first swap: it carries construction and first-touch costs
        # that have nothing to do with the steady-state refresh.
        canvas = matrix.SwapOnVSync(canvas)
    except Exception:  # pylint: disable=broad-except
        return 0.0

    frames = 0
    started = time.perf_counter()
    while time.perf_counter() - started < seconds:
        canvas = matrix.SwapOnVSync(canvas)
        frames += 1
    elapsed = time.perf_counter() - started
    if elapsed <= 0 or frames <= 0:
        return 0.0
    return frames / elapsed

class FrameTimingRecorder:
    """Collects per-frame timings on the render thread; aggregates elsewhere.

    ``record`` is the only method the render thread calls, and it does no more
    than compare two floats and append a tuple.
    """

    def __init__(
        self,
        path: Optional[str] = None,
        flush_interval: float = FLUSH_INTERVAL,
        info: Optional[Dict[str, Any]] = None,
        refresh_hz: Optional[float] = None,
        gc_monitor: Optional[GcMonitor] = None,
    ):
        """
        :param refresh_hz: the panel's rate, measured independently (see the
            module docstring). Omit it to estimate from the frames alone, as
            the display service does.
        :param gc_monitor: tags frames after a long garbage collection and
            adds its counters to the stats (see "Garbage collection"). The
            display manager passes the process's :func:`install_gc_monitor`.
        """
        self.path = path or default_stats_path()
        self.flush_interval = flush_interval
        self.info = dict(info or {})

        # Render-thread state.
        self._pending: List[_Frame] = []
        self._static_frames = 0
        self._previous: Optional[Tuple[float, bool, int]] = None
        # The interval ended by a static frame that followed a scrolling one,
        # until the next frame shows whether the scroll went on.
        self._unsure: Optional[_Frame] = None
        # Work noted since the last frame (kind -> bytes), for the next one.
        self._ops: Optional[Dict[str, int]] = None
        self.gc_monitor = gc_monitor
        self._gc_seen = gc_monitor.long_pauses if gc_monitor is not None else 0
        self._last_flush: Optional[float] = None
        self._queue: "queue.SimpleQueue" = queue.SimpleQueue()
        self._worker: Optional[threading.Thread] = None

        # Worker-thread state. Nothing on the render thread reads these.
        self.started = time.time()
        self.refresh_period: Optional[float] = (
            1.0 / refresh_hz if refresh_hz and refresh_hz > 0 else None)
        # The first estimate, until a second window agrees with it.
        self._refresh_candidate: Optional[float] = None
        self.totals: Dict[str, Any] = {
            "static_frames": 0,
            "scroll_frames": 0,
            "late_frames": 0,
            "missed_refreshes": 0,
            "late_by": {"1": 0, "2": 0, "3-5": 0, "6+": 0},
            "early_frames": 0,
            # Frames judged against a known refresh period: the denominator
            # for the late and early rates. Frames before the period is known
            # are neither, and must not dilute them.
            "timed_frames": 0,
            "freezes": 0,
            "freeze_seconds": 0.0,
            "freeze_by": {label: 0 for _, label in FREEZE_BUCKETS},
            # Freezes that ended a screen handover rather than stalled a
            # scroll: in neither of the two above. Additive; see "What is
            # counted".
            "handover_freezes": 0,
            "worst_interval_ms": 0.0,
            # Per kind of noted render-thread work; see "Operations".
            "op_frames": {},
            "late_op_frames": {},
            "op_freezes": {},
            "op_bytes": {},
        }
        self.histograms: Dict[str, Dict[int, int]] = {
            "blit": {}, "wait": {}, "work": {}, "interval_per_hold": {},
        }
        self._binding_gil: Optional[bool] = None
        self._binding_checked = False

        # Read by the stall watchdog from its own thread: one tuple assignment,
        # so it always sees a consistent (time, scrolling, thread) triple.
        self.last_frame: Optional[Tuple[float, bool, int]] = None
        #: Whether a scroll is running *now*, supplied by the display manager.
        #: The last frame's flag alone would call the end of every scroll a
        #: stall.
        self.scrolling_now: Optional[Callable[[], bool]] = None
        self.watchdog: Optional["StallWatchdog"] = None

    def close(self) -> None:
        """Stop the stall watchdog, if one was started."""
        watchdog, self.watchdog = self.watchdog, None
        if watchdog is not None:
            watchdog.stop()

    # -- render thread ------------------------------------------------------

    def note_op(self, kind: str, nbytes: int = 0) -> None:
        """Tag the next presented frame with work done before it.

        Render thread only, like :meth:`record`, which consumes the tag: the
        interval the next frame ends is the one this work landed in. Several
        notes before one frame accumulate, per kind. See "Operations" (and
        :data:`HANDOVER_OP`, the one note made from another thread).

        :param kind: a short name for the work, e.g. ``"extend"``, ``"patch"``.
        :param nbytes: how much the work moved, summed into ``op_bytes``.
        """
        ops = self._ops
        if ops is None:
            ops = self._ops = {}
        ops[kind] = ops.get(kind, 0) + int(nbytes)

    def drop_op(self, kind: str) -> None:
        """Forget a note of ``kind`` that no frame has carried yet.

        For work that may present nothing: the display controller notes a
        handover before a screen's first ``display()`` and drops it once that
        returns. When the call drew a frame, the frame already took the tag
        and this does nothing; when it drew nothing (no content), the tag
        would otherwise land on whatever frame came next -- seconds or minutes
        later, and nothing to do with the handover. Other kinds noted for the
        same frame are kept.
        """
        ops = self._ops
        if ops is not None:
            ops.pop(kind, None)

    def record(self, blit: float, wait: float, hold: int, scrolling: bool,
               presented_at: float) -> None:
        """One frame reached the panel.

        :param blit: seconds spent copying the frame into the canvas.
        :param wait: seconds SwapOnVSync blocked.
        :param hold: the refreshes this frame was held for.
        :param scrolling: whether a scroll was running for this frame: the
            scroll state ``update_display`` acted on, sampled once before the
            blit and swap.
        :param presented_at: ``time.perf_counter()`` when the swap returned.
        """
        previous = self._previous
        self._previous = (presented_at, scrolling, hold)
        self.last_frame = (presented_at, scrolling, threading.get_ident())
        ops, self._ops = self._ops, None
        monitor = self.gc_monitor
        if monitor is not None and monitor.long_pauses != self._gc_seen:
            # A long collection ran since the last frame: the interval this
            # frame ends is the one it landed in. Read here rather than
            # noted, since note_op is the render thread's and a collection
            # runs on whichever thread triggered it.
            self._gc_seen = monitor.long_pauses
            ops = dict(ops) if ops else {}
            ops[GC_OP] = ops.get(GC_OP, 0)
        if not scrolling:
            self._static_frames += 1
            # The scroll ended, or its state went missing for this frame: the
            # next frame says which. Its hold may have been dropped with the
            # state, so the interval is due at the scroll's own.
            self._unsure = None
            if previous is not None and previous[1]:
                self._unsure = (presented_at - previous[0], blit, wait,
                                previous[2], ops)
        elif self.watchdog is None and self.scrolling_now is not None \
                and os.environ.get("LEDMATRIX_STALL_WATCHDOG", "1") != "0":
            self.watchdog = StallWatchdog(self, **watchdog_settings())
            self.watchdog.start()
        elif previous is not None:
            interval = presented_at - previous[0]
            unsure, self._unsure = self._unsure, None
            if previous[1]:
                if interval < GAP_SECONDS:
                    self._pending.append((interval, blit, wait, hold, ops))
            elif unsure is not None and interval < RESUME_SECONDS:
                # One static frame between two scrolling ones: the scroll never
                # stopped, only its state did. Both intervals were motion.
                self._static_frames -= 1
                if unsure[0] < GAP_SECONDS:
                    self._pending.append(unsure)
                self._pending.append((interval, blit, wait, hold, ops))

        if self._last_flush is None:
            self._last_flush = presented_at
        elif presented_at - self._last_flush >= self.flush_interval:
            self._hand_off()
            self._last_flush = presented_at

    def _hand_off(self) -> None:
        batch, self._pending = self._pending, []
        static, self._static_frames = self._static_frames, 0
        self._queue.put((batch, static))
        if self._worker is None or not self._worker.is_alive():
            self._worker = threading.Thread(
                target=self._run, daemon=True, name="frame-timing")
            self._worker.start()

    def drain(self) -> None:
        """Aggregate everything recorded so far, on the calling thread.

        For a caller that owns the recorder outright and wants exact numbers at
        a moment of its choosing -- the benchmark, between warm-up and run and
        at the end. Construct it with ``flush_interval=float('inf')`` so the
        worker never runs; the two must not aggregate at once.
        """
        batch, self._pending = self._pending, []
        static, self._static_frames = self._static_frames, 0
        self.aggregate(batch, static)

    # -- worker thread ------------------------------------------------------

    def _run(self) -> None:
        while True:
            batch, static = self._queue.get()
            try:
                self.aggregate(batch, static)
                self.write()
            except Exception:  # never let telemetry take anything down
                logger.debug("Frame timing flush failed", exc_info=True)

    def aggregate(self, batch: List[_Frame], static: int) -> None:
        """Fold one window of frames into the running totals.

        Each frame is ``(interval, blit, wait, hold, ops)``; ``ops`` (the work
        noted before it, or None) may be left off.
        """
        totals = self.totals
        totals["static_frames"] += static

        per_hold = sorted(frame[0] / max(1, frame[3]) for frame in batch
                          if frame[0] < FREEZE_SECONDS)
        if len(per_hold) >= MIN_FRAMES_FOR_REFRESH:
            estimate = per_hold[len(per_hold) // 10]
            current = self.refresh_period
            if estimate <= 0:
                pass
            elif current is None:
                # Adopt the first period only once two windows in a row agree:
                # one loaded window at startup, most of its frames a refresh
                # late, would otherwise fix a period twice the real one for
                # the life of the process, since later windows may only lower
                # it by MAX_REFRESH_DROP.
                candidate = self._refresh_candidate
                if candidate and abs(estimate - candidate) <= candidate * MAX_REFRESH_DROP:
                    self.refresh_period = min(candidate, estimate)
                else:
                    self._refresh_candidate = estimate
            elif current * (1.0 - MAX_REFRESH_DROP) <= estimate < current:
                self.refresh_period = estimate
        period = self.refresh_period

        histograms = self.histograms
        for frame in batch:
            interval, blit, wait, hold = frame[:4]
            ops = frame[4] if len(frame) > 4 else None
            totals["worst_interval_ms"] = max(totals["worst_interval_ms"],
                                              interval * 1000.0)
            if ops:
                for kind, nbytes in ops.items():
                    _bump(totals["op_bytes"], kind, nbytes)
            if interval >= FREEZE_SECONDS:
                for kind in ops or ():
                    _bump(totals["op_freezes"], kind)
                if ops and HANDOVER_OP in ops:
                    # The next screen drawing its first frame, not a scroll
                    # that stalled: counted apart, so the freezes keep
                    # meaning the second. See "What is counted".
                    totals["handover_freezes"] += 1
                    continue
                totals["freezes"] += 1
                totals["freeze_seconds"] += interval
                label = next(name for limit, name in FREEZE_BUCKETS
                             if interval < limit)
                totals["freeze_by"][label] += 1
                continue
            totals["scroll_frames"] += 1
            for name, value in (("blit", blit), ("wait", wait),
                                ("work", max(0.0, interval - blit - wait)),
                                ("interval_per_hold", interval / max(1, hold))):
                bucket = _bucket(value)
                histogram = histograms[name]
                histogram[bucket] = histogram.get(bucket, 0) + 1
            if period:
                totals["timed_frames"] += 1
                missed = round(interval / period) - hold
                for kind in ops or ():
                    _bump(totals["op_frames"], kind)
                    if missed >= 1:
                        _bump(totals["late_op_frames"], kind)
                if missed >= 1:
                    totals["late_frames"] += 1
                    totals["missed_refreshes"] += missed
                    key = ("1" if missed == 1 else "2" if missed == 2
                           else "3-5" if missed <= 5 else "6+")
                    totals["late_by"][key] += 1
                elif missed <= -1:
                    totals["early_frames"] += 1

    def snapshot(self) -> Dict[str, Any]:
        """The JSON document: cumulative since this process started."""
        if not self._binding_checked:
            self._binding_gil = binding_releases_gil()
            self._binding_checked = True
        info = dict(self.info)
        info.setdefault("pi_model", _pi_model())
        period = self.refresh_period
        return {
            "version": SCHEMA_VERSION,
            "pid": os.getpid(),
            "started": self.started,
            "updated": time.time(),
            "bucket_ms": BUCKET_MS,
            "freeze_seconds": FREEZE_SECONDS,
            "measured_refresh_hz": round(1.0 / period, 2) if period else None,
            "binding_releases_gil": self._binding_gil,
            "info": info,
            "totals": copy.deepcopy(self.totals),
            # Additive: absent from older files and when no monitor is set.
            **({"gc": self.gc_monitor.snapshot()}
               if self.gc_monitor is not None else {}),
            # JSON keys are strings; readers convert back.
            "histograms": {name: {str(k): v for k, v in sorted(h.items())}
                           for name, h in self.histograms.items()},
        }

    def write(self) -> None:
        """Replace the stats file atomically with the current snapshot."""
        directory = os.path.dirname(self.path) or "."
        fd, tmp = tempfile.mkstemp(dir=directory, prefix=".frame_stats.",
                                   suffix=".tmp")
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(self.snapshot(), handle)
            os.chmod(tmp, 0o644)
            os.replace(tmp, self.path)
        except Exception:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise


class _WatchdogSettings(TypedDict, total=False):
    """The StallWatchdog keyword arguments watchdog_settings() may set."""
    threshold: float
    poll: float


def watchdog_settings() -> _WatchdogSettings:
    """StallWatchdog arguments from ``LEDMATRIX_STALL_WATCHDOG_MS``, if set.

    The poll comes down with the threshold, or a stall shorter than one poll
    would go unseen.
    """
    try:
        ms = float(os.environ.get("LEDMATRIX_STALL_WATCHDOG_MS") or 0)
    except ValueError:
        ms = 0.0
    if ms <= 0:
        return {}
    threshold = ms / 1000.0
    return {"threshold": threshold,
            "poll": min(WATCHDOG_POLL_SECONDS, threshold / 3)}


class StallWatchdog:
    """Log what the render thread is doing when a scroll stops presenting.

    See the module docstring. Polls; never touches the render thread.
    """

    def __init__(
        self,
        recorder: FrameTimingRecorder,
        threshold: float = STALL_SECONDS,
        poll: float = WATCHDOG_POLL_SECONDS,
        log_interval: float = STALL_LOG_INTERVAL,
        clock: Callable[[], float] = time.perf_counter,
    ):
        self.recorder = recorder
        self.threshold = threshold
        self.poll = poll
        self.log_interval = log_interval
        self.clock = clock
        self.stalls = 0
        self._last_dump: Optional[float] = None
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()

    def start(self) -> None:
        self._thread = threading.Thread(
            target=self._run, daemon=True, name="stall-watchdog")
        self._thread.start()

    def stop(self, timeout: float = 1.0) -> None:
        """End the polling thread (DisplayManager.cleanup calls this)."""
        self._stop.set()
        thread = self._thread
        if thread is not None and thread is not threading.current_thread():
            thread.join(timeout)

    def _run(self) -> None:
        last_wake = self.clock()
        stall_from: Optional[float] = None   # presented_at of the stalled frame
        dumped = False
        while not self._stop.wait(self.poll):
            now = self.clock()
            late = max(0.0, now - last_wake - self.poll)
            last_wake = now
            try:
                stall_from, dumped = self.check(now, late, stall_from, dumped)
            except Exception:  # never let a diagnostic take anything down
                logger.debug("Stall watchdog check failed", exc_info=True)

    def check(self, now: float, late: float, stall_from: Optional[float],
              dumped: bool) -> Tuple[Optional[float], bool]:
        """One look. Returns the updated (stall_from, dumped) state."""
        frame = self.recorder.last_frame
        if frame is None:
            return None, False
        presented_at, scrolling, ident = frame

        if stall_from is not None and presented_at != stall_from:
            # A frame arrived: the stall is over.
            if dumped:
                logger.warning(
                    "Render stall over: no frame for %.0fms",
                    (presented_at - stall_from) * 1000.0)
            return None, False

        scrolling_now = self.recorder.scrolling_now
        if (stall_from is not None and now - stall_from >= GAP_SECONDS
                and (scrolling_now is None or not scrolling_now())):
            # The scroll ended without another frame: nothing more to time.
            # Only past GAP_SECONDS: the scroll state expires after 2s without
            # activity, which a stall outlasts, and its end still wants saying.
            return None, False
        age = now - presented_at
        if (stall_from is None and scrolling and age >= self.threshold
                and scrolling_now is not None and scrolling_now()):
            self.stalls += 1
            if self._last_dump is None or now - self._last_dump >= self.log_interval:
                self._last_dump = now
                logger.warning(self.describe(ident, age, late))
                return presented_at, True
            return presented_at, False
        return stall_from, dumped

    def describe(self, ident: int, age: float, late: float) -> str:
        """The stack dump: the stalled thread in full, the rest in brief.

        A stall while a ``handover`` note is still waiting for its frame is
        the next screen's first ``display()`` taking its time, not a scroll
        that stopped, and is labelled a handover gap.
        """
        names = {t.ident: t.name for t in threading.enumerate()}
        frames = sys._current_frames()
        pending = getattr(self.recorder, "_ops", None)
        where = ("in a handover gap" if pending and HANDOVER_OP in pending
                 else "mid-scroll")
        monitor = getattr(self.recorder, "gc_monitor", None)
        last_long = getattr(monitor, "last_long", None)
        gc_note = ""
        if last_long is not None and time.perf_counter() - last_long[0] <= age:
            gc_note = (f"; a {last_long[1] * 1000.0:.0f}ms garbage collection "
                       "ran inside it")
        lines = [
            f"Render stall: no frame for {age * 1000.0:.0f}ms {where} "
            f"(watchdog woke {late * 1000.0:.0f}ms late"
            + ("; the interpreter itself was blocked" if late >= age / 2 else "")
            + gc_note + ")",
            f"-- {names.get(ident, ident)} (presents frames):",
        ]
        stalled = frames.get(ident)
        if stalled is not None:
            lines.extend(line.rstrip() for line in
                         traceback.format_stack(stalled, limit=12))
        for other, frame in frames.items():
            if other in (ident, threading.get_ident()):
                continue
            top = traceback.extract_stack(frame, limit=3)
            where = " <- ".join(
                f"{os.path.basename(f.filename)}:{f.lineno} {f.name}"
                for f in reversed(top))
            lines.append(f"-- {names.get(other, other)}: {where}")
        return "\n".join(lines)