Viewing files at commit 13de5e5480b3Commit detailsView latest

scripts/oakbench/execution.py 197 lines · 6.5 KB · python
1
"""Timed command execution with admitted-output capture.
2
3
This is the measurement primitive every CLI lane shares: run a command under a
4
monotonic clock, capture full output bytes to disk, and admit only a capped
5
character window into token estimates (full byte counts are always recorded).
6
7
Memory: the child's peak RSS is captured at reap time via os.wait4 (rusage),
8
so no wrapper binary or sampling thread perturbs the timing. ru_maxrss units
9
differ by platform (bytes on macOS, KiB on Linux); peak_rss_bytes is always
10
bytes. None means unmeasured (ADR-0002), never zero — e.g. on platforms
11
without wait4.
12
"""
13
14
from __future__ import annotations
15
16
import os
17
import signal
18
import subprocess
19
import sys
20
import tempfile
21
import time
22
from dataclasses import dataclass
23
from pathlib import Path
24
from typing import Any, Optional
25
26
_HAS_WAIT4 = hasattr(os, "wait4")
27
PEAK_RSS_SOURCE = "os_wait4_ru_maxrss" if _HAS_WAIT4 else None
28
29
30
@dataclass(frozen=True)
31
class TimedCapture:
32
    elapsed_ms: float
33
    returncode: int
34
    stdout_text: str
35
    stdout_bytes: int
36
    stdout_truncated: bool
37
    stderr_text: str
38
    stderr_bytes: int
39
    stderr_truncated: bool
40
    peak_rss_bytes: Optional[int] = None
41
    stdout_full_text: Optional[str] = None
42
    stdout_full_text_truncated: Optional[bool] = None
43
    stderr_full_text: Optional[str] = None
44
    stderr_full_text_truncated: Optional[bool] = None
45
    timed_out: bool = False
46
47
48
def stop_owned_process(proc: subprocess.Popen[Any]) -> None:
49
    """Stop the timed child and its owned process group, then always reap it."""
50
    try:
51
        if os.name == "posix":
52
            os.killpg(proc.pid, signal.SIGTERM)
53
        else:
54
            proc.terminate()
55
    except (OSError, ProcessLookupError):
56
        pass
57
    try:
58
        proc.wait(timeout=1.0)
59
    except subprocess.TimeoutExpired:
60
        pass
61
62
    # The process-group leader may exit on SIGTERM while a descendant ignores
63
    # it. Probe and kill the owned group even after the leader has been reaped.
64
    if os.name == "posix":
65
        try:
66
            os.killpg(proc.pid, 0)
67
        except (OSError, ProcessLookupError):
68
            pass
69
        else:
70
            try:
71
                os.killpg(proc.pid, signal.SIGKILL)
72
            except (OSError, ProcessLookupError):
73
                pass
74
    elif proc.poll() is None:
75
        try:
76
            proc.kill()
77
        except (OSError, ProcessLookupError):
78
            pass
79
80
    if proc.poll() is None:
81
        proc.wait()
82
83
84
class _RusagePopen(subprocess.Popen):
85
    """Popen that records the child's rusage when it is reaped.
86
87
    Overriding _try_wait keeps a single wait4 syscall as both the reap and the
88
    measurement; there is no race against the kernel discarding rusage and no
89
    extra process in the timed path.
90
    """
91
92
    rusage = None
93
94
    def _try_wait(self, wait_flags):  # type: ignore[override]
95
        try:
96
            pid, sts, rusage = os.wait4(self.pid, wait_flags)
97
        except ChildProcessError:
98
            pid = self.pid
99
            sts = 0
100
        else:
101
            if pid == self.pid:
102
                self.rusage = rusage
103
        return (pid, sts)
104
105
106
def _peak_rss_bytes(rusage: Any) -> Optional[int]:
107
    if rusage is None:
108
        return None
109
    maxrss = getattr(rusage, "ru_maxrss", None)
110
    if not maxrss or maxrss < 0:
111
        return None
112
    # macOS reports bytes; Linux (and most other Unixes) report KiB.
113
    return int(maxrss) if sys.platform == "darwin" else int(maxrss) * 1024
114
115
116
def read_admitted_text(handle: Any, char_limit: int) -> tuple[str, int, bool]:
117
    handle.flush()
118
    handle.seek(0, os.SEEK_END)
119
    byte_count = handle.tell()
120
    handle.seek(0)
121
    if char_limit <= 0:
122
        return "", byte_count, byte_count > 0
123
    raw = handle.read(char_limit * 4 + 16)
124
    text = raw.decode("utf-8", "replace")[:char_limit]
125
    return text, byte_count, byte_count > len(text.encode("utf-8", "replace"))
126
127
128
def read_full_text_with_byte_limit(handle: Any, byte_limit: int) -> tuple[str | None, bool]:
129
    handle.flush()
130
    handle.seek(0, os.SEEK_END)
131
    byte_count = handle.tell()
132
    if byte_limit < 0:
133
        return None, byte_count > 0
134
    if byte_count > byte_limit:
135
        return None, True
136
    handle.seek(0)
137
    return handle.read().decode("utf-8", "replace"), False
138
139
140
def run_timed(
141
    command: list[str],
142
    cwd: Path,
143
    env: dict[str, str],
144
    admitted_output_chars: int,
145
    full_output_bytes: int | None = None,
146
    timeout_seconds: float | None = None,
147
) -> TimedCapture:
148
    if timeout_seconds is not None and timeout_seconds <= 0:
149
        raise ValueError("timeout_seconds must be positive")
150
    popen_cls = _RusagePopen if _HAS_WAIT4 else subprocess.Popen
151
    start = time.perf_counter()
152
    with tempfile.TemporaryFile() as stdout_fh, tempfile.TemporaryFile() as stderr_fh:
153
        proc = popen_cls(
154
            command,
155
            cwd=cwd,
156
            env=env,
157
            stdout=stdout_fh,
158
            stderr=stderr_fh,
159
            start_new_session=timeout_seconds is not None and os.name == "posix",
160
        )
161
        timed_out = False
162
        try:
163
            returncode = proc.wait(timeout=timeout_seconds)
164
        except subprocess.TimeoutExpired:
165
            timed_out = True
166
            stop_owned_process(proc)
167
            returncode = 124
168
        elapsed_ms = (time.perf_counter() - start) * 1000
169
        stdout_text, stdout_bytes, stdout_truncated = read_admitted_text(stdout_fh, admitted_output_chars)
170
        stderr_text, stderr_bytes, stderr_truncated = read_admitted_text(stderr_fh, admitted_output_chars)
171
        stdout_full_text = None
172
        stdout_full_text_truncated = None
173
        stderr_full_text = None
174
        stderr_full_text_truncated = None
175
        if full_output_bytes is not None:
176
            stdout_full_text, stdout_full_text_truncated = read_full_text_with_byte_limit(
177
                stdout_fh, full_output_bytes
178
            )
179
            stderr_full_text, stderr_full_text_truncated = read_full_text_with_byte_limit(
180
                stderr_fh, full_output_bytes
181
            )
182
    return TimedCapture(
183
        elapsed_ms=elapsed_ms,
184
        returncode=returncode,
185
        stdout_text=stdout_text,
186
        stdout_bytes=stdout_bytes,
187
        stdout_truncated=stdout_truncated,
188
        stderr_text=stderr_text,
189
        stderr_bytes=stderr_bytes,
190
        stderr_truncated=stderr_truncated,
191
        peak_rss_bytes=_peak_rss_bytes(getattr(proc, "rusage", None)),
192
        stdout_full_text=stdout_full_text,
193
        stdout_full_text_truncated=stdout_full_text_truncated,
194
        stderr_full_text=stderr_full_text,
195
        stderr_full_text_truncated=stderr_full_text_truncated,
196
        timed_out=timed_out,
197
    )