Viewing files at commit 2e081b95a118Commit detailsView latest

scripts/oakbench/execution.py 206 lines · 6.9 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
    stdout_raw: bytes | None = None
47
48
49
def stop_owned_process(proc: subprocess.Popen[Any]) -> None:
50
    """Stop the timed child and its owned process group, then always reap it."""
51
    try:
52
        if os.name == "posix":
53
            os.killpg(proc.pid, signal.SIGTERM)
54
        else:
55
            proc.terminate()
56
    except (OSError, ProcessLookupError):
57
        pass
58
    try:
59
        proc.wait(timeout=1.0)
60
    except subprocess.TimeoutExpired:
61
        pass
62
63
    # The process-group leader may exit on SIGTERM while a descendant ignores
64
    # it. Probe and kill the owned group even after the leader has been reaped.
65
    if os.name == "posix":
66
        try:
67
            os.killpg(proc.pid, 0)
68
        except (OSError, ProcessLookupError):
69
            pass
70
        else:
71
            try:
72
                os.killpg(proc.pid, signal.SIGKILL)
73
            except (OSError, ProcessLookupError):
74
                pass
75
    elif proc.poll() is None:
76
        try:
77
            proc.kill()
78
        except (OSError, ProcessLookupError):
79
            pass
80
81
    if proc.poll() is None:
82
        proc.wait()
83
84
85
class _RusagePopen(subprocess.Popen):
86
    """Popen that records the child's rusage when it is reaped.
87
88
    Overriding _try_wait keeps a single wait4 syscall as both the reap and the
89
    measurement; there is no race against the kernel discarding rusage and no
90
    extra process in the timed path.
91
    """
92
93
    rusage = None
94
95
    def _try_wait(self, wait_flags):  # type: ignore[override]
96
        try:
97
            pid, sts, rusage = os.wait4(self.pid, wait_flags)
98
        except ChildProcessError:
99
            pid = self.pid
100
            sts = 0
101
        else:
102
            if pid == self.pid:
103
                self.rusage = rusage
104
        return (pid, sts)
105
106
107
def _peak_rss_bytes(rusage: Any) -> Optional[int]:
108
    if rusage is None:
109
        return None
110
    maxrss = getattr(rusage, "ru_maxrss", None)
111
    if not maxrss or maxrss < 0:
112
        return None
113
    # macOS reports bytes; Linux (and most other Unixes) report KiB.
114
    return int(maxrss) if sys.platform == "darwin" else int(maxrss) * 1024
115
116
117
def read_admitted_text(handle: Any, char_limit: int) -> tuple[str, int, bool]:
118
    handle.flush()
119
    handle.seek(0, os.SEEK_END)
120
    byte_count = handle.tell()
121
    handle.seek(0)
122
    if char_limit <= 0:
123
        return "", byte_count, byte_count > 0
124
    raw = handle.read(char_limit * 4 + 16)
125
    text = raw.decode("utf-8", "replace")[:char_limit]
126
    return text, byte_count, byte_count > len(text.encode("utf-8", "replace"))
127
128
129
def read_full_text_with_byte_limit(handle: Any, byte_limit: int) -> tuple[str | None, bool]:
130
    handle.flush()
131
    handle.seek(0, os.SEEK_END)
132
    byte_count = handle.tell()
133
    if byte_limit < 0:
134
        return None, byte_count > 0
135
    if byte_count > byte_limit:
136
        return None, True
137
    handle.seek(0)
138
    return handle.read().decode("utf-8", "replace"), False
139
140
141
def run_timed(
142
    command: list[str],
143
    cwd: Path,
144
    env: dict[str, str],
145
    admitted_output_chars: int,
146
    full_output_bytes: int | None = None,
147
    timeout_seconds: float | None = None,
148
    raw_output_bytes: int | None = None,
149
) -> TimedCapture:
150
    if timeout_seconds is not None and timeout_seconds <= 0:
151
        raise ValueError("timeout_seconds must be positive")
152
    if raw_output_bytes is not None and raw_output_bytes < 0:
153
        raise ValueError("raw_output_bytes must be nonnegative")
154
    popen_cls = _RusagePopen if _HAS_WAIT4 else subprocess.Popen
155
    start = time.perf_counter()
156
    with tempfile.TemporaryFile() as stdout_fh, tempfile.TemporaryFile() as stderr_fh:
157
        proc = popen_cls(
158
            command,
159
            cwd=cwd,
160
            env=env,
161
            stdout=stdout_fh,
162
            stderr=stderr_fh,
163
            start_new_session=timeout_seconds is not None and os.name == "posix",
164
        )
165
        timed_out = False
166
        try:
167
            returncode = proc.wait(timeout=timeout_seconds)
168
        except subprocess.TimeoutExpired:
169
            timed_out = True
170
            stop_owned_process(proc)
171
            returncode = 124
172
        elapsed_ms = (time.perf_counter() - start) * 1000
173
        stdout_text, stdout_bytes, stdout_truncated = read_admitted_text(stdout_fh, admitted_output_chars)
174
        stderr_text, stderr_bytes, stderr_truncated = read_admitted_text(stderr_fh, admitted_output_chars)
175
        stdout_raw = None
176
        if raw_output_bytes is not None and stdout_bytes <= raw_output_bytes:
177
            stdout_fh.seek(0)
178
            stdout_raw = stdout_fh.read()
179
        stdout_full_text = None
180
        stdout_full_text_truncated = None
181
        stderr_full_text = None
182
        stderr_full_text_truncated = None
183
        if full_output_bytes is not None:
184
            stdout_full_text, stdout_full_text_truncated = read_full_text_with_byte_limit(
185
                stdout_fh, full_output_bytes
186
            )
187
            stderr_full_text, stderr_full_text_truncated = read_full_text_with_byte_limit(
188
                stderr_fh, full_output_bytes
189
            )
190
    return TimedCapture(
191
        elapsed_ms=elapsed_ms,
192
        returncode=returncode,
193
        stdout_text=stdout_text,
194
        stdout_bytes=stdout_bytes,
195
        stdout_truncated=stdout_truncated,
196
        stderr_text=stderr_text,
197
        stderr_bytes=stderr_bytes,
198
        stderr_truncated=stderr_truncated,
199
        peak_rss_bytes=_peak_rss_bytes(getattr(proc, "rusage", None)),
200
        stdout_full_text=stdout_full_text,
201
        stdout_full_text_truncated=stdout_full_text_truncated,
202
        stderr_full_text=stderr_full_text,
203
        stderr_full_text_truncated=stderr_full_text_truncated,
204
        timed_out=timed_out,
205
        stdout_raw=stdout_raw,
206
    )