"""Cross-process serialization for benchmark measurement phases.

Co-running benchmarks poison each other's timings: a lane measuring latency

while another process is compiling Rust or running its own timed workload is

recording machine contention, not VCS behavior. The harness therefore

serializes measured work across processes while still allowing editing, builds,

fixture generation, and result writing to run in parallel outside this lock.

Set ``OAK_BENCH_LOCK=off`` to disable locking. That is only safe on a machine

running one benchmark at a time. The lock file defaults to

``tempfile.gettempdir() / "oak-bench-measurement.lock"`` and can be overridden

with ``OAK_BENCH_LOCK_PATH``.

"""

from __future__ import annotations

import contextlib

import errno

import os

import sys

import tempfile

import time

from dataclasses import dataclass

from pathlib import Path

from typing import Iterator, TextIO

try:

    import fcntl

except ImportError:  # pragma: no cover - exercised on platforms without fcntl

    fcntl = None  # type: ignore[assignment]

PROGRESS_INTERVAL_S = 30.0

POLL_INTERVAL_S = 0.1

BLOCKED_ERRNOS = {errno.EACCES, errno.EAGAIN}

if hasattr(errno, "EWOULDBLOCK"):

    BLOCKED_ERRNOS.add(errno.EWOULDBLOCK)

@dataclass(frozen=True)

class LockInfo:

    wait_ms: float | None

    enabled: bool

    path: str | None = None

def measurement_lock_path() -> Path:

    """Canonical identity of the lock every measurement lane will use."""

    configured = os.environ.get("OAK_BENCH_LOCK_PATH")

    if configured:

        return Path(configured).expanduser().resolve()

    return (Path(tempfile.gettempdir()) / "oak-bench-measurement.lock").resolve()

def _lock_path() -> Path:

    return measurement_lock_path()

def _read_holder(handle: TextIO) -> str:

    try:

        handle.seek(0)

        text = handle.read().strip()

    except OSError:

        return "unknown"

    if not text:

        return "unknown"

    return text.splitlines()[0].strip() or "unknown"

def _write_holder(handle: TextIO, label: str) -> None:

    try:

        handle.seek(0)

        handle.truncate()

        handle.write(f"{label} pid={os.getpid()}\n")

        handle.flush()

    except OSError:

        pass

def _warn_disabled(reason: str) -> None:

    print(f"[runlock] measurement lock disabled: {reason}", file=sys.stderr, flush=True)

@contextlib.contextmanager

def measurement_lock(label: str) -> Iterator[LockInfo]:

    """Acquire the process-wide benchmark measurement lock."""

    path = _lock_path()

    if os.environ.get("OAK_BENCH_LOCK", "").lower() == "off":

        yield LockInfo(wait_ms=None, enabled=False, path=str(path))

        return

    if fcntl is None:

        _warn_disabled("fcntl is not available on this platform")

        yield LockInfo(wait_ms=None, enabled=False, path=str(path))

        return

    try:

        path.parent.mkdir(parents=True, exist_ok=True)

        fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o666)

        handle = os.fdopen(fd, "r+", encoding="utf-8")

    except OSError as exc:

        _warn_disabled(f"could not open {path}: {exc}")

        yield LockInfo(wait_ms=None, enabled=False, path=str(path))

        return

    wait_start = time.perf_counter()

    next_progress = wait_start + PROGRESS_INTERVAL_S

    acquired = False

    try:

        while True:

            try:

                fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)

                acquired = True

                break

            except BlockingIOError:

                pass

            except OSError as exc:

                if getattr(exc, "errno", None) not in BLOCKED_ERRNOS:

                    _warn_disabled(f"could not acquire {path}: {exc}")

                    yield LockInfo(wait_ms=None, enabled=False, path=str(path))

                    return

            now = time.perf_counter()

            if now >= next_progress:

                waited_s = int(now - wait_start)

                holder = _read_holder(handle)

                print(

                    f"[runlock] waiting for measurement lock held by {holder} ({waited_s}s)",

                    file=sys.stderr,

                    flush=True,

                )

                next_progress += PROGRESS_INTERVAL_S

            time.sleep(min(POLL_INTERVAL_S, max(0.0, next_progress - time.perf_counter())))

        wait_ms = (time.perf_counter() - wait_start) * 1000

        _write_holder(handle, label)

        yield LockInfo(wait_ms=wait_ms, enabled=True, path=str(path))

    finally:

        try:

            if acquired:

                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)

        finally:

            handle.close()