tricu

stable_costs.py (4813B)

---

 #!/usr/bin/env python3
 """Small cold-invocation cost check, reusing stable_abstraction.py artifacts.
 
 python3 test/stable_costs.py result/bin/tricu /tmp/tricu-stable
 Each sample is a new process: no repeated pure-result sharing. GC-backed
 invocation deltas exclude loading, include complete result traversal and ending
 GC. This is a feasibility check, not a warm-runtime benchmark or compiler peak
 memory measurement. Exact output equality is checked outside the timed region.
 """
 import argparse
 import json
 from pathlib import Path
 import re
 import statistics
 import subprocess
 
 ROOT = Path(__file__).resolve().parent.parent
 
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('tricu', type=Path)
     parser.add_argument('directory', type=Path)
     args = parser.parse_args()
     parser.error("Retired implicit recovery preparation: this historical workflow has no verified native-support input route. Use current portable candidate tooling; do not reuse its old support/cache artifacts.")
     tricu = str(args.tricu.resolve())
     artifacts = args.directory.resolve()
     root = artifacts / 'costs'
     root.mkdir(exist_ok=True)
     store = root / 'store'
 
     def run(command):
         result = subprocess.run([tricu, *map(str, command)], cwd=ROOT,
                                 capture_output=True, timeout=120)
         assert result.returncode == 0, (command, result.stderr)
         return result
 
     # Same algorithm and library parsers; handwritten fixed point and thunks.
     reference = root / 'reference.tri'
     reference.write_text('!import "prelude" !Local\n!import "parser" P\n' +
         'repeat_ self input parser reversed = (P.parserMatchReply\n'
         '  (problem rest consumed committed : lazyBool\n'
         '    (_ : err problem rest)\n'
         '    (_ : ok (reverse reversed) input) (or? consumed committed))\n'
         '  (value rest consumed committed : lazyBool\n'
         '    (_ : self rest parser (pair value reversed))\n'
         '    (_ : err "non-consuming parser" input) consumed)\n'
         '  (parser input))\n'
         'main input = y repeat_ input (P.parserThen P.parserAny P.parserAny) []\n')
     reference_bundle = root / 'reference.arboricx'
     run(['arboricx', 'compile', '--file', reference, '--store', artifacts / 'store',
          '--name', 'main', '--output', reference_bundle])
     stable_bundle = artifacts / 'repeat-empty.arboricx'
     if not stable_bundle.exists():
         stable_bundle = artifacts / 'repeat-pairs.arboricx'
     pins = {}
     for label, bundle in [('reference', reference_bundle), ('stable', stable_bundle)]:
         reply = run(['arboricx', 'import', '--file', bundle, '--store', store]).stdout
         pins[label] = re.search(rb'Created module alias .* -> ([0-9a-f]{64})', reply)[1].decode()
 
     results = []
     for size in [512, 2048]:
         text = 'ab' * (size // 2)
         # Independently construct the expected Result tree using literal data.
         golden = root / 'golden.tri'
         golden.write_text('!import "prelude" !Local\n' +
                           'main = ok ' + json.dumps('b' * (size // 2)) + ' []\n')
         expected = run(['eval', golden, '--store', artifacts / 'store', '--format', 'tree']).stdout
         for sample in range(3):
             for label in (['reference', 'stable'] if sample % 2 == 0 else ['stable', 'reference']):
                 reply = run(['arboricx', 'run', '--store', store, '--module', pins[label],
                     '--name', 'main', '--arg', text, '--format', 'tree', '--profile', '+RTS', '-T', '-RTS'])
                 assert reply.stdout == expected, (label, size, 'wrong output')
                 profile = [json.loads(line) for line in reply.stderr.splitlines() if line.startswith(b'{')]
                 results.append(dict(label=label, input_bytes=size, sample=sample, profile=profile))
     (root / 'samples.json').write_text(json.dumps(results, indent=2) + '\n')
     for size in [512, 2048]:
         for label in pins:
             selected = [r['profile'] for r in results if r['input_bytes'] == size and r['label'] == label]
             # Keep labels explicit: fail rather than silently time the wrong region.
             intervals = []
             for profile in selected:
                 begin = next(p for p in profile if p['profile'] == 'invoke-begin-1')
                 end = next(p for p in profile if p['profile'] == 'invoke-end-1')
                 intervals.append((end['wall_ns'] - begin['wall_ns'],
                                   end['allocated_bytes'] - begin['allocated_bytes']))
             print(f'{size} bytes {label}: median {statistics.median(i[0] for i in intervals)/1e6:.2f} ms; '
                   f'{statistics.median(i[1] for i in intervals)/1e6:.2f} MB allocated')
 
 
 if __name__ == '__main__':
     main()