body_retention.py (9120B)
1 #!/usr/bin/env python3 2 """Phase-separated real-socket body-cost investigation (Linux, port 8080 free). 3 4 Pass a Nix-built v2 --runner. Each size/phase starts a 5 fresh server; repeated requests reuse that server. Exact outputs are required. 6 RSS is sampled with PID-namespace validation, including idle periods. GHCRTS=-s 7 requests allocation/maximum-live-heap statistics on orderly interrupt; missing 8 statistics remain null, never inferred from RSS. No forced GC. Upload regression 9 guards do not imply bounded whole-value downloads. 10 """ 11 import argparse 12 import hashlib 13 import json 14 import os 15 from pathlib import Path 16 import re 17 import signal 18 import subprocess 19 import threading 20 import time 21 import urllib.error 22 import urllib.request 23 24 from process_rss import ChildRSS 25 26 27 def main(): 28 parser = argparse.ArgumentParser(description=__doc__) 29 parser.add_argument('--runner', type=Path, required=True) 30 parser.add_argument('--phases', nargs='+') 31 parser.add_argument('--bounded-streaming', action='store_true', 32 help='require streaming phase RSS <=128 MiB and RTS maximum live heap <=16 MiB') 33 parser.add_argument('--directory', type=Path, required=True) 34 parser.add_argument('--sizes', type=int, nargs='+', default=[65536, 1048576]) 35 parser.add_argument('--repeats', type=int, default=3) 36 args = parser.parse_args() 37 if not 1 <= args.repeats <= 5 or any(not 1 <= n <= 1048576 for n in args.sizes): 38 parser.error('bounded sizes 1..1048576, repeats 1..5 required') 39 directory = args.directory.resolve(); directory.mkdir(parents=True, exist_ok=False) 40 opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) 41 root = Path(__file__).resolve().parent.parent 42 sources = [Path(__file__).resolve(), root/'lib/http.tri', root/'src/Runtime/Value.hs', 43 root/'lib/handlers.tri'] 44 sources += sorted((root/'src/IO').glob('*.hs')) 45 sources += sorted((root/'lib/io').glob('*.tri')) 46 sources += [root/'lib/io.tri', root/'lib/arboricx/streaming.tri', 47 root/'lib/arboricx/handlers.tri', root/'lib/handlers/artifacts.tri', 48 root/'lib/handlers/log.tri'] 49 report = dict(schema='tricu.body-retention.v1', io_version=2, rows=[], 50 source_sha256={str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in sources}, 51 ghcrts='-s', repeats=args.repeats, 52 runner_sha256={str(p.resolve()): hashlib.sha256(p.resolve().read_bytes()).hexdigest() 53 for p in [args.runner]}, 54 limits=['fresh process per phase/size, repeats share process', 55 'RSS includes heap capacity; it is not live tree size', 56 'RTS maximum residency is GC-sampled maximum live heap, not post-request live heap', 57 'publication/download include existing contracts and HTTP policy, not matched obligations', 58 'single sequential experiment, not concurrent capacity or controlled throughput']) 59 60 def request(path, data=None, status=200): 61 req = urllib.request.Request('http://127.0.0.1:8080'+path, data=data) 62 try: 63 response = opener.open(req, timeout=180) 64 except urllib.error.HTTPError as exc: 65 response = exc 66 with response: 67 assert response.status == status, (path, response.status) 68 body = response.read() 69 assert len(body) == int(response.headers['Content-Length']) 70 return body 71 72 def save(): 73 (directory/'report.json').write_text(json.dumps(report, indent=2)+'\n') 74 75 phases = [('streaming-publish', args.runner), ('streaming-download', args.runner)] 76 if args.phases: 77 unknown = set(args.phases) - {phase for phase, _ in phases} 78 if unknown: 79 parser.error('unknown/unconfigured phases: '+str(sorted(unknown))) 80 phases = [(phase, executable) for phase, executable in phases if phase in args.phases] 81 for size in args.sizes: 82 body = (bytes(range(256))*((size+255)//256))[:size] 83 digest = hashlib.sha256(body).hexdigest() 84 for phase, executable in phases: 85 work = directory/f'{size}-{phase}'; work.mkdir() 86 object_path = work/'store/objects'/digest[:3]/(digest+'.arboricx') 87 if phase.endswith('download'): 88 object_path.parent.mkdir(parents=True); object_path.write_bytes(body) 89 row = dict(phase=phase, bytes=size, runner=str(executable.resolve()), samples=[], completed=False) 90 report['rows'].append(row) 91 stop = threading.Event(); samples = []; errors = [] 92 with (work/'server.log').open('w+') as log: 93 env = dict(os.environ, GHCRTS='-s') 94 proc = subprocess.Popen([str(executable.resolve())], cwd=work, stdout=log, stderr=log, env=env) 95 meter = ChildRSS(proc) 96 started = time.monotonic() 97 def watch(): 98 try: 99 while not stop.is_set(): 100 rss = meter.sample() 101 if rss is not None: 102 samples.append((time.monotonic()-started, rss)) 103 if rss > 3*1024*1024: 104 raise RuntimeError('3 GiB RSS guard exceeded') 105 stop.wait(.02) 106 except Exception as exc: 107 errors.append(str(exc)); proc.kill() 108 monitor = threading.Thread(target=watch, daemon=True); monitor.start() 109 try: 110 for attempt in range(100): 111 if proc.poll() is not None: 112 raise AssertionError('server exited: '+str(work/'server.log')) 113 try: 114 assert request('/_arboricx/health') == b'OK\n' 115 break 116 except (OSError, urllib.error.URLError): 117 time.sleep(.1) 118 else: 119 raise AssertionError('startup timeout') 120 time.sleep(1) 121 row['ready_rss_kib'] = meter.sample() 122 for repeat in range(args.repeats): 123 begin = time.monotonic(); index = len(samples) 124 if phase.endswith('download'): 125 assert request('/_arboricx/bundle/hash/'+digest) == body 126 elif phase.endswith('publish'): 127 assert request('/_arboricx/bundle', body, 201) == digest.encode() 128 assert object_path.read_bytes() == body 129 elapsed = time.monotonic()-begin 130 peak = max((rss for _, rss in samples[index:]), default=meter.sample()) 131 idle = [] 132 for delay in (0, 1, 2): 133 time.sleep(delay); idle.append(meter.sample()) 134 row['samples'].append(dict(seconds=elapsed, peak_rss_kib=peak, 135 idle_rss_kib=idle, idle_offsets_seconds=[0, 1, 3])) 136 save() 137 assert not errors, errors 138 row['completed'] = True 139 finally: 140 proc.send_signal(signal.SIGINT) if proc.poll() is None else None 141 try: 142 proc.wait(timeout=10) 143 row['shutdown'] = 'interrupt' 144 except subprocess.TimeoutExpired: 145 proc.kill(); proc.wait(); row['shutdown'] = 'killed' 146 stop.set(); monitor.join(timeout=2) 147 row['returncode'] = proc.returncode 148 row['sampling_errors'] = errors 149 (work/'rss.json').write_text(json.dumps(samples)+'\n') 150 log.flush(); log.seek(0); output = log.read() 151 def statistic(pattern): 152 match = re.search(pattern, output) 153 return int(match[1].replace(',', '')) if match else None 154 row['allocated_bytes'] = statistic(r'([\d,]+) bytes allocated in the heap') 155 row['maximum_live_bytes'] = statistic(r'([\d,]+) bytes maximum residency') 156 row['peak_rss_kib'] = max((rss for _, rss in samples), default=None) 157 save() 158 assert not errors, errors 159 if args.bounded_streaming: 160 selected = [row for row in report['rows'] if row['phase'] in ['streaming-publish', 'streaming-download']] 161 assert selected, '--bounded-streaming needs a streaming phase' 162 for row in selected: 163 assert row['completed'] and not row['sampling_errors'], row 164 assert row['peak_rss_kib'] is not None and row['peak_rss_kib'] <= 128*1024, row 165 assert row['maximum_live_bytes'] is not None and row['maximum_live_bytes'] <= 16*1024*1024, row 166 report['bounded_streaming_checks'] = 'passed' 167 save() 168 print(json.dumps(report, indent=2)) 169 170 171 if __name__ == '__main__': 172 main()