Each WebSocket send from an idle socket leaves 4 FunctionTemplateInfo objects in old space #7745

// A Durable Object that holds WebSockets and sends to each every 50 ms.
// Every 50 ms it runs four timer callbacks 10 ms apart in every mode, so only the
// send pattern differs between modes:
//   none   no sends
//   one    callback 0 sends 1 message
//   same   callback 0 sends 4 messages (one turn)
//   split  callbacks 0-3 send 1 message each (four turns)
import { DurableObject } from "cloudflare:workers";

const SENDS = {
  none: [0, 0, 0, 0],
  one: [1, 0, 0, 0],
  same: [4, 0, 0, 0],
  split: [1, 1, 1, 1],
};

export class Pumper extends DurableObject {
  sockets = new Set();
  mode = "none";
  interval = undefined;

  fetch(request) {
    const url = new URL(request.url);
    this.mode = url.searchParams.get("mode") ?? "none";
    const hibernate = url.searchParams.get("api") === "hibernate";
    const [client, server] = Object.values(new WebSocketPair());
    if (hibernate) {
      this.ctx.acceptWebSocket(server);
    } else {
      server.accept();
      server.addEventListener("close", () => this.drop(server));
    }
    this.sockets.add(server);
    this.interval ??= setInterval(() => this.onInterval(), 50);
    return new Response(null, { status: 101, webSocket: client });
  }

  webSocketClose(ws) {
    this.drop(ws);
  }

  drop(ws) {
    this.sockets.delete(ws);
    if (this.sockets.size === 0) {
      clearInterval(this.interval);
      this.interval = undefined;
    }
  }

  onInterval() {
    SENDS[this.mode].forEach((count, i) => {
      setTimeout(() => {
        for (const ws of this.sockets) {
          for (let n = 0; n < count; n++) ws.send("x");
        }
      }, i * 10);
    });
  }
}

export default {
  fetch(request, env) {
    const url = new URL(request.url);
    if (url.pathname === "/gc") {
      gc();
      return new Response("ok");
    }
    return env.PUMPER.get(env.PUMPER.idFromName("one")).fetch(request);
  },
};

// Starts workerd, runs each phase with SOCKETS clients for SECONDS, and forces a full GC at
// the start and end of each phase. Per phase it prints the objects that died in between
// (--trace-gc-object-stats) and the old generation's growth between GCs, which is what was
// allocated directly in the old generation (--trace-gc-nvp).
// Usage: WORKERD=path/to/workerd node run.mjs [seconds] [sockets]
import { spawn } from "node:child_process";
import { createWriteStream, readFileSync } from "node:fs";
import http from "node:http";
import { setTimeout as sleep } from "node:timers/promises";

const WORKERD = process.env.WORKERD ?? "node_modules/.bin/workerd";
const SECONDS = Number(process.argv[2] ?? 30);
const SOCKETS = Number(process.argv[3] ?? 16);
const BASE = "127.0.0.1:8787";
const PHASES = [
  ["none", "accept"],
  ["one", "accept"],
  ["same", "accept"],
  ["split", "accept"],
  ["one", "hibernate"],
  ["split", "hibernate"],
];
// messages per send turn, so send turns/s (pump starts/s) = messages received/s / this
const PER_TURN = { none: Infinity, one: 1, same: 4, split: 1 };
const TYPES = ["FUNCTION_TEMPLATE_INFO_TYPE", "SHARED_FUNCTION_INFO_TYPE"];

const log = createWriteStream("gc.log");
const workerd = spawn(WORKERD, ["serve", "config.capnp"], { stdio: ["ignore", "pipe", "pipe"] });
workerd.stdout.pipe(log);
workerd.stderr.pipe(log);
process.on("exit", () => workerd.kill());

// one fresh connection per request
const get = (path) =>
  new Promise((resolve, reject) => {
    const req = http.get(`http://${BASE}${path}`, { agent: false, timeout: 10_000 }, (res) => {
      res.resume();
      res.on("end", resolve);
    });
    req.on("timeout", () => req.destroy(new Error(`timeout: ${path}`)));
    req.on("error", reject);
  });
for (let i = 0; ; i++) {
  try {
    await get("/gc"); // forced full GC 0
    break;
  } catch (error) {
    if (i === 50) throw error;
    await sleep(100);
  }
}

// Phase p runs between forced full GCs 2p+1 and 2p+2.
const received = [];
for (const [mode, api] of PHASES) {
  const sockets = Array.from({ length: SOCKETS }, () => new WebSocket(`ws://${BASE}/?mode=${mode}&api=${api}`));
  await Promise.all(sockets.map((ws) => new Promise((resolve) => ws.addEventListener("open", resolve))));
  let count = 0;
  for (const ws of sockets) ws.addEventListener("message", () => count++);
  await sleep(2000); // warm up before the measured window
  await get("/gc");
  count = 0;
  await sleep(SECONDS * 1000);
  received.push(count);
  await get("/gc");
  for (const ws of sockets) ws.close();
  await sleep(1000);
}
// workerd buffers V8's trace output: force full GCs until the log holds one complete
// record past the last forced GC the analysis needs
const forcedInLog = () => (readFileSync("gc.log", "utf8").match(/"gc":"mc".*"reason":"testing".*\}\n/g) ?? []).length;
for (let i = 0; forcedInLog() < 2 + 2 * PHASES.length; i++) {
  if (i === 300) throw new Error("the GC trace never reached the log");
  await get("/gc");
  await sleep(100);
}
workerd.kill();
await new Promise((resolve) => log.end(resolve));

// Both traces go through the same stdout, in order: the n-th mark-compact record of
// --trace-gc-nvp and the n-th GC id of --trace-gc-object-stats are the same GC.
const records = []; // every GC, from --trace-gc-nvp
const deadById = new Map(); // GC id -> instance type -> { count, bytes }, in order of appearance
const timeById = new Map();
for (const line of readFileSync("gc.log", "utf8").split("\n")) {
  const nvp = line.match(/ms: GC: (\{.*\})$/);
  if (nvp) {
    records.push(JSON.parse(nvp[1]));
    continue;
  }
  if (!line.startsWith('{ "isolate"')) continue;
  let r;
  try {
    r = JSON.parse(line);
  } catch {
    continue; // the last line can be cut off when workerd exits
  }
  if (r.type === "gc_descriptor") timeById.set(r.id, r.time);
  if (!deadById.has(r.id)) deadById.set(r.id, {});
  if (r.type === "instance_type_data" && r.key === "dead") {
    deadById.get(r.id)[r.instance_type_name] = { count: r.count, bytes: r.overall };
  }
}
const fullGcs = records.flatMap((r, index) => (r.gc === "mc" ? [{ ...r, index }] : []));
const ids = [...deadById.keys()];
// the newest GC can be missing from one of the two traces when workerd exits
if (Math.abs(fullGcs.length - ids.length) > 1) {
  throw new Error(`${fullGcs.length} full GCs in nvp, ${ids.length} in object stats`);
}
const forced = fullGcs.flatMap((gc, n) => (gc.reason === "testing" ? [n] : []));
if (forced.length < 1 + 2 * PHASES.length) throw new Error(`only ${forced.length} forced full GCs in the trace`);

const results = PHASES.map(([mode, api], p) => {
  const [from, to] = [forced[2 * p + 1], forced[2 * p + 2]];
  const seconds = (timeById.get(ids[to]) - timeById.get(ids[from])) / 1000;
  const dead = {};
  for (let n = from + 1; n <= to; n++) {
    for (const [type, { count, bytes }] of Object.entries(deadById.get(ids[n]))) {
      dead[type] ??= { count: 0, bytes: 0 };
      dead[type].count += count;
      dead[type].bytes += bytes;
    }
  }
  // growth between the end of one GC and the start of the next is direct old-generation allocation
  let directOld = 0;
  for (let i = fullGcs[from].index + 1; i <= fullGcs[to].index; i++) {
    directOld += records[i].start_old_gen_consumed_size - records[i - 1].end_old_gen_consumed_size;
  }
  return {
    label: `${mode}-${api}`,
    messages: received[p] / SECONDS,
    turns: Math.round(received[p] / SECONDS / PER_TURN[mode]),
    deadKb: Object.values(dead).reduce((sum, e) => sum + e.bytes, 0) / 1024 / seconds,
    counts: TYPES.map((t) => (dead[t]?.count ?? 0) / seconds),
    oldKb: directOld / 1024 / seconds,
  };
});

const idleOldKb = results[0].oldKb;
console.log(`${SOCKETS} sockets, ${SECONDS} s per phase, 4 timer callbacks 10 ms apart every 50 ms in every phase`);
console.log(
  "phase            msgs/s  send turns/s  dead KB/s  " + TYPES.map((t) => t.replace("_TYPE", "") + "/s").join("  ") +
    "  old KB/s  FTI per turn  old KB per turn",
);
for (const r of results) {
  const cols = r.counts.map((c, i) => c.toFixed(1).padStart(TYPES[i].length - 3));
  console.log(
    `${r.label.padEnd(16)} ${r.messages.toFixed(0).padStart(6)}  ${String(r.turns).padStart(12)}  ${
      r.deadKb.toFixed(1).padStart(9)
    }  ${cols.join("  ")}  ${r.oldKb.toFixed(1).padStart(8)}  ${
      (r.turns ? (r.counts[0] / r.turns).toFixed(2) : "-").padStart(12)
    }  ${(r.turns ? ((r.oldKb - idleOldKb) / r.turns).toFixed(2) : "-").padStart(15)}`,
  );
}