diff --git a/scripts/cursor_resume_bench.py b/scripts/cursor_resume_bench.py new file mode 100644 index 00000000..9e3ec5b9 --- /dev/null +++ b/scripts/cursor_resume_bench.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +""" +Benchmark: REQ latency under concurrent ingester load. + +Measures how cursor pause/resume in the query scheduler affects +REQ response times when the ingester threads are busy. + +Seeding is done via `strfry import --no-verify` (stdin JSONL) since +the relay's websocket ingester verifies signatures. + +Usage: + # 1. Seed test data (only needed once): + python3 scripts/cursor_resume_bench.py --seed-only --seed-events 10000 + + # 2. Run the benchmark (relay must be running): + python3 scripts/cursor_resume_bench.py --relay ws://localhost:7777 + + # 3. Parse the scan perf logs: + python3 scripts/scan_perf_report.py relay.log +""" + +import argparse +import asyncio +import json +import time +import random +import hashlib +import subprocess +import sys +from dataclasses import dataclass, field + +try: + import websockets +except ImportError: + print("pip install websockets", file=sys.stderr) + sys.exit(1) + + +def make_hex(n=64): + return ''.join(random.choices('0123456789abcdef', k=n)) + + +def make_event(kind=1, pubkey=None, created_at=None): + """Generate a fake nostr event dict.""" + pk = pubkey or make_hex(64) + ts = created_at or (int(time.time()) - random.randint(0, 86400)) + content = f"bench-{random.randint(0, 999999)}" + eid = hashlib.sha256(f"{pk}{ts}{kind}{content}{random.random()}".encode()).hexdigest() + return { + "id": eid, + "pubkey": pk, + "created_at": ts, + "kind": kind, + "tags": [], + "content": content, + "sig": make_hex(128), + } + + +def seed_via_import(strfry_bin, count, kinds): + """Seed the DB using `strfry import --no-verify` (bypasses sig checks).""" + authors = [make_hex(64) for _ in range(20)] + cmd = [strfry_bin, "import", "--no-verify"] + + print(f" Seeding {count} events via `{' '.join(cmd)}`...") + proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, text=True) + + for i in range(count): + ev = make_event( + kind=random.choice(kinds), + pubkey=random.choice(authors), + ) + proc.stdin.write(json.dumps(ev) + "\n") + + proc.stdin.close() + _, stderr = proc.communicate() + if stderr: + # strfry import logs to stderr + for line in stderr.strip().split('\n'): + if line: + print(f" {line}") + print(f" Seed complete (exit {proc.returncode})") + return proc.returncode == 0 + + +@dataclass +class SweepResult: + write_rate: int = 0 + reqs: int = 0 + latencies_ms: list = field(default_factory=list) + errors: int = 0 + + def p(self, pct): + if not self.latencies_ms: + return 0.0 + s = sorted(self.latencies_ms) + return s[min(int(len(s) * pct / 100), len(s) - 1)] + + +async def reader_loop(url, duration, filters, result): + """Issue REQ queries and record EOSE latency.""" + t0 = time.monotonic() + seq = 0 + try: + async with websockets.connect(url) as ws: + while time.monotonic() - t0 < duration: + filt = random.choice(filters) + sub_id = f"b-{id(result) % 9999}-{seq}" + seq += 1 + + start = time.monotonic() + await ws.send(json.dumps(["REQ", sub_id, filt])) + + eose = False + while not eose: + left = 30.0 - (time.monotonic() - start) + if left <= 0: + break + try: + msg = await asyncio.wait_for(ws.recv(), timeout=left) + data = json.loads(msg) + if data[0] == "EOSE": + eose = True + except asyncio.TimeoutError: + break + + result.latencies_ms.append((time.monotonic() - start) * 1000) + result.reqs += 1 + await ws.send(json.dumps(["CLOSE", sub_id])) + await asyncio.sleep(0.3) + except Exception as e: + result.errors += 1 + + +async def writer_loop(url, interval, duration, result): + """Send EVENTs over websocket. These get rejected (bad sigs) but the + ingester threads still parse and verify them, competing for CPU with + the reqWorker threads that handle REQ queries.""" + if interval <= 0: + return + t0 = time.monotonic() + try: + async with websockets.connect(url) as ws: + while time.monotonic() - t0 < duration: + ev = make_event() + await ws.send(json.dumps(["EVENT", ev])) + try: + await asyncio.wait_for(ws.recv(), timeout=0.3) + except asyncio.TimeoutError: + pass + await asyncio.sleep(interval) + except Exception: + result.errors += 1 + + +async def sweep_point(url, write_rate, n_writers, n_readers, duration, filters): + result = SweepResult(write_rate=write_rate) + tasks = [] + + if write_rate > 0: + wi = max(n_writers / write_rate, 0.001) + for _ in range(n_writers): + tasks.append(writer_loop(url, wi, duration, result)) + + for _ in range(n_readers): + tasks.append(reader_loop(url, duration, filters, result)) + + await asyncio.gather(*tasks) + return result + + +async def run(args): + rates = [int(r) for r in args.rates.split(",")] + filters = [ + {"kinds": [1], "limit": 500}, + {"kinds": [1, 0, 3], "limit": 200}, + {"limit": 100}, + ] + + print(f"\nRelay: {args.relay}") + print(f"Rates: {rates}") + print(f"Readers: {args.readers}") + print(f"Duration: {args.duration}s per point\n") + + hdr = f"{'Rate':>6} {'Reqs':>5} {'p50':>7} {'p95':>7} {'p99':>7} {'max':>7} {'err':>4}" + print(hdr) + print("-" * len(hdr)) + + out = [] + for i, rate in enumerate(rates): + r = await sweep_point( + args.relay, rate, args.writers, args.readers, args.duration, filters + ) + mx = max(r.latencies_ms) if r.latencies_ms else 0 + print(f"{rate:>6} {r.reqs:>5} {r.p(50):>7.1f} {r.p(95):>7.1f} " + f"{r.p(99):>7.1f} {mx:>7.1f} {r.errors:>4}") + out.append({ + "write_rate": rate, "reqs": r.reqs, + "p50_ms": round(r.p(50), 2), "p95_ms": round(r.p(95), 2), + "p99_ms": round(r.p(99), 2), "max_ms": round(mx, 2), + "errors": r.errors, + }) + if i < len(rates) - 1: + await asyncio.sleep(2) + + with open(args.output, "w") as f: + json.dump(out, f, indent=2) + print(f"\nSaved to {args.output}") + + +def main(): + p = argparse.ArgumentParser(description="Cursor resume latency benchmark") + p.add_argument("--relay", default="ws://localhost:7777") + p.add_argument("--strfry-bin", default="./strfry", + help="Path to strfry binary (for seeding via import)") + p.add_argument("--rates", default="0,100,500,1000,2000") + p.add_argument("--seed-events", type=int, default=5000) + p.add_argument("--seed-only", action="store_true", + help="Only seed data, don't run benchmark") + p.add_argument("--skip-seed", action="store_true") + p.add_argument("--writers", type=int, default=5) + p.add_argument("--readers", type=int, default=20) + p.add_argument("--duration", type=float, default=30) + p.add_argument("--output", default="cursor_resume_results.json") + + args = p.parse_args() + + if not args.skip_seed: + seed_via_import(args.strfry_bin, args.seed_events, [1, 0, 3]) + + if args.seed_only: + return + + asyncio.run(run(args)) + + +if __name__ == "__main__": + main() diff --git a/scripts/scan_perf_report.py b/scripts/scan_perf_report.py new file mode 100644 index 00000000..e0594229 --- /dev/null +++ b/scripts/scan_perf_report.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +""" +Parse strfry scan performance logs and report cursor resume statistics. + +Reads log lines emitted when relay.logging.dbScanPerf = true, and reports: + - Per-scan-type latency distribution + - Collect time distribution + - Refill timing and frequency + +Usage: + # Pipe live logs: + ./strfry relay 2>&1 | tee relay.log + python3 scripts/scan_perf_report.py relay.log + + # Or parse existing log: + python3 scripts/scan_perf_report.py /var/log/strfry.log --output report.json +""" + +import argparse +import re +import sys +import json +from collections import defaultdict +from dataclasses import dataclass +from typing import List + + +# Match the enhanced log format with refill/collect diagnostics +LOG_PATTERN = re.compile( + r"REQ='(?P[^']+)'\s+" + r"scan=(?P\w+)\s+" + r"indexOnly=(?P\d+)\s+" + r"time=(?P\d+)us\s+" + r"saveRestores=(?P\d+)\s+" + r"recsFound=(?P\d+)\s+" + r"work=(?P\d+)" + r"(?:\s+refills=(?P\d+))?" + r"(?:\s+avgRefillUs=(?P\d+))?" + r"(?:\s+maxRefillUs=(?P\d+))?" + r"(?:\s+collects=(?P\d+))?" + r"(?:\s+avgCollectUs=(?P\d+))?" + r"(?:\s+maxCollectUs=(?P\d+))?" +) + + + +@dataclass +class ScanRecord: + sub_id: str + scan_type: str + index_only: bool + time_us: int + save_restores: int + recs_found: int + work: int + refills: int = 0 + avg_refill_us: int = 0 + max_refill_us: int = 0 + collects: int = 0 + avg_collect_us: int = 0 + max_collect_us: int = 0 + + +def parse_log(path: str) -> List[ScanRecord]: + records = [] + with open(path) as f: + for line in f: + m = LOG_PATTERN.search(line) + if not m: + continue + d = m.groupdict() + records.append(ScanRecord( + sub_id=d["sub_id"], + scan_type=d["scan_type"], + index_only=bool(int(d["index_only"])), + time_us=int(d["time_us"]), + save_restores=int(d["save_restores"]), + recs_found=int(d["recs_found"]), + work=int(d["work"]), + refills=int(d["refills"] or 0), + avg_refill_us=int(d["avg_refill_us"] or 0), + max_refill_us=int(d["max_refill_us"] or 0), + collects=int(d["collects"] or 0), + avg_collect_us=int(d["avg_collect_us"] or 0), + max_collect_us=int(d["max_collect_us"] or 0), + )) + return records + + +def percentile(values: List[float], p: int) -> float: + if not values: + return 0.0 + s = sorted(values) + idx = min(int(len(s) * p / 100), len(s) - 1) + return s[idx] + + +def report(records: List[ScanRecord]): + if not records: + print("No scan metric lines found.") + print("Ensure relay.logging.dbScanPerf = true in strfry.conf") + return + + paused = [r for r in records if r.save_restores > 0] + has_collect_data = any(r.collects > 0 for r in records) + + print(f"\n{'='*60}") + print(f" Scan Performance Report ({len(records)} queries)") + print(f"{'='*60}\n") + + print(f" Total scans: {len(records)}") + print(f" Paused scans: {len(paused)} ({100*len(paused)/len(records):.1f}%)") + + times = [r.time_us for r in records] + print(f"\n Latency (µs):") + print(f" p50: {percentile(times, 50):>10.0f}") + print(f" p95: {percentile(times, 95):>10.0f}") + print(f" p99: {percentile(times, 99):>10.0f}") + print(f" max: {max(times):>10}") + + if paused: + paused_times = [r.time_us for r in paused] + unpaused_times = [r.time_us for r in records if r.save_restores == 0] + + print(f"\n Paused vs Unpaused (µs):") + if unpaused_times: + print(f" Unpaused avg: {sum(unpaused_times)/len(unpaused_times):>10.0f}") + print(f" Paused avg: {sum(paused_times)/len(paused_times):>10.0f}") + print(f" Max restores: {max(r.save_restores for r in paused):>10}") + + # Per-scan-type breakdown + by_type = defaultdict(list) + for r in records: + by_type[r.scan_type].append(r) + + print(f"\n {'Type':<12} {'Count':>6} {'Paused':>6} {'p50µs':>9} {'p99µs':>9} {'AvgWork':>9}") + print(f" {'-'*12} {'-'*6} {'-'*6} {'-'*9} {'-'*9} {'-'*9}") + + for scan_type in sorted(by_type.keys()): + recs = by_type[scan_type] + paused_count = sum(1 for r in recs if r.save_restores > 0) + times = [r.time_us for r in recs] + avg_work = sum(r.work for r in recs) / len(recs) + print(f" {scan_type:<12} {len(recs):>6} {paused_count:>6} " + f"{percentile(times, 50):>9.0f} {percentile(times, 99):>9.0f} " + f"{avg_work:>9.0f}") + + # Collect call analysis (only with enhanced instrumentation) + if has_collect_data: + print(f"\n Collect Analysis:") + collect_records = [r for r in records if r.collects > 0] + if collect_records: + avg_collects = [r.avg_collect_us for r in collect_records] + max_collects = [r.max_collect_us for r in collect_records] + print(f" Queries with collect data: {len(collect_records)}") + print(f" Avg collect time p50: {percentile(avg_collects, 50):>8.0f} µs") + print(f" Avg collect time p99: {percentile(avg_collects, 99):>8.0f} µs") + print(f" Max collect time: {max(max_collects):>8} µs") + + refill_records = [r for r in records if r.refills > 0] + if refill_records: + print(f"\n Refill Analysis:") + print(f" Queries with refills: {len(refill_records)}") + avg_refills = [r.avg_refill_us for r in refill_records] + max_refills = [r.max_refill_us for r in refill_records] + print(f" Avg refill time p50: {percentile(avg_refills, 50):>8.0f} µs") + print(f" Avg refill time p99: {percentile(avg_refills, 99):>8.0f} µs") + print(f" Max refill time: {max(max_refills):>8} µs") + + print() + + +def main(): + parser = argparse.ArgumentParser(description="Parse strfry scan performance logs") + parser.add_argument("logfile", nargs="?", default="relay.log", + help="Path to strfry log file (default: relay.log)") + parser.add_argument("--output", "-o", default=None, + help="Write structured data to JSON file") + + args = parser.parse_args() + + try: + records = parse_log(args.logfile) + except FileNotFoundError: + print(f"Error: log file not found: {args.logfile}", file=sys.stderr) + sys.exit(1) + + report(records) + + if args.output: + data = [vars(r) for r in records] + with open(args.output, "w") as f: + json.dump(data, f, indent=2) + print(f"Structured data saved to {args.output}") + + +if __name__ == "__main__": + main() diff --git a/src/DBQuery.h b/src/DBQuery.h index e3ec1665..9a0e39c0 100644 --- a/src/DBQuery.h +++ b/src/DBQuery.h @@ -36,10 +36,16 @@ struct DBScan : NonCopyable { return resumeKey.size() > 0; } + uint64_t collectCount = 0; + uint64_t totalCollectTimeUs = 0; + uint64_t maxCollectTimeUs = 0; + uint64_t collect(lmdb::txn &txn, DBScan &s, uint64_t scanIndex, uint64_t limit, std::deque &output) { uint64_t added = 0; while (active() && limit > 0) { + uint64_t collectStart = hoytech::curr_time_us(); + bool finished = env.generic_foreachFull(txn, s.indexDbi, resumeKey, lmdb::to_sv(resumeVal), [&](auto k, auto v) { if (limit == 0) { resumeKey = std::string(k); @@ -82,6 +88,11 @@ struct DBScan : NonCopyable { return true; }, true); + uint64_t collectTimeUs = hoytech::curr_time_us() - collectStart; + totalCollectTimeUs += collectTimeUs; + if (collectTimeUs > maxCollectTimeUs) maxCollectTimeUs = collectTimeUs; + collectCount++; + if (finished) resumeKey = ""; } @@ -101,6 +112,15 @@ struct DBScan : NonCopyable { uint64_t nextInitIndex = 0; uint64_t approxWork = 0; + // Reusable buffers to avoid per-refill allocation churn + std::deque refillBuffer; + std::deque mergeBuffer; + + // Diagnostic counters for refill timing + uint64_t refillCount = 0; + uint64_t totalRefillTimeUs = 0; + uint64_t maxRefillTimeUs = 0; + DBScan(const NostrFilter &f) : f(f) { indexOnly = f.indexOnlyScans; @@ -268,12 +288,19 @@ struct DBScan : NonCopyable { cursors[ev.scanIndex()].outstanding--; if (cursors[ev.scanIndex()].outstanding == 0) { - std::deque moreEvents; - std::deque newEventQueue; - approxWork += cursors[ev.scanIndex()].collect(txn, *this, ev.scanIndex(), refillScanDepth, moreEvents); + refillBuffer.clear(); + mergeBuffer.clear(); - std::merge(eventQueue.begin(), eventQueue.end(), moreEvents.begin(), moreEvents.end(), std::back_inserter(newEventQueue), cmp); - eventQueue.swap(newEventQueue); + uint64_t refillStart = hoytech::curr_time_us(); + approxWork += cursors[ev.scanIndex()].collect(txn, *this, ev.scanIndex(), refillScanDepth, refillBuffer); + uint64_t refillTimeUs = hoytech::curr_time_us() - refillStart; + + totalRefillTimeUs += refillTimeUs; + if (refillTimeUs > maxRefillTimeUs) maxRefillTimeUs = refillTimeUs; + refillCount++; + + std::merge(eventQueue.begin(), eventQueue.end(), refillBuffer.begin(), refillBuffer.end(), std::back_inserter(mergeBuffer), cmp); + eventQueue.swap(mergeBuffer); } } } @@ -339,13 +366,29 @@ struct DBQuery : NonCopyable { totalWork += scanner->approxWork; if (logMetrics) { + // Aggregate cursor-level collect diagnostics + uint64_t totalCollects = 0; + uint64_t totalCollectTimeUs = 0; + uint64_t maxCollectTimeUs = 0; + for (const auto &c : scanner->cursors) { + totalCollects += c.collectCount; + totalCollectTimeUs += c.totalCollectTimeUs; + if (c.maxCollectTimeUs > maxCollectTimeUs) maxCollectTimeUs = c.maxCollectTimeUs; + } + LI << "[" << sub.connId << "] REQ='" << sub.subId.sv() << "'" << " scan=" << scanner->desc << " indexOnly=" << scanner->indexOnly << " time=" << currScanTime << "us" << " saveRestores=" << currScanSaveRestores << " recsFound=" << sentEventsCurr.size() - << " work=" << scanner->approxWork; + << " work=" << scanner->approxWork + << " refills=" << scanner->refillCount + << " avgRefillUs=" << (scanner->refillCount ? scanner->totalRefillTimeUs / scanner->refillCount : 0) + << " maxRefillUs=" << scanner->maxRefillTimeUs + << " collects=" << totalCollects + << " avgCollectUs=" << (totalCollects ? totalCollectTimeUs / totalCollects : 0) + << " maxCollectUs=" << maxCollectTimeUs; ; }