Viewing files at commit 5717cc9fc7edCommit detailsView latest

scripts/oakbench/execution.py 234 lines · 8.1 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
# CPU and block I/O come from the same wait4 rusage as peak RSS: the direct
29
# child's own usage (on macOS and Linux, including descendants it reaped
30
# before exiting; daemons it leaves running are not included).
31
RUSAGE_SOURCE = "os_wait4_child_rusage" if _HAS_WAIT4 else None
32
33
34
@dataclass(frozen=True)
35
class TimedCapture:
36
    elapsed_ms: float
37
    returncode: int
38
    stdout_text: str
39
    stdout_bytes: int
40
    stdout_truncated: bool
41
    stderr_text: str
42
    stderr_bytes: int
43
    stderr_truncated: bool
44
    peak_rss_bytes: Optional[int] = None
45
    stdout_full_text: Optional[str] = None
46
    stdout_full_text_truncated: Optional[bool] = None
47
    stderr_full_text: Optional[str] = None
48
    stderr_full_text_truncated: Optional[bool] = None
49
    timed_out: bool = False
50
    stdout_raw: bytes | None = None
51
    cpu_user_ms: Optional[float] = None
52
    cpu_sys_ms: Optional[float] = None
53
    io_read_blocks: Optional[int] = None
54
    io_write_blocks: Optional[int] = None
55
    major_page_faults: Optional[int] = None
56
57
58
def stop_owned_process(proc: subprocess.Popen[Any]) -> None:
59
    """Stop the timed child and its owned process group, then always reap it."""
60
    try:
61
        if os.name == "posix":
62
            os.killpg(proc.pid, signal.SIGTERM)
63
        else:
64
            proc.terminate()
65
    except (OSError, ProcessLookupError):
66
        pass
67
    try:
68
        proc.wait(timeout=1.0)
69
    except subprocess.TimeoutExpired:
70
        pass
71
72
    # The process-group leader may exit on SIGTERM while a descendant ignores
73
    # it. Probe and kill the owned group even after the leader has been reaped.
74
    if os.name == "posix":
75
        try:
76
            os.killpg(proc.pid, 0)
77
        except (OSError, ProcessLookupError):
78
            pass
79
        else:
80
            try:
81
                os.killpg(proc.pid, signal.SIGKILL)
82
            except (OSError, ProcessLookupError):
83
                pass
84
    elif proc.poll() is None:
85
        try:
86
            proc.kill()
87
        except (OSError, ProcessLookupError):
88
            pass
89
90
    if proc.poll() is None:
91
        proc.wait()
92
93
94
class _RusagePopen(subprocess.Popen):
95
    """Popen that records the child's rusage when it is reaped.
96
97
    Overriding _try_wait keeps a single wait4 syscall as both the reap and the
98
    measurement; there is no race against the kernel discarding rusage and no
99
    extra process in the timed path.
100
    """
101
102
    rusage = None
103
104
    def _try_wait(self, wait_flags):  # type: ignore[override]
105
        try:
106
            pid, sts, rusage = os.wait4(self.pid, wait_flags)
107
        except ChildProcessError:
108
            pid = self.pid
109
            sts = 0
110
        else:
111
            if pid == self.pid:
112
                self.rusage = rusage
113
        return (pid, sts)
114
115
116
def _peak_rss_bytes(rusage: Any) -> Optional[int]:
117
    if rusage is None:
118
        return None
119
    maxrss = getattr(rusage, "ru_maxrss", None)
120
    if not maxrss or maxrss < 0:
121
        return None
122
    # macOS reports bytes; Linux (and most other Unixes) report KiB.
123
    return int(maxrss) if sys.platform == "darwin" else int(maxrss) * 1024
124
125
126
def rusage_fields(rusage: Any) -> dict[str, Any]:
127
    """CPU user/sys ms and block I/O counts; None when unmeasured (ADR-0002).
128
129
    ru_inblock/ru_oublock count real block I/O, so a warm page cache yields a
130
    measured 0, not an unmeasured value.
131
    """
132
    if rusage is None:
133
        return {"cpu_user_ms": None, "cpu_sys_ms": None, "io_read_blocks": None,
134
                "io_write_blocks": None, "major_page_faults": None}
135
    return {
136
        "cpu_user_ms": round(float(rusage.ru_utime) * 1000.0, 3),
137
        "cpu_sys_ms": round(float(rusage.ru_stime) * 1000.0, 3),
138
        "io_read_blocks": int(rusage.ru_inblock),
139
        "io_write_blocks": int(rusage.ru_oublock),
140
        "major_page_faults": int(rusage.ru_majflt),
141
    }
142
143
144
def read_admitted_text(handle: Any, char_limit: int) -> tuple[str, int, bool]:
145
    handle.flush()
146
    handle.seek(0, os.SEEK_END)
147
    byte_count = handle.tell()
148
    handle.seek(0)
149
    if char_limit <= 0:
150
        return "", byte_count, byte_count > 0
151
    raw = handle.read(char_limit * 4 + 16)
152
    text = raw.decode("utf-8", "replace")[:char_limit]
153
    return text, byte_count, byte_count > len(text.encode("utf-8", "replace"))
154
155
156
def read_full_text_with_byte_limit(handle: Any, byte_limit: int) -> tuple[str | None, bool]:
157
    handle.flush()
158
    handle.seek(0, os.SEEK_END)
159
    byte_count = handle.tell()
160
    if byte_limit < 0:
161
        return None, byte_count > 0
162
    if byte_count > byte_limit:
163
        return None, True
164
    handle.seek(0)
165
    return handle.read().decode("utf-8", "replace"), False
166
167
168
def run_timed(
169
    command: list[str],
170
    cwd: Path,
171
    env: dict[str, str],
172
    admitted_output_chars: int,
173
    full_output_bytes: int | None = None,
174
    timeout_seconds: float | None = None,
175
    raw_output_bytes: int | None = None,
176
) -> TimedCapture:
177
    if timeout_seconds is not None and timeout_seconds <= 0:
178
        raise ValueError("timeout_seconds must be positive")
179
    if raw_output_bytes is not None and raw_output_bytes < 0:
180
        raise ValueError("raw_output_bytes must be nonnegative")
181
    popen_cls = _RusagePopen if _HAS_WAIT4 else subprocess.Popen
182
    start = time.perf_counter()
183
    with tempfile.TemporaryFile() as stdout_fh, tempfile.TemporaryFile() as stderr_fh:
184
        proc = popen_cls(
185
            command,
186
            cwd=cwd,
187
            env=env,
188
            stdout=stdout_fh,
189
            stderr=stderr_fh,
190
            start_new_session=timeout_seconds is not None and os.name == "posix",
191
        )
192
        timed_out = False
193
        try:
194
            returncode = proc.wait(timeout=timeout_seconds)
195
        except subprocess.TimeoutExpired:
196
            timed_out = True
197
            stop_owned_process(proc)
198
            returncode = 124
199
        elapsed_ms = (time.perf_counter() - start) * 1000
200
        stdout_text, stdout_bytes, stdout_truncated = read_admitted_text(stdout_fh, admitted_output_chars)
201
        stderr_text, stderr_bytes, stderr_truncated = read_admitted_text(stderr_fh, admitted_output_chars)
202
        stdout_raw = None
203
        if raw_output_bytes is not None and stdout_bytes <= raw_output_bytes:
204
            stdout_fh.seek(0)
205
            stdout_raw = stdout_fh.read()
206
        stdout_full_text = None
207
        stdout_full_text_truncated = None
208
        stderr_full_text = None
209
        stderr_full_text_truncated = None
210
        if full_output_bytes is not None:
211
            stdout_full_text, stdout_full_text_truncated = read_full_text_with_byte_limit(
212
                stdout_fh, full_output_bytes
213
            )
214
            stderr_full_text, stderr_full_text_truncated = read_full_text_with_byte_limit(
215
                stderr_fh, full_output_bytes
216
            )
217
    return TimedCapture(
218
        elapsed_ms=elapsed_ms,
219
        returncode=returncode,
220
        stdout_text=stdout_text,
221
        stdout_bytes=stdout_bytes,
222
        stdout_truncated=stdout_truncated,
223
        stderr_text=stderr_text,
224
        stderr_bytes=stderr_bytes,
225
        stderr_truncated=stderr_truncated,
226
        peak_rss_bytes=_peak_rss_bytes(getattr(proc, "rusage", None)),
227
        stdout_full_text=stdout_full_text,
228
        stdout_full_text_truncated=stdout_full_text_truncated,
229
        stderr_full_text=stderr_full_text,
230
        stderr_full_text_truncated=stderr_full_text_truncated,
231
        timed_out=timed_out,
232
        stdout_raw=stdout_raw,
233
        **rusage_fields(getattr(proc, "rusage", None)),
234
    )