"""Bounded, filesystem-only monitoring of known simulation SQLite files."""
from __future__ import annotations
import logging
import os
import threading
import time
from collections import deque
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from pathlib import Path
POLL_SECONDS = 2.0
GROWTH_SECONDS = 20.0
HISTORY_LIMIT = 11
COMPANIONS = ("", "-wal", "-shm", "-journal")
[docs]
@dataclass(frozen=True)
class FileSize:
"""One database's aggregate size, including its SQLite companion files."""
path: Path
"""Absolute main database path identifying the group."""
size: int | None
"""Combined logical size in bytes, or None when any member cannot be read."""
bytes_per_second: float | None
"""Net size change per second, or None until two observations exist."""
error: str | None = None
"""Filesystem error text for an unreadable file."""
[docs]
class FileSizeTracker:
"""Measure known files without opening SQLite or scanning directories.
Parameters
----------
paths : iterable of Path
Main database paths. WAL, shared-memory, and rollback-journal companions
are included automatically when present.
clock : callable, optional
Monotonic clock used for growth measurements.
stat : callable, optional
File metadata reader, replaceable for deterministic tests.
"""
def __init__(
self,
paths: Iterable[Path],
clock: Callable[[], float] = time.monotonic,
stat: Callable[[Path], os.stat_result] = Path.stat,
) -> None:
self.paths = tuple(dict.fromkeys(Path(path).absolute() for path in paths))
self._candidates = {path: tuple(Path(str(path) + suffix) for suffix in COMPANIONS) for path in self.paths}
self._clock = clock
self._stat = stat
self._history: dict[Path, tuple[tuple[int, int, float] | None, deque[tuple[float, int]]]] = {}
[docs]
def sample(self) -> tuple[FileSize, ...]:
"""Read sizes and net growth over at most twenty seconds.
Each main database and its companions share one size history. Companion
creation and removal contribute to net growth without resetting it.
Rates reset when the whole group disappears, the main file is replaced,
or any member is unreadable, and need two observations to become defined.
Returns
-------
tuple of FileSize
Existing or unreadable database groups and their size/growth estimates.
"""
now = self._clock()
rows = []
for path, candidates in self._candidates.items():
prior = self._history.get(path)
identity = prior[0] if prior is not None else None
size = 0
found = False
errors = []
for candidate in candidates:
try:
info = self._stat(candidate)
except FileNotFoundError:
continue
except OSError as error:
errors.append(f"{candidate.name}: {error.strerror or type(error).__name__}")
continue
found = True
size += info.st_size
if candidate == path:
# Only main-file replacement resets the aggregate history;
# companion turnover is ordinary SQLite activity.
identity = (info.st_dev, info.st_ino, getattr(info, "st_birthtime", 0.0))
if errors:
self._history.pop(path, None)
rows.append(FileSize(path, None, None, "; ".join(errors)))
continue
if not found:
self._history.pop(path, None)
continue
history = prior[1] if prior is not None and prior[0] == identity else deque(maxlen=HISTORY_LIMIT)
if history and now <= history[-1][0]:
history.clear()
history.append((now, size))
while len(history) > 1 and history[0][0] < now - GROWTH_SECONDS:
history.popleft()
self._history[path] = (identity, history)
elapsed = now - history[0][0]
rate = (size - history[0][1]) / elapsed if elapsed > 0 else None
rows.append(FileSize(path, size, rate))
return tuple(rows)
@dataclass
class _RunFiles:
tracker: FileSizeTracker
rows: tuple[FileSize, ...] = ()
[docs]
class DatabaseFileMonitor:
"""Share one background metadata poller among the live dashboard's runs.
Parameters
----------
refresh : callable
Throttled dashboard refresh, called without holding the monitor lock.
interval : float, optional
Poll interval in seconds; defaults to two seconds.
Notes
-----
Filesystem calls never hold the display or registration locks. Stopping
signals the daemon thread and waits briefly, so a slow network filesystem
cannot delay simulation cleanup indefinitely.
"""
def __init__(self, refresh: Callable[[], None], interval: float = POLL_SECONDS) -> None:
self._refresh = refresh
self._interval = interval
self._lock = threading.Lock()
self._runs: dict[str, _RunFiles] = {}
self._thread: threading.Thread | None = None
self._stop = threading.Event()
[docs]
def register(self, key: str, paths: Iterable[Path]) -> None:
"""Register known paths, starting the shared poller only when needed."""
state = _RunFiles(FileSizeTracker(paths))
with self._lock:
self._runs[key] = state
if self._thread is None:
self._start_locked()
def _start_locked(self) -> None:
self._stop = threading.Event()
self._thread = threading.Thread(
target=self._run, args=(self._stop,), name="passengersim-database-files", daemon=True
)
self._thread.start()
[docs]
def remove(self, key: str) -> None:
"""Forget a finished run and stop polling when no registered runs remain."""
thread = None
with self._lock:
self._runs.pop(key, None)
if not self._runs:
self._stop.set()
thread = self._thread
self._join(thread)
[docs]
def close(self) -> None:
"""Release all histories and stop polling when the dashboard closes."""
with self._lock:
self._runs.clear()
self._stop.set()
thread = self._thread
self._join(thread)
@staticmethod
def _join(thread: threading.Thread | None) -> None:
if thread is not None and thread is not threading.current_thread():
thread.join(timeout=0.25)
[docs]
def snapshot(self) -> dict[str, tuple[tuple[Path, ...], tuple[FileSize, ...]]]:
"""Copy cached display data without touching the filesystem.
Returns
-------
dict
Run identifiers mapped to known main paths and observed file rows.
"""
with self._lock:
return {key: (state.tracker.paths, state.rows) for key, state in self._runs.items()}
def _run(self, stop: threading.Event) -> None:
try:
while not stop.is_set():
with self._lock:
runs = list(self._runs.items())
for key, state in runs:
if stop.is_set():
return
rows = state.tracker.sample()
with self._lock:
# A removed/replaced run must not reappear after a slow stat.
if self._runs.get(key) is state and not stop.is_set():
state.rows = rows
if not stop.is_set():
try:
self._refresh()
except Exception:
# Display failures must not terminate metadata collection.
logging.getLogger(__name__).debug("Database file panel refresh failed", exc_info=True)
stop.wait(self._interval)
finally:
with self._lock:
self._thread = None
# Registration may occur while a previous slow stat is retiring.
# Start its replacement only after the old polling loop exits.
if self._runs:
self._start_locked()