"""Linux /proc sampling outside the server process; no synthetic resource data."""
from __future__ import annotations

import os
from pathlib import Path
import threading
import time


def cgroup_cpu_stat(path=Path("/sys/fs/cgroup/cpu.stat")):
    values = {key: int(value) for key, value in (line.split() for line in path.read_text().splitlines())}
    if not {"nr_throttled", "throttled_usec"} <= values.keys():
        raise RuntimeError(f"{path} lacks cgroup v2 throttling counters")
    return {"path": str(path), "observed_monotonic": time.monotonic(), "values": values}


def cgroup_cpu_delta(before, after):
    delta = {key: after["values"][key] - value for key, value in before["values"].items()
             if key in after["values"]}
    if any(value < 0 for value in delta.values()):
        raise RuntimeError("cgroup CPU counters reset during the measured run")
    return delta


def parse_cpu_set(value):
    if not value:
        raise ValueError("an explicit, nonempty CPU set is required")
    result = set()
    for part in value.split(","):
        bounds = part.strip().split("-")
        if len(bounds) not in (1, 2) or any(not item.isascii() or not item.isdigit() for item in bounds):
            raise ValueError(f"invalid CPU set {value!r}; use comma-separated numbers or ranges")
        start, end = int(bounds[0]), int(bounds[-1])
        if start > end or end - start > 65536:
            raise ValueError(f"invalid CPU range {part!r}")
        result.update(range(start, end + 1))
    return result


def validate_cpu_sets(values, allowed):
    groups = {label: parse_cpu_set(value) for label, value in values.items()}
    for label, cpus in groups.items():
        if not cpus <= allowed:
            raise ValueError(f"{label} CPUs {sorted(cpus - allowed)} are outside the allowed affinity")
        for other, other_cpus in groups.items():
            if label < other and cpus & other_cpus:
                raise ValueError(f"{label} and {other} CPU sets overlap: {sorted(cpus & other_cpus)}")
    return groups


def affinity_snapshot(root, expected, expected_processes=None):
    """Verify every currently visible thread in the root process and children."""
    tasks = []
    for pid in descendants(root):
        try:
            threads = list(Path(f"/proc/{pid}/task").iterdir())
        except FileNotFoundError:
            continue
        for thread in threads:
            try:
                actual = set(os.sched_getaffinity(int(thread.name)))
            except ProcessLookupError:
                continue
            if actual != expected:
                raise RuntimeError(f"affinity mismatch for PID {pid} TID {thread.name}: "
                                   f"expected {sorted(expected)}, got {sorted(actual)}")
            tasks.append({"pid": pid, "tid": int(thread.name), "cpus": sorted(actual)})
    if not tasks or not any(task["pid"] == root for task in tasks):
        raise RuntimeError(f"cannot inspect affinity of live process {root}")
    processes = len({task["pid"] for task in tasks})
    if expected_processes is not None and processes != expected_processes:
        raise RuntimeError(f"process count mismatch for PID {root}: "
                           f"expected {expected_processes}, inspected {processes}")
    return {"root_pid": root, "processes": processes,
            "threads": len(tasks), "tasks": tasks}


def wait_for_affinity(process, expected, timeout=2):
    """Allow taskset to execute before inspecting the generator's inherited mask."""
    deadline = time.monotonic() + timeout
    while True:
        if process.poll() is not None:
            raise RuntimeError(f"generator exited before affinity verification: {process.returncode}")
        try:
            return affinity_snapshot(process.pid, expected)
        except RuntimeError:
            if time.monotonic() >= deadline:
                raise
            time.sleep(.01)


def descendants(pid, proc_root=Path("/proc")):
    """Find the process tree through stat PPids, including nginx workers.

    /proc/PID/task/PID/children is optional on some Linux kernels and may also
    omit children created by another thread. Missing that file is not evidence
    that a live process is a leaf. A single PPid-map scan avoids either case.
    """
    children, seen = {}, set()
    for entry in proc_root.iterdir():
        if not entry.name.isdecimal():
            continue
        current = int(entry.name)
        try:
            fields = (entry / "stat").read_text().rsplit(")", 1)[1].split()
            parent = int(fields[1])
        except (FileNotFoundError, ProcessLookupError):
            # A non-root process may finish while the snapshot is assembled.
            continue
        except (OSError, ValueError, IndexError) as error:
            raise RuntimeError(f"cannot discover process tree from PID {current}: {error}") from error
        if fields[0] in {"Z", "X", "x"}:
            # A dead/zombie worker cannot satisfy the configured live count.
            continue
        seen.add(current)
        children.setdefault(parent, []).append(current)
    if pid not in seen:
        raise RuntimeError(f"no readable live root process for PID {pid}")
    result, pending = [], [pid]
    while pending:
        current = pending.pop()
        if current in result:
            continue
        result.append(current)
        pending.extend(sorted(children.get(current, []), reverse=True))
    return result


def process_stats(pid):
    try:
        data = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
        if data[0] in {"Z", "X", "x"}:
            return None
        ticks = os.sysconf("SC_CLK_TCK")
        pages = os.sysconf("SC_PAGE_SIZE")
        return {"pid": pid, "start_ticks": int(data[19]),
                "cpu_seconds": (int(data[11]) + int(data[12])) / ticks,
                "rss_bytes": int(data[21]) * pages,
                "fd": len(list(Path(f"/proc/{pid}/fd").iterdir())),
                "threads": int(data[17])}
    except (FileNotFoundError, ProcessLookupError):
        return None
    except (OSError, ValueError, IndexError) as error:
        raise RuntimeError(f"cannot read resource counters for PID {pid}: {error}") from error


class Monitor:
    def __init__(self, roots, interval=.5, expected_processes=None):
        self.roots = roots
        self.expected_processes = expected_processes or {}
        self.process_identities = {}
        self.interval = interval
        if interval <= 0:
            raise ValueError("resource interval must be positive")
        self.samples = []
        self.errors = []
        self.stop_event = threading.Event()
        self.started = time.monotonic()
        self.thread = threading.Thread(target=self.run, daemon=True)

    def sample(self):
        groups = {}
        for label, root in self.roots.items():
            processes = [stats for pid in descendants(root) if (stats := process_stats(pid)) is not None]
            if not any(process["pid"] == root for process in processes):
                raise RuntimeError(f"no readable live root process for {label} (PID {root})")
            expected = self.expected_processes.get(label)
            if expected is not None and len(processes) != expected:
                raise RuntimeError(f"resource process count mismatch for {label} (PID {root}): "
                                   f"expected {expected}, read {len(processes)}")
            identities = {(process["pid"], process["start_ticks"]) for process in processes}
            if expected is not None:
                original = self.process_identities.setdefault(label, identities)
                if identities != original:
                    raise RuntimeError(f"process identities changed while measuring {label}: "
                                       f"expected {sorted(original)}, read {sorted(identities)}")
            groups[label] = {"processes": processes,
                             "rss_bytes": sum(p["rss_bytes"] for p in processes),
                             "fd": sum(p["fd"] for p in processes),
                             "threads": sum(p["threads"] for p in processes)}
        self.samples.append({"seconds": time.monotonic() - self.started, "groups": groups})

    def run(self):
        while not self.stop_event.wait(self.interval):
            try:
                self.sample()
            except Exception as error:
                self.errors.append(str(error))
                self.stop_event.set()

    def start(self):
        self.sample()
        self.thread.start()

    def stop(self):
        self.stop_event.set()
        self.thread.join(timeout=3)
        if self.thread.is_alive():
            self.errors.append("resource sampler did not stop")
        try:
            self.sample()
        except Exception as error:
            self.errors.append(str(error))

    def summary(self, requests):
        if self.errors:
            raise RuntimeError("resource sampling failed: " + "; ".join(self.errors))
        result = {}
        for label in self.roots:
            observations = [sample["groups"][label] for sample in self.samples]
            totals = {}
            for group in observations:
                for process in group["processes"]:
                    key = (process["pid"], process["start_ticks"])
                    low, high = totals.get(key, (process["cpu_seconds"], process["cpu_seconds"]))
                    totals[key] = (min(low, process["cpu_seconds"]), max(high, process["cpu_seconds"]))
            if not totals:
                raise RuntimeError(f"no readable process samples for {label}; refusing zero resource metrics")
            seconds = sum(high - low for low, high in totals.values())
            result[label] = {"cpu_seconds": seconds,
                             "cpu_us_per_request": seconds * 1_000_000 / requests if requests else None,
                             "rss_peak_bytes": max(group["rss_bytes"] for group in observations),
                             "rss_start_bytes": observations[0]["rss_bytes"],
                             "rss_end_bytes": observations[-1]["rss_bytes"],
                             "fd_peak": max(group["fd"] for group in observations),
                             "fd_start": observations[0]["fd"], "fd_end": observations[-1]["fd"],
                             "processes_seen": len(totals)}
        return result