MultiforaDB/tests/e2e/churn.js

// The churn gate: how large the data file settles at, relative to the live
// data, under sustained rewriting.
//
// This is the measurement PLAN D7.4 was decided on, and until now it was the
// only gate in `tests/e2e/results/m0-gates.txt` without a `reproduce:` line --
// the numbers were real but the harness was not committed, so nobody could
// re-run them against a change. That is what this file fixes.
//
// It spawns its own server on a fresh database, so it needs nothing running:
//
//   node tests/e2e/churn.js --docs 40000 --doc-size 16k --index \
//        --mode delete-refill --rounds 6
//   node tests/e2e/churn.js --docs 40000 --doc-size 16k --index \
//        --mode update --multiple 5
//
// What to read. The ratio alone does not say whether reclamation is working:
// a file that grows and is periodically halved by a rebuild averages out to a
// respectable number. So every round prints `reclaimed`, `allocTail` and
// `compactions` beside it. Reclamation is carrying the workload when
// `reclaimed` climbs while `allocTail` stays put. If instead `allocTail` grows
// by the full write volume of each round, the free list is decorative however
// good the ratio looks -- and that is a failure even at 1.2x.
//
// Small documents are expected not to improve on the delete-refill line, and
// that is a pass rather than a fault. Reclamation gives back whole system
// pages, so a 16 KiB page holds ~82 documents of 200 bytes and the chance all
// 82 are dead at once is nil. The mechanism to check there is that the
// counters decay correctly, not that the ratio moves.
//
// Options:
//   --docs <n>          documents in the collection (default 40000)
//   --doc-size <n[k|m]> payload bytes per document (default 16k)
//   --index             create one secondary index over a churned field
//   --mode <m>          delete-refill (default) | update
//   --rounds <n>        rounds in each mode; update mode splits its writes
//                       across them (default 6)
//   --multiple <n>      update mode: total writes as a multiple of --docs
//   --target <x>        fail unless the steady-state ratio is at or under x
//   --port <n>          listen port (default 27320)
//   --keep              leave the database file behind
const { MongoClient } = require('mongodb');
const { spawn } = require('child_process');
const fs = require('fs');
const path = require('path');

function parseSize(s) {
  const m = String(s).match(/^(\d+)([kKmMgG]?)$/);
  if (!m) throw new Error(`bad size: ${s}`);
  const mult = { '': 1, k: 1 << 10, m: 1 << 20, g: 1 << 30 }[m[2].toLowerCase()];
  return Number(m[1]) * mult;
}

function parseArgs(argv) {
  const o = {
    docs: 40000, docSize: 16 << 10, index: false, mode: 'delete-refill',
    rounds: 6, multiple: 5, target: null, port: 27320, keep: false,
  };
  for (let i = 0; i < argv.length; i++) {
    const a = argv[i];
    const next = () => {
      if (i + 1 >= argv.length) throw new Error(`${a} needs a value`);
      return argv[++i];
    };
    switch (a) {
      case '--docs': o.docs = Number(next()); break;
      case '--doc-size': o.docSize = parseSize(next()); break;
      case '--index': o.index = true; break;
      case '--mode': o.mode = next(); break;
      case '--rounds': o.rounds = Number(next()); break;
      case '--multiple': o.multiple = Number(next()); break;
      case '--target': o.target = Number(next()); break;
      case '--port': o.port = Number(next()); break;
      case '--keep': o.keep = true; break;
      default: throw new Error(`unknown option ${a}`);
    }
  }
  if (o.mode !== 'delete-refill' && o.mode !== 'update') {
    throw new Error(`--mode must be delete-refill or update, got ${o.mode}`);
  }
  return o;
}

const opt = parseArgs(process.argv.slice(2));
const BIN = process.env.MFDB_BIN || path.resolve(__dirname, '../../zig-out/bin/multiforadb');
const DBFILE = process.env.CHURN_DB ||
  path.resolve(__dirname, `../../.zig-cache/churn-${opt.port}.log`);
const URL = `mongodb://127.0.0.1:${opt.port}`;

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// A fixed seed, so two runs churn the same documents in the same order and a
// difference between them is the code rather than the dice.
let seed = 0x9e3779b9;
function rnd() {
  seed ^= seed << 13; seed >>>= 0;
  seed ^= seed >> 17;
  seed ^= seed << 5; seed >>>= 0;
  return seed / 0x100000000;
}
const pick = (n) => Math.floor(rnd() * n);

let server = null;
let serverDead = false;
// Bounded: the listeners below run for the life of the process and only the
// last few lines are ever read, so an unbounded string would hold a rope
// proportional to everything the server ever said.
let serverLog = '';
const noteServerLog = (d) => {
  serverLog = (serverLog + d).slice(-65536);
};

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) => {
    fs.rmSync(DBFILE, { force: true });
    fs.rmSync(DBFILE + '.data', { force: true });
    serverDead = false;
    server = spawn(BIN, ['--port', String(opt.port), '--db', DBFILE], {
      stdio: ['ignore', 'pipe', 'pipe'],
    });
    server.stdout.on('data', noteServerLog);
    server.stderr.on('data', noteServerLog);
    server.on('error', (e) => reject(new Error(`cannot start ${BIN}: ${e.message}`)));
    server.on('exit', (code, sig) => {
      // A child that dies must fail the start, or the poll below would find a
      // *stale* server on the same port and measure the wrong database.
      serverDead = true;
      if (code !== null && sig === null) noteServerLog(`\n[child exited rc=${code}]`);
    });
    const deadline = Date.now() + 15000;
    (async () => {
      while (Date.now() < deadline) {
        if (serverDead) {
          reject(new Error(`server child exited during start (port ${opt.port} busy?)\n${serverLog}`));
          return;
        }
        // A fresh client per attempt, deliberately: a MongoClient whose first
        // connect fails tears its topology down and every later command on it
        // fails the same way, so reusing one turns "not up yet" into "never
        // comes up".
        const c = new MongoClient(URL, { serverSelectionTimeoutMS: 1000 });
        try {
          await c.connect();
          await c.db('admin').command({ ping: 1 });
          await c.close();
          return resolve();
        } 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() {
  if (!server) return;
  const exited = new Promise((r) => server.once('exit', r));
  server.kill('SIGTERM');
  await Promise.race([exited, sleep(5000)]);
  serverDead = true;
  server = null;
}

function dataFileBytes() {
  try {
    return fs.statSync(DBFILE + '.data').size;
  } catch {
    return 0;
  }
}

const MB = (n) => (n / (1 << 20)).toFixed(1);

// Live bytes are computed here rather than read off the server, so the ratio
// means the same thing whatever binary is under test. A `multifora` section is
// a recent addition; without it the counters print as n/a and the ratio -- the
// number the gate is actually about -- is still measured, which is what makes
// a run against an older build comparable.
let docBytes = 0;

async function stats(client, count) {
  const s = await client.db('admin').command({ serverStatus: 1 });
  const m = s.multifora || null;
  return {
    live: count * docBytes,
    count,
    file: dataFileBytes(),
    m: m && {
      live: Number(m.liveBytes),
      dead: Number(m.deadBytes),
      reclaimed: Number(m.reclaimedBytes),
      runs: Number(m.slabRuns),
      freeReady: Number(m.freeReadyBytes),
      allocTail: Number(m.allocTailBytes),
      compactions: Number(m.compactions),
    },
  };
}

function report(label, s) {
  const ratio = s.live > 0 ? s.file / s.live : 0;
  let line = `  ${label.padEnd(12)} ratio ${ratio.toFixed(2)}x  file ${MB(s.file)}MB  live ${MB(s.live)}MB`;
  if (s.m) {
    // What the database is actually occupying, as opposed to what it has ever
    // had to occupy. The file never shrinks, so `ratio` is a high-water mark
    // and cannot come down however well reclamation works; `inUse` is the
    // number that moves when it does.
    const inUse = s.m.allocTail - s.m.freeReady;
    line += `  inUse ${(inUse / s.live).toFixed(2)}x` +
      `  dead ${MB(s.m.dead)}MB  reclaimed ${MB(s.m.reclaimed)}MB` +
      `  allocTail ${MB(s.m.allocTail)}MB  freeReady ${MB(s.m.freeReady)}MB` +
      `  runs ${s.m.runs}  compactions ${s.m.compactions}`;
  } else {
    line += '  (no multifora section: counters n/a)';
  }
  console.log(line);
  return ratio;
}

// One document of about `opt.docSize` payload bytes. `k` is the field a
// secondary index covers and an update rewrites, so index maintenance is part
// of the churn rather than a constant.
const PAD = 'x'.repeat(Math.max(1, opt.docSize));
function makeDoc(id) {
  return { _id: id, k: id % 1000, pad: PAD };
}

// Batches sized so one insertMany stays well under the 48 MB wire limit
// whatever --doc-size is.
function batchSize() {
  return Math.max(1, Math.min(1000, Math.floor((8 << 20) / (opt.docSize + 64))));
}

async function insertRange(coll, from, to) {
  const bs = batchSize();
  for (let i = from; i < to; i += bs) {
    const docs = [];
    for (let j = i; j < Math.min(i + bs, to); j++) docs.push(makeDoc(j));
    await coll.insertMany(docs, { ordered: false });
  }
}

async function main() {
  console.log(
    `churn: ${opt.docs} x ${opt.docSize} B, mode ${opt.mode}` +
    `${opt.index ? ', one secondary index' : ''}` +
    `${opt.mode === 'delete-refill' ? `, ${opt.rounds} rounds` : `, ${opt.multiple}x writes`}`,
  );
  console.log(`churn: platform ${process.platform}/${process.arch}, binary ${BIN}`);
  await startServer();
  const client = new MongoClient(URL, { serverSelectionTimeoutMS: 5000 });
  await client.connect();
  const db = client.db('churn');
  const coll = db.collection('c');

  const t0 = Date.now();
  // The exact serialized size of one document, so `live` is a real byte count
  // rather than the payload size the caller asked for.
  docBytes = require('mongodb').BSON.serialize(makeDoc(0)).length;
  await insertRange(coll, 0, opt.docs);
  if (opt.index) await coll.createIndex({ k: 1 });
  const base = await stats(client, opt.docs);
  report('loaded', base);

  const ratios = [];
  let nextId = opt.docs;
  if (opt.mode === 'delete-refill') {
    // Half the collection dies and is replaced by fresh documents, so the live
    // size is flat and everything the file gains is garbage that was not
    // reclaimed.
    const half = Math.floor(opt.docs / 2);
    // The ids actually live, so a round deletes exactly `half` documents and
    // the collection stays the same size. Sampling blind from the id space
    // re-picks already-dead ids, which deletes fewer than it inserts and turns
    // a flat-live measurement into a growing one.
    const live = Array.from({ length: opt.docs }, (_, i) => i);
    for (let r = 0; r < opt.rounds; r++) {
      const ids = [];
      for (let i = 0; i < half; i++) {
        const at = pick(live.length);
        ids.push(live[at]);
        live[at] = live[live.length - 1];
        live.pop();
      }
      // One delete spec per id, not one `$in` over thousands of them. The
      // server's planner refuses to use an index for an `$in` wider than
      // `index.max_combos` (100), so a 5000-element one falls back to a full
      // collection scan that re-filters every document against every member --
      // quadratic in `--docs`, and it was the whole of this harness's runtime:
      // the documented 40k x 16 KiB gate took 62 s and takes 11 s now, and the
      // 150k x 200 B line went from ~480 s to 3 s. Reported ratios are
      // unchanged, which is the point: this was the instrument's cost, not the
      // database's.
      const bs = 5000;
      for (let i = 0; i < ids.length; i += bs) {
        await coll.bulkWrite(
          ids.slice(i, i + bs).map((id) => ({ deleteOne: { filter: { _id: id } } })),
          { ordered: false },
        );
      }
      await insertRange(coll, nextId, nextId + ids.length);
      for (let i = 0; i < ids.length; i++) live.push(nextId + i);
      nextId += ids.length;
      ratios.push(report(`round ${r + 1}`, await stats(client, opt.docs)));
    }
  } else {
    // The same documents rewritten over and over: every rewrite leaves the old
    // copy behind, and old copies die in insertion order, which is the best
    // case for reclaiming whole windows.
    const total = opt.docs * opt.multiple;
    const per = Math.floor(total / opt.rounds);
    for (let r = 0; r < opt.rounds; r++) {
      let done = 0;
      while (done < per) {
        const ops = [];
        for (let i = 0; i < Math.min(2000, per - done); i++) {
          const id = pick(opt.docs);
          ops.push({ updateOne: { filter: { _id: id }, update: { $set: { k: pick(1000) } } } });
        }
        await coll.bulkWrite(ops, { ordered: false });
        done += ops.length;
      }
      ratios.push(report(`round ${r + 1}`, await stats(client, opt.docs)));
    }
  }

  // The one real count in the run, and the only one the check below needs.
  const count = await coll.countDocuments({});
  const final = await stats(client, count);
  const elapsed = ((Date.now() - t0) / 1000).toFixed(0);
  console.log(`churn: ${count} documents live at the end, ${elapsed}s`);

  // Steady state is the second half of the rounds: the first ones are still
  // filling the file out and say nothing about where it settles.
  const tail = ratios.slice(Math.floor(ratios.length / 2));
  const steady = tail.reduce((a, b) => a + b, 0) / tail.length;
  const drift = tail.length > 1 ? tail[tail.length - 1] - tail[0] : 0;
  console.log(
    `churn: steady state ${steady.toFixed(2)}x over the last ${tail.length} rounds, ` +
    `drift ${drift >= 0 ? '+' : ''}${drift.toFixed(2)}x`,
  );
  if (final.m) {
    console.log(
      `churn: reclaimed ${MB(final.m.reclaimed)}MB total, ` +
      `${final.m.compactions} collection rebuilds`,
    );
  }

  await client.close();
  if (!opt.keep) {
    fs.rmSync(DBFILE, { force: true });
    fs.rmSync(DBFILE + '.data', { force: true });
  }
  await stopServer();

  if (count !== opt.docs) {
    console.log(`CHURN_FAIL: ${count} documents live, expected ${opt.docs}`);
    process.exit(1);
  }
  if (opt.target !== null && steady > opt.target) {
    console.log(`CHURN_FAIL: steady state ${steady.toFixed(2)}x is above the ${opt.target}x target`);
    process.exit(1);
  }
  console.log('CHURN_OK');
}

main().catch((e) => {
  console.error('CHURN_FAIL', e);
  console.log('--- server log tail ---');
  console.log(serverLog.split('\n').slice(-40).join('\n'));
  process.exit(1);
});