okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauftest(stream): cover active output recovery and spill lifetimesb0282b1

main
branchesmainall refs →
10.6 KiB265 linesraw
1"""Compare exported daemons through the real gated-provider E2E harness.
2
3Run in nix develop:
4  python test/manual/active_output_benchmark.py --baseline /tmp/baseline/albedo-daemon \
5      --candidate /tmp/candidate/albedo-daemon --output /tmp/active-output-results.json
6
7Twenty attachments follow one discarded warmup per case, alternating builds.
8HTTP capture and complete reference hydration are timed separately. Throughput
9is first-to-last responsive text delivery, not provider request/boot overhead.
10RSS is sampled every 5 ms; Linux /proc is required for memory measurements.
11"""
12
13import argparse
14import json
15import os
16from pathlib import Path
17import platform
18import statistics
19import sys
20import threading
21import time
22import unittest
23
24sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "e2e"))
25import harness
26from active_output_test import ActiveOutputProbe, PausedOutput, active_text
27
28
29def summary(values):
30    ordered = sorted(values)
31    return {
32        "samples": len(values),
33        "median_ms": statistics.median(values),
34        "p95_ms": ordered[int((len(ordered) - 1) * 0.95)],
35        "raw_ms": values,
36    }
37
38
39class MemorySampler:
40    def __init__(self, pid):
41        self.pid, self.peak_kib = pid, 0
42        self.stop = threading.Event()
43        self.thread = threading.Thread(target=self.sample, daemon=True)
44
45    def sample(self):
46        while not self.stop.is_set():
47            status = Path(f"/proc/{self.pid}/status").read_text()
48            rss = next(
49                int(line.split()[1])
50                for line in status.splitlines()
51                if line.startswith("VmRSS:")
52            )
53            self.peak_kib = max(self.peak_kib, rss)
54            self.stop.wait(0.005)
55
56
57def capture(app, paused):
58    started = time.perf_counter()
59    batch = app.stream_page(paused.session)
60    captured = time.perf_counter()
61    descriptors = batch["snapshot"].get("active_output", [])
62    restored = {item["kind"]: active_text(app, item) for item in descriptors}
63    complete = time.perf_counter()
64    available = restored.get("text") == paused.prefix and (
65        not paused.thinking or restored.get("thinking") == paused.thinking
66    )
67    return ((captured - started) * 1000, (complete - started) * 1000, available)
68
69
70def delete_session(app, session):
71    resource = f"/sessions/{session}?view=configuration"
72    with app.api(resource) as response:
73        revision = response.getheader("ETag")
74        response.read()
75    app.api(resource, method="DELETE", headers={"If-Match": revision}).close()
76
77
78def main():
79    parser = argparse.ArgumentParser(description=__doc__)
80    parser.add_argument("--baseline", required=True)
81    parser.add_argument("--candidate", required=True)
82    parser.add_argument("--output", type=Path, required=True)
83    parser.add_argument("--samples", type=int, default=20)
84    parser.add_argument(
85        "--trials",
86        type=int,
87        default=5,
88        help="responsive throughput trials after warmup",
89    )
90    args = parser.parse_args()
91    if args.samples < 1 or args.trials < 1:
92        parser.error("samples and trials must be positive")
93    test = unittest.TestCase()
94    daemons, apps, samplers = {}, {}, {}
95    reports = []
96    try:
97        for label in ("baseline", "candidate"):
98            os.environ["ALBEDO_TEST_DAEMON"] = str(Path(getattr(args, label)).resolve())
99            harness.snapshot_daemon()
100            daemon = harness.Daemon("exclusive")
101            daemons[label] = daemon
102            with harness.on_daemon(daemon):
103                apps[label] = harness.Albedo(
104                    harness.Provider(lambda _: harness.Reply("text", value="unused")),
105                    prepare=lambda app: app.daemon.env.update(ALBEDO_INSPECT="1"),
106                )
107                apps[label].__enter__()
108            sampler = MemorySampler(daemon._pid)
109            samplers[label] = sampler
110            sampler.thread.start()
111
112        probes = {label: ActiveOutputProbe(app) for label, app in apps.items()}
113        other_sessions = {label: app.session() for label, app in apps.items()}
114        cases = [
115            ("small", ["x" * 256] * 64, ""),
116            ("text-and-thinking", ["x" * 256] * 64, "r" * 16384),
117            ("raw-boundary", ["x" * 256] * 256, ""),
118            ("escaped-boundary", ['"\\\n' * 128] * 100, ""),
119            ("unicode", ["α🙂" * 64] * 256, "β" * 8192),
120            ("one-MiB", ["x" * 256] * 4096, ""),
121        ]
122        for name, chunks, thinking in cases:
123            case = {
124                "name": name,
125                "chunks": len(chunks),
126                "text_bytes": len("".join(chunks).encode()),
127                "thinking_bytes": len(thinking.encode()),
128                "builds": {},
129            }
130            paused = {}
131            for label, app in apps.items():
132                paused[label] = PausedOutput(test, app, app.provider, chunks, thinking)
133                capture(app, paused[label])
134                case["builds"][label] = {
135                    "capture_ms": [],
136                    "hydrated_ms": [],
137                    "restored": [],
138                    "unrelated_session_ms": [],
139                    "responsive_delivery_ms": [],
140                    "actor_after_prefix": probes[label].call(
141                        "stats", paused[label].session
142                    ),
143                }
144            for index in range(args.samples):
145                for label in (
146                    ("baseline", "candidate")
147                    if index % 2 == 0
148                    else ("candidate", "baseline")
149                ):
150                    capture_ms, hydrated_ms, restored = capture(
151                        apps[label], paused[label]
152                    )
153                    measurements = case["builds"][label]
154                    measurements["capture_ms"].append(capture_ms)
155                    measurements["hydrated_ms"].append(hydrated_ms)
156                    measurements["restored"].append(restored)
157                    started = time.perf_counter()
158                    with apps[label].api(
159                        f"/sessions/{other_sessions[label]}?tail=0"
160                    ) as response:
161                        json.load(response)
162                    measurements["unrelated_session_ms"].append(
163                        (time.perf_counter() - started) * 1000
164                    )
165            for label, measurements in case["builds"].items():
166                measurements["capture"] = summary(measurements.pop("capture_ms"))
167                measurements["hydrated"] = summary(measurements.pop("hydrated_ms"))
168                measurements["unrelated_session"] = summary(
169                    measurements.pop("unrelated_session_ms")
170                )
171                measurements["all_attachments_restored"] = all(
172                    measurements.pop("restored")
173                )
174                spill = apps[label].home / "active-output"
175                measurements["spill_bytes_before_commit"] = sum(
176                    path.stat().st_size for path in spill.glob("*.data")
177                )
178                paused[label].finish(test)
179                measurements["spill_bytes_after_commit"] = sum(
180                    path.stat().st_size for path in spill.glob("*.data")
181                )
182                delete_session(apps[label], paused[label].session)
183                measurements["spill_bytes_after_delete"] = sum(
184                    path.stat().st_size for path in spill.glob("*.data")
185                )
186            for trial in range(args.trials):
187                for label in (
188                    ("baseline", "candidate")
189                    if trial % 2 == 0
190                    else ("candidate", "baseline")
191                ):
192                    app = apps[label]
193                    output = PausedOutput(test, app, app.provider, chunks, thinking)
194                    assert (
195                        output.prefix_at is not None
196                        and output.first_text_at is not None
197                    )
198                    case["builds"][label]["responsive_delivery_ms"].append(
199                        (output.prefix_at - output.first_text_at) * 1000
200                    )
201                    output.finish(test)
202                    delete_session(app, output.session)
203            for measurements in case["builds"].values():
204                measurements["responsive_delivery"] = summary(
205                    measurements.pop("responsive_delivery_ms")
206                )
207            reports.append(case)
208            print(
209                json.dumps(
210                    {
211                        "case": name,
212                        "builds": {
213                            label: {
214                                "capture_median_ms": values["capture"]["median_ms"],
215                                "hydrated_median_ms": values["hydrated"]["median_ms"],
216                                "delivery_median_ms": values["responsive_delivery"][
217                                    "median_ms"
218                                ],
219                                "restored": values["all_attachments_restored"],
220                                "spill_after_delete": values[
221                                    "spill_bytes_after_delete"
222                                ],
223                            }
224                            for label, values in case["builds"].items()
225                        },
226                    }
227                ),
228                flush=True,
229            )
230        args.output.write_text(
231            json.dumps(
232                {
233                    "launchers": {label: getattr(args, label) for label in apps},
234                    "method": {
235                        "platform": platform.platform(),
236                        "python": platform.python_version(),
237                        "vm_flags": harness.TEST_VM_FLAGS,
238                        "attach_samples": args.samples,
239                        "throughput_trials": args.trials,
240                        "rss_sample_ms": 5,
241                        "actor_memory": "after GC, outside timed HTTP and throughput trials",
242                        "hydrated_latency": "HTTP snapshot plus all active reference pages; baseline has no prefix",
243                    },
244                    "rss_peak_kib": {
245                        label: sampler.peak_kib for label, sampler in samplers.items()
246                    },
247                    "cases": reports,
248                },
249                indent=2,
250            )
251            + "\n"
252        )
253    finally:
254        test.doCleanups()
255        for sampler in samplers.values():
256            sampler.stop.set()
257            sampler.thread.join()
258        for app in apps.values():
259            app.provider.close()
260            app.__exit__()
261        harness.shutdown()
262
263
264if __name__ == "__main__":
265    main()