lmcache--lmcache/lmcache/v1/mp_observability/trace/reader.py

# SPDX-License-Identifier: Apache-2.0

"""Streaming reader for trace files.

The reader yields ``(Header, Iterator[Record])`` pairs.  Records are
yielded lazily so that arbitrarily large traces can be inspected
without loading the whole file into memory.

Trailing partial frames (truncated by SIGKILL or filesystem buffering)
are detected and the iterator stops cleanly with a WARNING log.
"""

# Future
from __future__ import annotations

# Standard
from typing import BinaryIO, Iterator
import struct

# First Party
from lmcache.logging import init_logger
from lmcache.v1.mp_observability.trace.format import (
    FORMAT_VERSION,
    MAGIC,
    TRACE_SCHEMA_VERSION,
    Header,
    Record,
    decode_header,
    decode_record,
)

logger = init_logger(__name__)

_LEN_STRUCT = struct.Struct(">I")
_LEN_PREFIX = _LEN_STRUCT.size


class TraceReader:
    """Streaming reader for a binary trace file.

    Usage::

        with TraceReader("/tmp/run.lct") as r:
            header = r.header
            for record in r.records():
                ...

    Closing the reader closes the underlying file handle.
    """

    def __init__(self, path: str) -> None:
        self._path = path
        self._fh: BinaryIO | None = open(path, "rb")
        try:
            self._header = self._read_header()
        except Exception:
            self._fh.close()
            self._fh = None
            raise

    def __enter__(self) -> TraceReader:
        return self

    def __exit__(self, *_exc: object) -> None:
        self.close()

    @property
    def header(self) -> Header:
        """Return the file header.  Always present; populated by
        ``__init__``."""
        return self._header

    @property
    def path(self) -> str:
        """Path of the trace file."""
        return self._path

    def records(self) -> Iterator[Record]:
        """Yield every record in the file in order.

        Yields each :class:`Record` as it is read.  When the file ends
        cleanly (boundary aligned to a frame), iteration stops without
        error.  When a partial trailing frame is detected, a warning
        is logged and iteration stops.
        """
        if self._fh is None:
            raise RuntimeError("TraceReader is closed")
        while True:
            frame = self._read_frame(strict=False)
            if frame is None:
                return
            try:
                yield decode_record(frame)
            except Exception as e:
                logger.warning(
                    "TraceReader: skipping malformed record at offset %d: %s",
                    self._fh.tell(),
                    e,
                )
                continue

    def close(self) -> None:
        """Close the underlying file.  Idempotent."""
        if self._fh is not None:
            self._fh.close()
            self._fh = None

    # ---- internal -----------------------------------------------------

    def _read_header(self) -> Header:
        frame = self._read_frame(strict=True)
        if frame is None:
            raise ValueError(f"trace file {self._path!r} is empty")
        header = decode_header(frame)
        if header.magic != MAGIC:
            raise ValueError(
                f"trace file {self._path!r}: bad magic "
                f"(got {header.magic!r}, expected {MAGIC!r})"
            )
        if header.format_version != FORMAT_VERSION:
            raise ValueError(
                f"trace file {self._path!r}: unsupported format_version "
                f"{header.format_version} (this build expects {FORMAT_VERSION})"
            )
        if header.trace_schema_version != TRACE_SCHEMA_VERSION:
            raise ValueError(
                f"trace file {self._path!r}: unsupported trace_schema_version "
                f"{header.trace_schema_version} "
                f"(this build expects {TRACE_SCHEMA_VERSION})"
            )
        return header

    def _read_frame(self, strict: bool) -> bytes | None:
        """Read one length-prefixed frame.

        Returns ``None`` on clean EOF (when ``strict=False``).  On
        truncation in the middle of a frame, logs a WARNING and
        returns ``None``.  In ``strict=True`` mode, both partial and
        empty reads raise.
        """
        assert self._fh is not None
        prefix = self._fh.read(_LEN_PREFIX)
        if not prefix:
            if strict:
                raise ValueError("unexpected EOF reading frame length")
            return None
        if len(prefix) < _LEN_PREFIX:
            msg = (
                f"truncated frame length prefix at offset "
                f"{self._fh.tell() - len(prefix)}"
            )
            if strict:
                raise ValueError(msg)
            logger.warning("TraceReader: %s", msg)
            return None
        (length,) = _LEN_STRUCT.unpack(prefix)
        body = self._fh.read(length)
        if len(body) < length:
            msg = (
                f"truncated frame body at offset "
                f"{self._fh.tell() - len(body)} (got {len(body)} of {length})"
            )
            if strict:
                raise ValueError(msg)
            logger.warning("TraceReader: %s", msg)
            return None
        return body