zig_generation_executor.py (6886B)
1 #!/usr/bin/env python3 2 """Zig execution child for bootstrap_generations. 3 4 The supervisor authenticates/imports captured bundles, supplies six arguments 5 and independently audits all publications. This child rechecks captured bundle 6 digests and runs the selected entrypoint through the owned ABI. It grants broad 7 file IO to trusted compiler jobs, matching the generation runner's unsafe-IO scope. 8 """ 9 import argparse 10 import ctypes as C 11 import hashlib 12 import json 13 from pathlib import Path 14 import resource 15 import time 16 17 18 def save(path, value): 19 path.write_text(json.dumps(value, indent=2) + '\n') 20 21 22 def child(args): 23 # No asynchronous subprocesses in the Zig IO driver; the parent additionally 24 # guards this whole process group with a wall deadline and RSS sampling. 25 resource.setrlimit(resource.RLIMIT_AS, (args.memory_mib * 1024**2,) * 2) 26 resource.setrlimit(resource.RLIMIT_STACK, (8 * 1024**2,) * 2) 27 resource.setrlimit(resource.RLIMIT_CPU, (args.seconds, args.seconds)) 28 resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) 29 lib = C.CDLL(str(args.library)) 30 handle, ctx_t = C.c_uint64, C.c_void_p 31 def bind(name, types, result=C.c_int): 32 f = getattr(lib, name) 33 f.argtypes, f.restype = types, result 34 return f 35 init = bind('arboricx_init', [], ctx_t) 36 free = bind('arboricx_free', [ctx_t], None) 37 version = bind('arb_abi_version', [], C.c_uint32) 38 assert version() == 2 39 load = bind('arb_load_bundle', [ctx_t, C.c_void_p, C.c_size_t, C.c_char_p, C.POINTER(handle)]) 40 string = bind('arb_of_string', [ctx_t, C.c_char_p, C.POINTER(handle)]) 41 app = bind('arb_app', [ctx_t, handle, handle, C.POINTER(handle)]) 42 release = bind('arb_release', [ctx_t, handle]) 43 reduce = bind('arb_reduce', [ctx_t, handle, C.c_uint64, C.POINTER(handle)]) 44 unwrap = bind('arb_unwrap_result', [ctx_t, handle, C.POINTER(C.c_int), C.POINTER(handle), C.POINTER(handle)]) 45 to_string = bind('arb_to_string', [ctx_t, handle, C.POINTER(C.c_void_p), C.POINTER(C.c_size_t)]) 46 number = bind('arb_to_number', [ctx_t, handle, C.POINTER(C.c_uint64)]) 47 pair = bind('arb_get_fork_children', [ctx_t, handle, C.POINTER(handle), C.POINTER(handle)]) 48 free_buf = bind('arboricx_free_buf', [ctx_t, C.c_void_p, C.c_size_t], None) 49 class Perms(C.Structure): 50 _fields_ = [('read', C.c_int), ('write', C.c_int)] 51 run_io = bind('arb_run_io', [ctx_t, handle, C.POINTER(Perms), C.POINTER(handle)]) 52 def check(status): 53 if status != 0: 54 raise RuntimeError(f'ABI status {status}') 55 ctx = init() 56 if not ctx: 57 raise MemoryError('context initialization') 58 def owned(f, *arguments): 59 out = handle() 60 check(f(ctx, *arguments, C.byref(out))) 61 return out.value 62 def apply(f, x): 63 result = owned(app, f, x) 64 check(release(ctx, f)); check(release(ctx, x)) 65 return result 66 run = args.output 67 request = json.loads((run / 'execution-request.json').read_text()) 68 started = time.monotonic() 69 try: 70 if args.gc_policy: 71 configure = bind('arb_set_gc_policy', [ctx_t, C.c_uint32, C.c_uint32]) 72 check(configure(ctx, *args.gc_policy)) 73 compiler_bytes = (run / 'compiler.arboricx').read_bytes() 74 adapter_bytes = (run / 'io-adapter.arboricx').read_bytes() 75 if hashlib.sha256(compiler_bytes).hexdigest() != request['compiler_bundle_sha256'] or hashlib.sha256(adapter_bytes).hexdigest() != request['adapter_sha256']: 76 raise ValueError('captured executable identity mismatch') 77 compiler = owned(load, compiler_bytes, len(compiler_bytes), request['worker_export'].encode()) 78 adapter = owned(load, adapter_bytes, len(adapter_bytes), b'main') 79 current = apply(adapter, compiler) 80 for argument in request['arguments']: 81 if '\0' in argument: 82 raise ValueError('NUL in compiler argument') 83 current = apply(current, owned(string, argument.encode())) 84 prepared = time.monotonic() 85 program = owned(reduce, current, 1_000_000_000) 86 check(release(ctx, current)) 87 forced = time.monotonic() 88 result = owned(run_io, program, C.byref(Perms(1, 1))) 89 check(release(ctx, program)) 90 ran = time.monotonic() 91 ok, value, rest = C.c_int(), handle(), handle() 92 check(unwrap(ctx, result, C.byref(ok), C.byref(value), C.byref(rest))) 93 diagnostic = None 94 if not ok.value: 95 ptr, length = C.c_void_p(), C.c_size_t() 96 status = to_string(ctx, value, C.byref(ptr), C.byref(length)) 97 if status == 0: 98 diagnostic = C.string_at(ptr, length.value).decode('utf-8', 'replace') 99 free_buf(ctx, ptr, length.value) 100 else: 101 # Preserve a useful bounded prefix of structured diagnostics. 102 # This recognizes only a generic list prefix, not compiler tags. 103 prefix, tail = handle(), handle() 104 tag = C.c_uint64() 105 normalized = owned(reduce, value.value, 1_000_000) 106 if pair(ctx, normalized, C.byref(prefix), C.byref(tail)) == 0: 107 if number(ctx, prefix, C.byref(tag)) == 0: 108 diagnostic = dict(first_number=tag.value) 109 check(release(ctx, prefix)) 110 check(release(ctx, tail)) 111 check(release(ctx, normalized)) 112 if diagnostic is None: 113 diagnostic = f'non-string diagnostic (decode status {status})' 114 record = dict(ok=bool(ok.value), diagnostic=diagnostic, 115 prepare_seconds=prepared-started, initial_reduce_seconds=forced-prepared, 116 io_seconds=ran-forced, total_seconds=time.monotonic()-started) 117 if args.gc_policy: 118 record['gc_policy'] = args.gc_policy 119 save(run / 'execution.json', record) 120 if not ok.value: 121 raise RuntimeError(diagnostic) 122 print('ok: Zig map-builder completed', flush=True) 123 print(json.dumps(record), flush=True) 124 finally: 125 free(ctx) 126 127 128 if __name__ == '__main__': 129 parser = argparse.ArgumentParser(description=__doc__) 130 parser.add_argument('--library', type=Path, required=True) 131 parser.add_argument('--output', type=Path, required=True) 132 parser.add_argument('--memory-mib', type=int, required=True) 133 parser.add_argument('--gc-policy', type=int, nargs=2, metavar=('MIN_ALLOCATIONS', 'LIVE_MULTIPLIER')) 134 parser.add_argument('--seconds', type=int, required=True) 135 args = parser.parse_args() 136 if not 0 < args.memory_mib <= 65536 or not 0 < args.seconds <= 7200: 137 parser.error('invalid resource boundaries') 138 if args.gc_policy and any(not 0 < n <= 0xffffffff for n in args.gc_policy): 139 parser.error('GC policy requires positive u32 values') 140 child(args)