diff --git a/REBUILD.md b/REBUILD.md index 8211339..5e8d563 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -41,8 +41,11 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal: Order: oldest era first (2008–2018 holds 63% of the readable files on 17% of the lines), then forward. -2. **Edge worker** (node, TypeScript reusing `lib/`; Rust later only if - CPU-bound). Per lease: compressed chunks of 10–50 k lines, in order → +2. **Edge worker** (node, TypeScript reusing `lib/`). Measured (`edge:bench`, + 2026-10-09): 73 k headers/s per core on an a.b.ath range from 2013, + 103 k/s on a recent a.b.boneless range (2 M headers each), ~1.2 GB peak + per process; the 25 B lines/day goal needs ~3–4 cores over all nodes, + so no Rust. Per lease: compressed chunks of 10–50 k lines, in order → parse subjects → skip crossposts outside their owner group (of the indexed groups in the article's Xref, the one with the lowest priority number; see the group list) → group articles into files → files diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts new file mode 100644 index 0000000..72b76dc --- /dev/null +++ b/commands/edge_bench.ts @@ -0,0 +1,130 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import { mkdir, readFile, readdir, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { unzipSync } from 'node:zlib' +import env from '#start/env' +import { XoverClient, parseOverviews } from '#lib/nntp/index' +import { RangeIndexer } from '#lib/edge/index' +import { DEFAULT_RULES, type Rules } from '#lib/rules/index' + +/** + * Measures the edge worker on a real range (REBUILD.md: TypeScript or Rust): + * --capture saves a range's compressed XOVER chunks to a directory once; + * without it the saved chunks are replayed through the range indexer, timing + * inflate, parse and indexing per chunk, with memory and output sizes. + */ +export default class EdgeBench extends BaseCommand { + static commandName = 'edge:bench' + static description = 'Measures the edge worker on a captured range' + static options: CommandOptions = { startApp: true } + + @flags.string({ description: 'Directory of the captured chunks', required: true }) + declare dir: string + + @flags.boolean({ description: 'Capture the range first (needs NNTP_* in the environment)' }) + declare capture: boolean + + @flags.string({ description: 'Group to capture' }) + declare group?: string + + @flags.number({ description: 'First article to capture' }) + declare start?: number + + @flags.number({ description: 'Articles to capture', default: 2_000_000 }) + declare articles: number + + @flags.number({ description: 'Articles per XOVER', default: 50_000 }) + declare chunk: number + + @flags.number({ description: 'Chunks without a new part before an open file is given up', default: 5 }) + declare idle: number + + @flags.string({ description: 'Indexed groups in priority order, comma-separated (owner rule)' }) + declare priorities?: string + + async run() { + if (this.capture) await this.captureRange() + await this.replay() + } + + private async captureRange() { + if (!this.group || this.start === undefined) throw new Error('--group and --start are needed to capture') + await mkdir(this.dir, { recursive: true }) + const client = await XoverClient.connect({ + host: env.get('NNTP_HOST')!, + port: env.get('NNTP_PORT')!, + user: env.get('NNTP_USER'), + password: env.get('NNTP_PASSWORD'), + secure: env.get('NNTP_SECURE', true), + compress: true, + timeoutMs: 300_000, + }) + await client.group(this.group) + const end = this.start + this.articles - 1 + let index = 0 + for (let from = this.start; from <= end; from += this.chunk, index++) { + const to = Math.min(from + this.chunk - 1, end) + const started = Date.now() + const raw = await client.xoverRaw(from, to) + await writeFile(join(this.dir, `${String(index).padStart(4, '0')}-${from}-${to}.gz`), raw) + this.logger.info(`chunk ${index}: ${from}–${to}, ${(raw.length / 1e6).toFixed(1)} MB in ${((Date.now() - started) / 1000).toFixed(1)} s`) + } + await writeFile(join(this.dir, 'range.json'), JSON.stringify({ group: this.group, start: this.start, end })) + client.close() + } + + private async replay() { + const { group, start, end } = JSON.parse(await readFile(join(this.dir, 'range.json'), 'utf8')) as { group: string; start: number; end: number } + const priorities = Object.fromEntries((this.priorities ?? group).split(',').map((g, i) => [g.trim(), i + 1])) + if (!(group in priorities)) priorities[group] = Object.keys(priorities).length + 1 + const rules: Rules = { ...DEFAULT_RULES, priorities, droppedSignatures: [] } + const files = (await readdir(this.dir)).filter((f) => f.endsWith('.gz')).sort() + const indexer = new RangeIndexer({ group, start, end, rules, idleChunks: this.idle }) + const t = { inflate: 0, parse: 0, index: 0 } + let compressed = 0 + let text = 0 + let peakRss = 0 + const out = { headers: 0, releases: 0, files: 0, segments: 0, payload: 0, candidates: 0, fragments: 0, samples: 0 } + const counts = new Map() + const cpuStart = process.cpuUsage() + const wall = Date.now() + for (const file of files) { + const raw = await readFile(join(this.dir, file)) + compressed += raw.length + let at = performance.now() + const data = unzipSync(raw).toString('utf8') + t.inflate += performance.now() - at + text += data.length + at = performance.now() + const overviews = parseOverviews(data) + t.parse += performance.now() - at + at = performance.now() + const next = Number(file.split('-')[2].replace('.gz', '')) + 1 + indexer.add(overviews, next) + if (file === files.at(-1)) indexer.finish() + const batch = indexer.take() + t.index += performance.now() - at + out.headers += batch.headers + out.releases += batch.releases.length + for (const r of batch.releases) { + out.files += r.files + out.segments += r.segments + out.payload += r.payload.length + } + out.candidates += batch.candidates.length + out.fragments += batch.fragments.length + out.samples += batch.samples.length + for (const c of batch.counts) counts.set(`${c.rule} ${c.key}`.trim(), (counts.get(`${c.rule} ${c.key}`.trim()) ?? 0) + c.articles) + peakRss = Math.max(peakRss, process.memoryUsage().rss) + } + const cpu = process.cpuUsage(cpuStart) + const cpuS = (cpu.user + cpu.system) / 1e6 + const total = t.inflate + t.parse + t.index + this.logger.info(`${group} ${start}–${end}: ${files.length} chunks, ${(compressed / 1e6).toFixed(0)} MB compressed, ${(text / 1e6).toFixed(0)} MB text`) + this.logger.info(`${out.headers.toLocaleString('en')} headers in ${(total / 1000).toFixed(1)} s (${Math.round(out.headers / (total / 1000)).toLocaleString('en')}/s on one core; CPU ${cpuS.toFixed(1)} s, wall ${((Date.now() - wall) / 1000).toFixed(1)} s)`) + this.logger.info(` inflate ${(t.inflate / 1000).toFixed(1)} s, parse ${(t.parse / 1000).toFixed(1)} s, index ${(t.index / 1000).toFixed(1)} s; peak RSS ${(peakRss / 1e6).toFixed(0)} MB`) + this.logger.info(` kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`) + for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(` dropped ${k}: ${v.toLocaleString('en')} articles`) + } +} diff --git a/start/env.ts b/start/env.ts index b1f15a7..358c134 100644 --- a/start/env.ts +++ b/start/env.ts @@ -34,4 +34,11 @@ export default await Env.create(new URL('../', import.meta.url), { DB_POOL_MAX: Env.schema.number.optional(), DB_SEARCH_PATH: Env.schema.string.optional(), DB_DEBUG: Env.schema.boolean.optional(), + + // NNTP (the coordinator's relay; edge:bench --capture) + NNTP_HOST: Env.schema.string.optional(), + NNTP_PORT: Env.schema.number.optional(), + NNTP_USER: Env.schema.string.optional(), + NNTP_PASSWORD: Env.schema.string.optional(), + NNTP_SECURE: Env.schema.boolean.optional(), })