test_embedding.py (10858B)
1 #!/usr/bin/env python3 2 """Fresh-process ctypes ABI checks. Optional compiled copier exercises real IO.""" 3 import argparse 4 import ctypes as C 5 import concurrent.futures 6 import json 7 from pathlib import Path 8 import tempfile 9 import struct 10 import time 11 from tricu import Library, TricuError, U64, U32, PTR, SIZE 12 13 14 def expect(status, operation): 15 try: 16 operation() 17 except TricuError as e: 18 assert e.status == status, e 19 else: 20 raise AssertionError(f'expected status {status}') 21 22 23 def envelope(ctx, action, version=2): 24 return ctx.fork(ctx.string('tricuIO'), ctx.fork(ctx.number(version), action)) 25 26 27 def checks(lib, fixtures): 28 native = lib.native 29 assert native.tricu_context_new(None) == 3 30 with lib.context() as ctx, lib.context() as other: 31 assert ctx.identity != other.identity 32 leaf = ctx.leaf() 33 stem = ctx.stem(leaf) 34 fork = ctx.fork(leaf, stem) 35 assert [ctx.kind(x) for x in (leaf, stem, fork)] == [0, 1, 2] 36 child = ctx.child(fork, 1) 37 ctx.release(fork) 38 assert ctx.kind(child) == 1 39 alias = ctx.retain(child) 40 assert alias != child 41 ctx.release(child) 42 expect(2, lambda: ctx.kind(child)) 43 expect(2, lambda: other.kind(alias)) 44 expect(4, lambda: ctx.child(leaf, 0)) 45 expect(4, lambda: ctx.to_number(stem)) 46 expect(4, lambda: ctx.to_bytes(stem)) 47 expect(2, lambda: ctx.release(0)) 48 assert ctx.kind(alias) == 1 49 for value in (0, 1, 42, 2**64-1): 50 assert ctx.to_number(ctx.number(value)) == value 51 # This raw tree is 2**64, exceeding the uint64 upper bound. 52 huge = ctx.number(1) 53 for _ in range(64): 54 huge = ctx.fork(leaf, huge) 55 expect(3, lambda: ctx.to_number(huge)) 56 for body in (b'', b'\x00a\xff\x80', bytes(range(256))*16): 57 assert ctx.to_bytes(ctx.bytes(body)) == body 58 for text in ('', 'hello\x00world', 'tree 🌳 café\r\n\ufeff'): 59 assert ctx.to_string(ctx.string(text)) == text 60 for bad in (b'\xff', b'\xc0\x80', b'\xed\xa0\x80', b'\xf4\x90\x80\x80'): 61 output = U64(12345) 62 assert native.tricu_from_utf8(ctx.identity, bad, len(bad), C.byref(output)) == 3 63 assert output.value == 12345 64 # Invalid scalar value inside an otherwise proper string. 65 surrogate = ctx.fork(ctx.number(0xd800), leaf) 66 expect(4, lambda: ctx.to_string(surrogate)) 67 expect(3, lambda: ctx.load_bundle(b'bad bundle', 'id')) 68 data = (fixtures/'id.arboricx').read_bytes() 69 expect(3, lambda: ctx.load_bundle(data, 'missing')) 70 corrupt = bytearray(data) 71 node_offset = struct.unpack('>IHHIQQI', data[64:96])[4] 72 corrupt[node_offset+12] = 255 73 expect(3, lambda: ctx.load_bundle(bytes(corrupt), 'id')) 74 worker = ctx.load_bundle(data, 'id') 75 value = ctx.string('from a portable bundle') 76 application = ctx.apply(worker, value) 77 ctx.release(worker) 78 ctx.release(value) 79 assert ctx.to_string(ctx.reduce(application)) == 'from a portable bundle' 80 append = ctx.load_bundle((fixtures/'append.arboricx').read_bytes(), 'append') 81 assert ctx.to_bytes(ctx.apply(ctx.apply(append, ctx.bytes(b'abc')), ctx.bytes(b'def'))) == b'abcdef' 82 for code, outcome in ((0, 0), (2, 1), (9, 2)): 83 program = envelope(ctx, ctx.fork(ctx.number(code), ctx.string('payload'))) 84 actual, value, cleanup = ctx.run_io(program) 85 assert actual == outcome and ctx.to_string(value) == 'payload' 86 assert ctx.kind(cleanup) == 0 87 expect(3, lambda: ctx.run_io(leaf)) 88 expect(3, lambda: ctx.run_io(envelope(ctx, leaf, version=1))) 89 # A malformed action produces a task-level failure. 90 outcome, diagnostic, cleanup = ctx.run_io(envelope(ctx, leaf)) 91 assert outcome == 1 and ctx.kind(diagnostic) == 2 and ctx.kind(cleanup) == 0 92 # IO outputs are checked before execution and all remain unchanged. 93 outcome, value, cleanup = U32(999), U64(998), U64(997) 94 assert native.tricu_run_io(ctx.identity, leaf, C.byref(outcome), C.byref(value), None) == 3 95 assert (outcome.value, value.value, cleanup.value) == (999, 998, 997) 96 assert native.tricu_run_io(ctx.identity, leaf, C.byref(outcome), C.byref(value), C.byref(value)) == 3 97 assert (outcome.value, value.value) == (999, 998) 98 # Failed calls preserve both buffer outputs, too. 99 ptr, size = PTR(123), SIZE(456) 100 assert native.tricu_to_bytes(ctx.identity, stem, C.byref(ptr), C.byref(size)) == 4 101 assert ptr.value == 123 and size.value == 456 102 assert native.tricu_from_bytes(ctx.identity, None, 1, C.byref(value)) == 3 103 assert value.value == 998 104 # Foreign buffers are independent of context roots and context lifetime. 105 assert native.tricu_to_utf8(ctx.identity, alias, C.byref(ptr), C.byref(size)) == 4 106 text = ctx.string('survives destruction') 107 assert native.tricu_to_utf8(ctx.identity, text, C.byref(ptr), C.byref(size)) == 0 108 old = ctx.identity 109 ctx.close() 110 try: 111 assert C.string_at(ptr, size.value) == b'survives destruction' 112 finally: 113 native.tricu_buffer_free(ptr) 114 assert native.tricu_context_free(old) == 1 115 assert native.tricu_leaf(old, C.byref(value)) == 1 and value.value == 998 116 for _ in range(100): 117 with lib.context() as ctx: 118 assert ctx.to_number(ctx.number(42)) == 42 119 # Calls release Python's GIL; both context isolation and same-context 120 # serialization are exercised with concurrent native entry. 121 def job(shared=None): 122 if shared is not None: 123 for _ in range(100): 124 handle = shared.number(17) 125 assert shared.to_number(handle) == 17 126 shared.release(handle) 127 else: 128 with lib.context() as context: 129 job(context) 130 with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: 131 list(pool.map(lambda _: job(), range(8))) 132 with lib.context() as ctx: 133 list(pool.map(lambda _: job(ctx), range(8))) 134 135 136 def copier(lib, bundle, body=None): 137 with tempfile.TemporaryDirectory(prefix='tricu-ffi-') as directory: 138 source, target = Path(directory)/'in', Path(directory)/'out' 139 if body is None: 140 body = bytes(range(256))*256 141 source.write_bytes(body) 142 with lib.context() as ctx: 143 worker = ctx.load_bundle(bundle.read_bytes(), 'copy') 144 descriptors = None 145 for _ in range(3): 146 program = ctx.apply(ctx.apply(worker, ctx.string(str(source))), ctx.string(str(target))) 147 outcome, result, cleanup = ctx.run_io(program) 148 assert outcome == 0 and ctx.kind(cleanup) == 0 149 # copy returns an ordinary successful Result wrapping unit. 150 assert ctx.kind(ctx.child(result, 0)) == 1 151 assert target.read_bytes() == body 152 if Path('/proc/self/fd').is_dir(): 153 count = len(list(Path('/proc/self/fd').iterdir())) 154 if descriptors is not None: 155 assert count == descriptors, 'file descriptors retained across IO runs' 156 descriptors = count 157 ctx.release(program) 158 ctx.release(result) 159 ctx.release(cleanup) 160 # Structured child cancellation settles transferred reader/writer 161 # resources before returning the child's outcome as ordinary data. 162 cancelled = ctx.load_bundle(bundle.read_bytes(), 'cancelledCopy') 163 program = ctx.apply(ctx.apply(cancelled, ctx.string(str(source))), ctx.string(str(target))) 164 outcome, result, cleanup = ctx.run_io(program) 165 assert outcome == 0 and ctx.kind(cleanup) == 0 166 assert ctx.to_string(ctx.child(result, 0)) == 'cancelled' 167 rest = ctx.child(result, 1) 168 assert ctx.to_string(ctx.child(rest, 0)) == 'requested' 169 assert target.read_bytes() == body[:4096] 170 # Reader acquisition succeeds, then writer acquisition fails. Scope 171 # cleanup must release the acquired reader before returning. 172 missing = Path(directory)/'missing-parent'/'out' 173 program = ctx.apply(ctx.apply(worker, ctx.string(str(source))), ctx.string(str(missing))) 174 outcome, result, cleanup = ctx.run_io(program) 175 assert outcome == 0 and ctx.kind(ctx.child(result, 0)) == 0 176 assert not missing.exists() 177 if descriptors is not None: 178 assert len(list(Path('/proc/self/fd').iterdir())) == descriptors 179 source.unlink() 180 target.unlink() 181 program = ctx.apply(ctx.apply(worker, ctx.string(str(source))), ctx.string(str(target))) 182 outcome, result, cleanup = ctx.run_io(program) 183 assert outcome == 0 and ctx.kind(ctx.child(result, 0)) == 0 184 assert not target.exists() 185 # Reopen/rename/delete after successful and failed runs; resources settle 186 # before run_io returns, independently of context destruction. 187 target.write_bytes(b'again') 188 target.rename(source) 189 assert source.read_bytes() == b'again' 190 191 192 def measurements(lib, fixtures): 193 report = {} 194 with lib.context() as ctx: 195 start = time.perf_counter() 196 for _ in range(10000): 197 root = ctx.number(42) 198 assert ctx.to_number(root) == 42 199 ctx.release(root) 200 report['small_roundtrips_10000_seconds'] = time.perf_counter()-start 201 worker = ctx.load_bundle((fixtures/'id.arboricx').read_bytes(), 'id') 202 body = bytes(range(256))*256 203 start = time.perf_counter() 204 for _ in range(3): 205 value = ctx.bytes(body) 206 application = ctx.apply(worker, value) 207 result = ctx.reduce(application) 208 assert ctx.to_bytes(result) == body 209 for handle in (value, application, result): 210 ctx.release(handle) 211 report['bytes_identity_65536_x3_seconds'] = time.perf_counter()-start 212 return report 213 214 215 def main(): 216 parser = argparse.ArgumentParser(description=__doc__) 217 parser.add_argument('library', type=Path) 218 parser.add_argument('fixtures', type=Path) 219 parser.add_argument('--copy-bundle', type=Path) 220 args = parser.parse_args() 221 lib = Library(args.library) 222 checks(lib, args.fixtures) 223 if args.copy_bundle: 224 copier(lib, args.copy_bundle) 225 print(json.dumps({'checks': 'passed', 'copier': bool(args.copy_bundle), 226 'measurements': measurements(lib, args.fixtures)}, sort_keys=True)) 227 228 229 if __name__ == '__main__': 230 main()