tricu

registry_download.py (5732B)

---

 #!/usr/bin/env python3
 """Production runner: compiler download latency, native memory and error controls.
 
 Localhost observations include request parsing, authentication and socket output;
 these are regression ceilings, not predictions of production network latency.
 """
 import argparse
 import concurrent.futures
 import hashlib
 import json
 from pathlib import Path
 import signal
 import socket
 import struct
 import subprocess
 import tempfile
 import threading
 import time
 from process_rss import ChildRSS
 from rts_metrics import parse_rts_stats
 
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('runner', type=Path)
     parser.add_argument('compiler', type=Path)
     parser.add_argument('--report', type=Path, required=True)
     args = parser.parse_args()
     data = args.compiler.read_bytes()
     digest = hashlib.sha256(data).hexdigest()
     route = '/_arboricx/bundle/hash/' + digest
     rows = []
     with tempfile.TemporaryDirectory(prefix='registry-download-') as temporary:
         root = Path(temporary)
         obj = root / 'store' / 'objects' / digest[:3] / (digest + '.arboricx')
         obj.parent.mkdir(parents=True)
         obj.write_bytes(data)
         stopped = threading.Event()
         peak = [0]
         errors = []
         with (root / 'server.log').open('w+') as log:
             import os
             proc = subprocess.Popen([str(args.runner.resolve())], cwd=root, stdout=log, stderr=log,
                                     env=dict(os.environ, GHCRTS='-T -s'))
             def monitor():
                 try:
                     meter = ChildRSS(proc)
                     while not stopped.is_set():
                         value = meter.sample()
                         if value is not None:
                             peak[0] = max(peak[0], value)
                             if value > 128 * 1024:
                                 raise AssertionError('128 MiB RSS guard exceeded')
                         stopped.wait(.01)
                 except Exception as error:
                     errors.append(repr(error))
                     proc.kill()
             watcher = threading.Thread(target=monitor, daemon=True)
             watcher.start()
             def request(path=route, method='GET', status=200, body=data):
                 begin = time.monotonic()
                 with socket.create_connection(('127.0.0.1', 8080), timeout=10) as peer:
                     peer.sendall(f'{method} {path} HTTP/1.1\r\nHost: local\r\n\r\n'.encode())
                     first = peer.recv(65536)
                     first_byte = time.monotonic() - begin
                     chunks = [first]
                     while chunk := peer.recv(65536):
                         chunks.append(chunk)
                 total = time.monotonic() - begin
                 headers, received = b''.join(chunks).split(b'\r\n\r\n', 1)
                 assert int(headers.split()[1]) == status, headers
                 assert received == (b'' if method == 'HEAD' else body)
                 assert f'Content-Length: {len(body)}'.encode() in headers
                 return dict(first_byte_seconds=first_byte, total_seconds=total)
             try:
                 for _ in range(100):
                     assert proc.poll() is None, 'server exited'
                     try:
                         request('/_arboricx/health', body=b'OK\n')
                         break
                     except OSError:
                         time.sleep(.1)
                 else:
                     raise AssertionError('startup timeout')
                 for _ in range(3):
                     rows.append(request())
                 request(method='HEAD')
                 with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
                     rows.extend(pool.map(lambda _: request(), range(8)))
                 # Corruption is rejected before 200 headers; HEAD authenticates too.
                 obj.write_bytes(b'X' + data[1:])
                 request(status=404, body=b'Artifact unavailable\n')
                 request(method='HEAD', status=404, body=b'Artifact unavailable\n')
                 obj.write_bytes(data + b'x')
                 request(status=404, body=b'Artifact unavailable\n')
                 obj.write_bytes(data)
                 request('/_arboricx/bundle/hash/' + '0' * 64, status=404, body=b'Artifact unavailable\n')
                 # Reset after headers while the body may still be in flight.
                 with socket.create_connection(('127.0.0.1', 8080), timeout=10) as peer:
                     peer.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
                     peer.sendall(f'GET {route} HTTP/1.1\r\n\r\n'.encode())
                     assert peer.recv(1) == b'H'
                     peer.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
                 request('/_arboricx/health', body=b'OK\n')
                 rows.append(request())
                 assert list((root / 'store' / 'tmp').iterdir()) == []
                 assert all(row['total_seconds'] < 5 for row in rows), rows
             finally:
                 if proc.poll() is None:
                     proc.send_signal(signal.SIGINT)
                 proc.wait(timeout=15)
                 stopped.set()
                 watcher.join()
             log.seek(0)
             stats = parse_rts_stats(log.read())
             assert not errors, errors
             assert stats['maximum_live_bytes'] < 16 * 1024 * 1024, stats
         report = dict(sha256=digest, bytes=len(data), requests=rows, peak_rss_kib=peak[0], rts=stats)
         args.report.write_text(json.dumps(report, indent=2) + '\n')
         print(json.dumps(report, indent=2))
 
 
 if __name__ == '__main__':
     main()