zig_observe.py (3361B)
1 #!/usr/bin/env python3 2 """Bounded test observer: invoke a named bundle export and emit its entire tree. 3 Reads a flat argument tree and writes the complete result in the same format. 4 """ 5 import argparse 6 import ctypes as C 7 from pathlib import Path 8 import resource 9 from executable_object import FLAT, verify_payload 10 11 12 def main(): 13 p=argparse.ArgumentParser(description=__doc__) 14 for key in ('library','bundle','argument','output'): p.add_argument('--'+key,type=Path,required=True) 15 p.add_argument('--export',required=True) 16 a=p.parse_args() 17 for kind,limit in [(resource.RLIMIT_AS,512*1024**2),(resource.RLIMIT_STACK,8*1024**2), 18 (resource.RLIMIT_CPU,60),(resource.RLIMIT_CORE,0)]: 19 resource.setrlimit(kind,(limit,limit)) 20 lib=C.CDLL(str(a.library)); H=C.c_uint64; CTX=C.c_void_p 21 def bind(name,types,result=C.c_int): 22 f=getattr(lib,name); f.argtypes=types; f.restype=result; return f 23 init=bind('arboricx_init',[],CTX); free=bind('arboricx_free',[CTX],None) 24 load=bind('arb_load_bundle',[CTX,C.c_void_p,C.c_size_t,C.c_char_p,C.POINTER(H)]) 25 leaf=bind('arb_leaf',[CTX,C.POINTER(H)]) 26 stem=bind('arb_stem',[CTX,H,C.POINTER(H)]) 27 fork=bind('arb_fork',[CTX,H,H,C.POINTER(H)]) 28 app=bind('arb_app',[CTX,H,H,C.POINTER(H)]) 29 reduce=bind('arb_reduce',[CTX,H,C.c_uint64,C.POINTER(H)]) 30 release=bind('arb_release',[CTX,H]) 31 is_leaf=bind('arb_is_leaf',[CTX,H,C.POINTER(C.c_int)]) 32 is_stem=bind('arb_is_stem',[CTX,H,C.POINTER(C.c_int)]) 33 child=bind('arb_get_stem_child',[CTX,H,C.POINTER(H)]) 34 children=bind('arb_get_fork_children',[CTX,H,C.POINTER(H),C.POINTER(H)]) 35 ctx=init() 36 if not ctx: raise MemoryError('context') 37 def check(status): 38 if status: raise RuntimeError('ABI status '+str(status)) 39 def owned(f,*args): 40 out=H(); check(f(ctx,*args,C.byref(out))); return out.value 41 try: 42 data=a.argument.read_bytes(); verify_payload(FLAT,data) 43 cursor=0 44 def construct(): 45 nonlocal cursor 46 tag=data[cursor]; cursor+=1 47 args=[construct() for _ in range(tag)] 48 value=owned((leaf,stem,fork)[tag],*args) 49 for h in args: check(release(ctx,h)) 50 return value 51 argument=construct(); bundle=a.bundle.read_bytes() 52 worker=owned(load,bundle,len(bundle),a.export.encode()) 53 term=owned(app,worker,argument) 54 check(release(ctx,worker)); check(release(ctx,argument)) 55 todo=[term]; output=bytearray() 56 while todo: 57 h=todo.pop(); normalized=owned(reduce,h,10_000_000); check(release(ctx,h)); h=normalized 58 flag=C.c_int(); check(is_leaf(ctx,h,C.byref(flag))) 59 if flag.value: output.append(0) 60 else: 61 check(is_stem(ctx,h,C.byref(flag))) 62 if flag.value: 63 output.append(1); todo.append(owned(child,h)) 64 else: 65 output.append(2); left,right=H(),H() 66 check(children(ctx,h,C.byref(left),C.byref(right))) 67 todo.extend((right.value,left.value)) 68 check(release(ctx,h)) 69 if len(output)>1024*1024: raise ValueError('test output boundary') 70 verify_payload(FLAT,output) 71 a.output.write_bytes(output) 72 finally: free(ctx) 73 74 75 if __name__=='__main__': main()