io_v2_streams.py (5225B)
1 #!/usr/bin/env python3 2 """Real binary pipes/redirects, ownership errors, and bounded native residency.""" 3 import argparse 4 import concurrent.futures 5 import hashlib 6 import json 7 import os 8 from pathlib import Path 9 import subprocess 10 import time 11 from source_order import Runner 12 from process_rss import ChildRSS 13 from rts_metrics import parse_rts_stats 14 from http_headers import result, tree_list 15 from stable_bundle import byte_tree 16 17 ROOT = Path(__file__).resolve().parent.parent 18 19 def main(): 20 parser = argparse.ArgumentParser(description=__doc__) 21 parser.add_argument('tricu', type=Path) 22 parser.add_argument('--directory', type=Path, required=True) 23 args = parser.parse_args() 24 directory = args.directory.resolve() 25 directory.mkdir(parents=True, exist_ok=False) 26 runner = Runner(args.tricu, directory) 27 source = ROOT/'demos/io-v2/streams.tri' 28 bundle = directory/'streams.arboricx' 29 runner.run('compile', ['compile', '--file', source, '--store', runner.store, '--output', bundle]) 30 pin = runner.imported(bundle, 'import') 31 checks, observations = [], [] 32 33 def command(name, stats=True): 34 return [str(runner.exe), 'arboricx', 'run', '--module', pin, '--name', name, 35 '--store', str(runner.store), '--io', '--no-result', 36 '+RTS', '-N1', '-M128m', *(['-T', '-s'] if stats else []), '-RTS'] 37 38 def invoke(label, name, body, redirected=False): 39 argv = command(name) 40 started = time.monotonic() 41 input_path, output_path = directory/(label+'.in'), directory/(label+'.out') 42 input_path.write_bytes(body) 43 with input_path.open('rb') as src, output_path.open('wb') as dst: 44 proc = subprocess.Popen(argv, stdin=src if redirected else subprocess.PIPE, 45 stdout=dst if redirected else subprocess.PIPE, stderr=subprocess.PIPE) 46 peak = None 47 try: 48 rss = ChildRSS(proc) 49 with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: 50 future = pool.submit(proc.communicate, None if redirected else body, timeout=120) 51 while not future.done(): 52 current = rss.sample() 53 if current is not None: 54 peak = max(peak or 0, current) 55 if peak > 128*1024: 56 proc.kill() 57 raise AssertionError('RSS bound exceeded') 58 time.sleep(.01) 59 stdout, stderr = future.result() 60 finally: 61 if proc.poll() is None: 62 proc.kill(); proc.communicate() 63 assert proc.returncode == 0, stderr 64 if redirected: 65 stdout = output_path.read_bytes() 66 else: 67 output_path.write_bytes(stdout) 68 assert stdout == body, label 69 text = stderr.decode() 70 (directory/(label+'.log')).write_text(text) 71 stats = parse_rts_stats(text) 72 assert stats['maximum_live_bytes'] <= 16*1024*1024, stats 73 assert peak is not None, 'RSS unavailable' 74 observations.append(dict(label=label, command=argv, seconds=time.monotonic()-started, 75 peak_kib=peak, rts_process=stats, sha256=hashlib.sha256(stdout).hexdigest())) 76 checks.append(label) 77 78 for name in ['main', 'handled', 'child']: 79 for size in [0, 1, 4095, 4096, 4097, 65536, 1048576]: 80 body = (bytes(range(256))*((size+255)//256))[:size] 81 invoke(f'{name}-{size}', name, body) 82 invoke('redirected', 'main', bytes(range(256))*257, True) 83 for name, operation in [('closeDenied', 'resource/close'), ('transferDenied', 'core/spawn')]: 84 expected = result(False, tree_list([byte_tree(x.encode()) for x in 85 ['borrowed-resource', operation, '', 'none']])) 86 assert runner.invoke(name, pin, name, ['--io']) == expected, name 87 checks.append(name) 88 run = subprocess.run(command('diagnostic', False), input=b'', capture_output=True, timeout=20) 89 assert (run.returncode, run.stdout, run.stderr) == (0, b'', bytes([0,255,13,10])), run 90 checks.append('binary-stderr') 91 reader, writer = os.pipe() 92 os.close(reader) 93 try: 94 proc = subprocess.Popen(command('broken', False), stdin=subprocess.DEVNULL, 95 stdout=writer, stderr=subprocess.PIPE) 96 finally: 97 os.close(writer) 98 _, diagnostic = proc.communicate(timeout=20) 99 assert (proc.returncode, diagnostic) == (0, b'failed'), (proc.returncode, diagnostic) 100 checks.append('broken-pipe-ordinary-result') 101 paths = [source, Path(__file__).resolve(), ROOT/'lib/io-v2.tri', 102 ROOT/'src/IOV2/Resources.hs', ROOT/'src/IOV2/Driver.hs', ROOT/'src/IOV2/Protocol.hs', 103 ROOT/'test/rts_metrics.py', ROOT/'bootstrap/build-stable.arboricx'] 104 (directory/'report.json').write_text(json.dumps(dict(checks=checks, observations=observations, 105 sources={str(p.relative_to(ROOT)): hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}), indent=2)+'\n') 106 print(f'PASS: {len(checks)} binary stream, lifecycle and pipe checks') 107 108 if __name__ == '__main__': 109 main()