| from pathlib import Path |
| import sys,time,json,platform,statistics,random,os,math,resource,tempfile |
| ROOT=Path(__file__).resolve().parents[1];sys.path.insert(0,str(ROOT)) |
| from phillnet_recall import RecallMemory,RecallPolicy |
|
|
| def main(): |
| rng=random.Random(781);policy=RecallPolicy();rows=[] |
| for n in (1000,10000): |
| with tempfile.TemporaryDirectory() as td: |
| path=Path(td)/'scale.sqlite' |
| with RecallMemory(path,policy=policy,clock=lambda:100000.,max_candidates=2048) as m: |
| t=time.perf_counter() |
| for i in range(n):m.remember(f'Unit{i:06} deployment service port is {10000+i}.',source=f'session:{i//10}',entity=f'Unit{i:06}',predicate='port',value=str(10000+i)) |
| ingest=time.perf_counter()-t;times=[];success=0;candidate_counts=[] |
| for i in rng.sample(range(n),100): |
| t=time.perf_counter_ns();r=m.retrieve(f'Unit{i:06}',top_k=1);times.append((time.perf_counter_ns()-t)/1e6) |
| success+=bool(r['items'] and str(10000+i) in r['items'][0]['text']);candidate_counts.append(r['trace']['candidate_count']) |
| overflow=m.retrieve('deployment service port',top_k=8) |
| m.db.execute('PRAGMA wal_checkpoint(TRUNCATE)') |
| rows.append({'entries':n,'ingest_seconds':ingest,'queries':100,'correct':success,'query_family':'unique identifier; favorable lexical selectivity', |
| 'candidate_min':min(candidate_counts),'candidate_max':max(candidate_counts),'median_ms':statistics.median(times),'p95_ms':sorted(times)[94], |
| 'generic_query_status':overflow['status'],'database_bytes':path.stat().st_size,'accounted_payload_bytes':m.stats()['accounted_payload_bytes']}) |
| result={'environment':{'python':sys.version,'platform':platform.platform(),'cpu_count':os.cpu_count(),'native':bool(policy.bank._lib),'max_rss_kib_linux':resource.getrusage(resource.RUSAGE_SELF).ru_maxrss},'rows':rows,'note':'Unreserved shared CPU, one process, 100 queries per size. No model-generation cost. Does not establish arbitrary-query latency or million-row scalability.'} |
| (ROOT/'reports/scale.json').write_text(json.dumps(result,indent=2)+'\n');print(json.dumps(result,indent=2)) |
| if __name__=='__main__':main() |
|
|
