Aleksey Shakhmatov
d4c9b04f21
Prose and benchmark tables use MultiforaDB; the binary, the CLI usage line, the log-message prefix and the default database file use multiforadb. Two consequences worth noting: - build.zig.zon's fingerprint is derived from the package name, so it had to change with it (Zig refuses to build otherwise). A consumer pinning this package by fingerprint needs updating. - the default --db path is now multiforadb.log, and getCmdLineOpts reports it as dbpath. An existing mongo-lite.log has to be passed explicitly with --db. The e2e harness abbreviated the old name as ML_; that is now MFDB_, including the documented ML_BIN override (MFDB_BIN) and the scratch file names. MD_ (mongod) is untouched. compare-run.sh spawned the server by absolute path under a sandbox/mongo-lite directory that no longer exists; that block already runs from tests/e2e, so it uses a relative path now. The archived reports under tests/e2e/results/ keep the old name: they record what the old binary measured.
348 lines
16 KiB
JavaScript
348 lines
16 KiB
JavaScript
// Big-collection harness: how multiforadb behaves with multi-GB collections.
|
|
//
|
|
// Spawns its own server, bulk-inserts up to ~5 GB of documents, measures
|
|
// insert throughput, log/compaction behavior and server RSS, benchmarks
|
|
// find/count/update/delete against the full dataset, then kills the server
|
|
// and measures reopen (replay) time and crash durability.
|
|
//
|
|
// node tests/e2e/big.js [options]
|
|
// --size <n> target collection size; k/m/g suffixes (default 5g)
|
|
// --doc-size <n> approximate bytes per document (default 32k)
|
|
// --batch <n> docs per insertMany call (default 500)
|
|
// --index <f> create a secondary index on field f *before* inserting
|
|
// (entry insert is O(n), so this makes the load quadratic)
|
|
// --oid use ObjectId _ids (driver-generated): O(1) _id lookups.
|
|
// Without it, int _ids fall back to a full scan (the _id
|
|
// fast path is skipped for serialization-ambiguous
|
|
// numeric classes), so findOne({_id}) costs a scan.
|
|
// --compact-threshold <n>
|
|
// pass through to the server: log bytes between
|
|
// compactions (default 16m). Raise for bulk loads.
|
|
// --port <n> server port (default 27221)
|
|
// --keep keep the db file after the run
|
|
// --quick tiny run (256m, 16k docs)
|
|
//
|
|
// Env: MFDB_BIN server binary (default ../../zig-out/bin/multiforadb)
|
|
// BIG_DB db file path (default .zig-cache/big.log)
|
|
const { MongoClient } = require('mongodb');
|
|
const { spawn } = require('child_process');
|
|
const fs = require('fs');
|
|
const path = require('path');
|
|
|
|
const BIN = process.env.MFDB_BIN || path.resolve(__dirname, '../../zig-out/bin/multiforadb');
|
|
const PORT = Number(process.env.BIG_PORT || 27221);
|
|
const DBFILE = process.env.BIG_DB || path.resolve(__dirname, '../../.zig-cache/big.log');
|
|
const URL = `mongodb://127.0.0.1:${PORT}`;
|
|
|
|
function parseSize(s) {
|
|
const m = /^(\d+(?:\.\d+)?)([kmgt]?)$/i.exec(String(s).trim());
|
|
if (!m) throw new Error(`bad size '${s}'`);
|
|
const mult = { '': 1, k: 1 << 10, m: 1 << 20, g: 1 << 30, t: 1 << 40 }[m[2].toLowerCase()];
|
|
return Math.round(parseFloat(m[1]) * mult);
|
|
}
|
|
|
|
let opt = { size: '5g', docSize: '32k', batch: 500, index: null, port: PORT, keep: false, oid: false, compactThreshold: null };
|
|
for (let i = 2; i < process.argv.length; i++) {
|
|
const a = process.argv[i];
|
|
if (a === '--quick') { opt.size = '256m'; opt.docSize = '16k'; }
|
|
else if (a === '--keep') opt.keep = true;
|
|
else if (a === '--oid') opt.oid = true;
|
|
else if (a === '--index') opt.index = process.argv[++i];
|
|
else if (a.startsWith('--size=')) opt.size = a.slice(7);
|
|
else if (a.startsWith('--doc-size=')) opt.docSize = a.slice(11);
|
|
else if (a.startsWith('--batch=')) opt.batch = Number(a.slice(8));
|
|
else if (a.startsWith('--port=')) opt.port = Number(a.slice(7));
|
|
else if (a.startsWith('--compact-threshold=')) opt.compactThreshold = a.slice(20);
|
|
else if (a === '--size') opt.size = process.argv[++i];
|
|
else if (a === '--doc-size') opt.docSize = process.argv[++i];
|
|
else if (a === '--batch') opt.batch = Number(process.argv[++i]);
|
|
else if (a === '--compact-threshold') opt.compactThreshold = process.argv[++i];
|
|
else { console.error(`unknown option ${a}`); process.exit(2); }
|
|
}
|
|
|
|
const SIZE = parseSize(opt.size);
|
|
const DOC_SIZE = parseSize(opt.docSize);
|
|
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
|
|
|
|
let server = null;
|
|
let serverLog = '';
|
|
let serverDead = false;
|
|
function cleanup() {
|
|
if (server && !serverDead) {
|
|
try { server.kill('SIGKILL'); } catch {}
|
|
}
|
|
}
|
|
process.on('exit', cleanup);
|
|
process.on('SIGINT', () => { cleanup(); process.exit(130); });
|
|
process.on('SIGTERM', () => { cleanup(); process.exit(143); });
|
|
function startServer() {
|
|
return new Promise((resolve, reject) => {
|
|
const t0 = Date.now();
|
|
serverDead = false;
|
|
const args = ['--port', String(opt.port), '--db', DBFILE];
|
|
if (opt.compactThreshold) args.push('--compact-threshold', opt.compactThreshold);
|
|
server = spawn(BIN, args, { stdio: ['ignore', 'pipe', 'pipe'] });
|
|
server.stdout.on('data', (d) => (serverLog += d));
|
|
server.stderr.on('data', (d) => (serverLog += d));
|
|
server.on('error', (e) => reject(new Error(`cannot start ${BIN}: ${e.message}`)));
|
|
server.on('exit', (code, sig) => {
|
|
serverDead = true;
|
|
if (code !== null && sig === null) serverLog += `\n[child exited rc=${code}]`;
|
|
});
|
|
const deadline = Date.now() + 120000;
|
|
(async () => {
|
|
while (Date.now() < deadline) {
|
|
if (serverDead) {
|
|
reject(new Error(`server child exited during start (port ${opt.port} busy?)\n${serverLog}`));
|
|
return;
|
|
}
|
|
const c = new MongoClient(URL, { serverSelectionTimeoutMS: 1000 });
|
|
try {
|
|
await c.connect();
|
|
await c.db('admin').command({ ping: 1 });
|
|
await c.close();
|
|
return resolve(Date.now() - t0);
|
|
} catch {
|
|
try { await c.close(); } catch {}
|
|
await sleep(100);
|
|
}
|
|
}
|
|
reject(new Error(`server did not come up on :${opt.port}\n${serverLog}`));
|
|
})();
|
|
});
|
|
}
|
|
async function stopServer(sig = 'SIGKILL') {
|
|
if (!server) return;
|
|
const exited = new Promise((r) => server.once('exit', r));
|
|
server.kill(sig);
|
|
await Promise.race([exited, sleep(5000)]);
|
|
serverDead = true;
|
|
server = null;
|
|
}
|
|
async function rssMB() {
|
|
if (!server) return 0;
|
|
try {
|
|
const out = (await new Promise((r) => require('child_process').exec(`ps -o rss= -p ${server.pid}`, (e, so) => r(so || '')))).trim();
|
|
return Math.round(Number(out) / 1024);
|
|
} catch { return 0; }
|
|
}
|
|
|
|
const report = [];
|
|
function row(label, value) { report.push([label, value]); console.log(` ${String(label).padEnd(46)} ${value}`); }
|
|
|
|
async function main() {
|
|
if (!fs.existsSync(BIN)) {
|
|
console.error(`server binary not found at ${BIN} — run \`zig build\` first`);
|
|
process.exit(1);
|
|
}
|
|
fs.rmSync(DBFILE, { force: true });
|
|
const fmt = (n) => (n >= 1e9 ? (n / 1e9).toFixed(2) + ' GB' : n >= 1e6 ? (n / 1e6).toFixed(1) + ' MB' : n >= 1e3 ? (n / 1e3).toFixed(1) + ' KB' : n + ' B');
|
|
console.log(`multiforadb big-collection harness`);
|
|
console.log(` size ${fmt(SIZE)} · doc ~${fmt(DOC_SIZE)} · batch ${opt.batch} · index ${opt.index || 'none'} · ids ${opt.oid ? 'ObjectId' : 'int'} · compact-threshold ${opt.compactThreshold || '16m'} · db ${DBFILE}`);
|
|
|
|
console.log('\n== server start (fresh log) ==');
|
|
const openMs = await startServer();
|
|
row('open + first ping (fresh log)', `${openMs} ms`);
|
|
|
|
const client = new MongoClient(URL, { serverSelectionTimeoutMS: 10000 });
|
|
await client.connect();
|
|
const db = client.db('big');
|
|
const coll = db.collection('items');
|
|
await coll.drop().catch(() => {});
|
|
|
|
if (opt.index) {
|
|
const t0 = Date.now();
|
|
await coll.createIndex({ [opt.index]: 1 });
|
|
row(`createIndex({${opt.index}: 1}) on empty coll`, `${Date.now() - t0} ms`);
|
|
}
|
|
|
|
// ---- insert ------------------------------------------------------------
|
|
console.log('\n== insert ==');
|
|
const nDocs = Math.max(1, Math.ceil(SIZE / DOC_SIZE));
|
|
const payloadLen = Math.max(1, DOC_SIZE - 130); // bson overhead for _id/k/p/ts/payload
|
|
const payload = 'x'.repeat(payloadLen);
|
|
const { ObjectId } = require('mongodb');
|
|
const t0 = Date.now();
|
|
const logSamples = [{ t: 0, size: fs.existsSync(DBFILE) ? fs.statSync(DBFILE).size : 0, rss: 0 }];
|
|
const sampler = setInterval(async () => {
|
|
logSamples.push({ t: Date.now() - t0, size: fs.existsSync(DBFILE) ? fs.statSync(DBFILE).size : 0, rss: await rssMB() });
|
|
}, 2000);
|
|
|
|
// Rate curve: avg MB/s per progress chunk, to show how throughput changes
|
|
// as the log grows (compaction rewrites + fsync-per-record dominate).
|
|
const rates = [];
|
|
let lastProgressT = t0;
|
|
let lastProgressDocs = 0;
|
|
let midDocId = null; // ObjectId of the mid doc (oid mode), captured at insert
|
|
let lastDocId = null;
|
|
let inserted = 0;
|
|
const midIndex = Math.floor(nDocs / 2);
|
|
try {
|
|
while (inserted < nDocs) {
|
|
const n = Math.min(opt.batch, nDocs - inserted);
|
|
const docs = new Array(n);
|
|
for (let i = 0; i < n; i++) {
|
|
const id = inserted + i + 1;
|
|
const doc = opt.oid
|
|
? { _id: new ObjectId(), k: id % 1000, p: id % 5000, ts: new Date(Date.UTC(2024, 0, 1) + id * 1000), payload }
|
|
: { _id: id, k: id % 1000, p: id % 5000, ts: new Date(Date.UTC(2024, 0, 1) + id * 1000), payload };
|
|
if (id === midIndex) midDocId = doc._id;
|
|
if (id === nDocs) lastDocId = doc._id;
|
|
docs[i] = doc;
|
|
}
|
|
await coll.insertMany(docs, { ordered: false });
|
|
inserted += n;
|
|
if (inserted % Math.max(1000, Math.floor(nDocs / 20)) < n) {
|
|
const dt = (Date.now() - t0) / 1000;
|
|
const mb = inserted * DOC_SIZE / 1e6;
|
|
const chunkMb = (inserted - lastProgressDocs) * DOC_SIZE / 1e6;
|
|
const chunkT = (Date.now() - lastProgressT) / 1000;
|
|
rates.push(+(chunkMb / chunkT).toFixed(1));
|
|
lastProgressT = Date.now();
|
|
lastProgressDocs = inserted;
|
|
console.log(` ${inserted.toLocaleString()} docs · ${fmt(inserted * DOC_SIZE)} · ${(mb / dt).toFixed(1)} MB/s avg`);
|
|
}
|
|
}
|
|
} finally {
|
|
clearInterval(sampler);
|
|
}
|
|
const insertMs = Date.now() - t0;
|
|
const bytes = inserted * DOC_SIZE;
|
|
row('docs inserted', inserted.toLocaleString());
|
|
row('approx bytes', fmt(bytes));
|
|
row('wall time', `${(insertMs / 1000).toFixed(1)} s`);
|
|
row('throughput', `${(bytes / 1e6 / (insertMs / 1000)).toFixed(1)} MB/s (${(inserted / (insertMs / 1000)).toFixed(0)} docs/s)`);
|
|
row('rate curve (MB/s per chunk)', rates.join(' → ') || 'n/a');
|
|
|
|
const compactions = logSamples.filter((s, i) => i > 0 && s.size < logSamples[i - 1].size - 2 * 1024 * 1024).length;
|
|
const sizes = logSamples.map((s) => s.size);
|
|
row('log file size (min → final)', `${fmt(Math.min(...sizes))} → ${fmt(sizes[sizes.length - 1])}`);
|
|
row('compaction events observed', compactions, '(log shrank by >2MB between samples)');
|
|
const peakRss = Math.max(...logSamples.map((s) => s.rss));
|
|
row('peak server RSS', `${peakRss} MB`, '(in-memory engine: docs live in RAM)');
|
|
|
|
// ---- find / read -------------------------------------------------------
|
|
console.log('\n== find / read on full dataset ==');
|
|
const bench = async (label, fn, min = 1) => {
|
|
const a = Date.now();
|
|
const res = await fn();
|
|
const ms = Date.now() - a;
|
|
row(`${label}`, `${ms < 1000 ? ms + ' ms' : (ms / 1000).toFixed(2) + ' s'}${res !== undefined ? ` (${res})` : ''}`);
|
|
return ms;
|
|
};
|
|
await bench('countDocuments({})', async () => {
|
|
const n = await coll.countDocuments({});
|
|
if (n !== inserted) throw new Error(`count ${n} != ${inserted}`);
|
|
return `${n.toLocaleString()} docs`;
|
|
});
|
|
await bench('findOne({_id: mid}) — ' + (opt.oid ? 'docs-map fast path' : 'scan (int _id: fast path skipped)'), async () => {
|
|
const d = await coll.findOne({ _id: opt.oid ? midDocId : midIndex });
|
|
if (!d) throw new Error('miss');
|
|
});
|
|
await bench('findOne({_id: last})', async () => {
|
|
const d = await coll.findOne({ _id: opt.oid ? lastDocId : nDocs });
|
|
if (!d) throw new Error('miss');
|
|
});
|
|
if (opt.index) {
|
|
await bench(`find({${opt.index}: 4242}).count() — via index`, async () => {
|
|
const n = await coll.countDocuments({ k: 4242 });
|
|
if (n < 1) throw new Error('no hits');
|
|
return `${n} hits`;
|
|
});
|
|
}
|
|
await bench('find({k: 4242}).count() — scan', async () => {
|
|
const n = await coll.countDocuments({ k: 4242 });
|
|
return `${n} hits`;
|
|
});
|
|
await bench('find({p: {$gte, $lt}}).count() — range scan', async () => {
|
|
const lo = 1000, hi = 2000;
|
|
const n = await coll.countDocuments({ p: { $gte: lo, $lt: hi } });
|
|
return `${n} hits`;
|
|
});
|
|
await bench('find({}).sort({_id:-1}).limit(20) — full scan + sort', async () => {
|
|
const docs = await coll.find({}).sort({ _id: -1 }).limit(20).toArray();
|
|
if (docs.length !== 20) throw new Error('bad page');
|
|
});
|
|
await bench('find({}, {proj: _id,k,p}).limit(500) — page', async () => {
|
|
const docs = await coll.find({}, { projection: { payload: 0 } }).limit(500).toArray();
|
|
if (docs.length !== 500) throw new Error('short page');
|
|
});
|
|
|
|
// ---- write ops against the full dataset ---------------------------------
|
|
console.log('\n== point write ops ==');
|
|
const midId = opt.oid ? midDocId : midIndex;
|
|
await bench('updateOne({_id: mid}, {$set}) — 1 fsync', async () => {
|
|
const r = await coll.updateOne({ _id: midId }, { $set: { touch: Date.now() } });
|
|
if (r.modifiedCount !== 1) throw new Error('miss');
|
|
});
|
|
await bench('updateMany({k: 7}, {$inc}) — ~N/1000 fsyncs', async () => {
|
|
const r = await coll.updateMany({ k: 7 }, { $inc: { hits: 1 } });
|
|
return `${r.modifiedCount} modified`;
|
|
});
|
|
await bench('deleteOne({_id: mid}) + re-insert', async () => {
|
|
await coll.deleteOne({ _id: midId });
|
|
if (opt.oid) {
|
|
await coll.insertOne({ _id: new ObjectId(), k: midIndex % 1000, p: midIndex % 5000, payload });
|
|
} else {
|
|
await coll.insertOne({ _id: midIndex, k: midIndex % 1000, p: midIndex % 5000, payload });
|
|
}
|
|
});
|
|
await bench('aggregate $group by k', async () => {
|
|
const out = await coll.aggregate([{ $group: { _id: '$k', n: { $sum: 1 } } }]).toArray();
|
|
return `${out.length} groups`;
|
|
});
|
|
|
|
// ---- reopen (replay) ----------------------------------------------------
|
|
console.log('\n== durability ==');
|
|
await client.close();
|
|
await stopServer('SIGKILL');
|
|
const reopenMs = await startServer();
|
|
row('kill -9 then reopen (replay of full log)', `${(reopenMs / 1000).toFixed(1)} s`);
|
|
const c2 = new MongoClient(URL, { serverSelectionTimeoutMS: 10000 });
|
|
await c2.connect();
|
|
const db2 = c2.db('big');
|
|
const coll2 = db2.collection('items');
|
|
const afterRestart = await coll2.countDocuments({});
|
|
row('count after restart', `${afterRestart.toLocaleString()} (${afterRestart === inserted ? 'OK' : 'MISMATCH!'})`);
|
|
const spot = await coll2.findOne({ _id: opt.oid ? lastDocId : nDocs });
|
|
row('last doc intact after restart', spot ? `payload ${spot.payload.length}B` : 'MISSING!');
|
|
if (afterRestart !== inserted || !spot) throw new Error('durability check failed');
|
|
|
|
// Crash-durability: every write is fsynced before it is acknowledged, so a
|
|
// kill -9 right after an insert must not lose it.
|
|
const crash = db2.collection('crash');
|
|
await crash.drop().catch(() => {});
|
|
let committed = 0;
|
|
for (let i = 1; i <= 200; i++) {
|
|
await crash.insertOne({ _id: i, seq: i });
|
|
committed = i;
|
|
}
|
|
await c2.close();
|
|
await stopServer('SIGKILL');
|
|
await startServer();
|
|
const c3 = new MongoClient(URL, { serverSelectionTimeoutMS: 10000 });
|
|
await c3.connect();
|
|
const db3 = c3.db('big');
|
|
const crashN = await db3.collection('crash').countDocuments({});
|
|
row('kill -9 after 200 committed writes', `${crashN}/200 survived (${crashN === 200 ? 'OK' : 'MISMATCH!'})`);
|
|
await c3.close();
|
|
|
|
if (!opt.keep) fs.rmSync(DBFILE, { force: true });
|
|
await stopServer('SIGKILL');
|
|
|
|
console.log('\n== summary ==');
|
|
console.log(` multiforadb handles a ${fmt(bytes)} collection fully in RAM (RSS ${peakRss} MB)`);
|
|
console.log(` insert: ${(bytes / 1e6 / (insertMs / 1000)).toFixed(1)} MB/s — fsync per write is by design (crash safety)`);
|
|
if (compactions > 0) {
|
|
console.log(` ${compactions} compaction rewrites observed: every 16MB of writes rewrites the whole log — for multi-GB loads the cumulative rewrite traffic dominates`);
|
|
}
|
|
console.log('BIG_OK');
|
|
}
|
|
|
|
main().catch((e) => {
|
|
console.error('BIG_FAIL', e);
|
|
console.log('--- server log tail ---');
|
|
console.log(serverLog.split('\n').slice(-40).join('\n'));
|
|
process.exit(1);
|
|
});
|