queueMicrotask keeps its callback alive until the next full GC #7746

// Runs iterations 50 ms apart. Each one builds a structure of about 180 KB that the next one
// drops, plus about 530 KB of garbage, and hands the structure to one callback that reads it later:
//   plain      read synchronously, nothing deferred
//   promise    Promise.resolve().then(callback)
//   microtask  queueMicrotask(callback)
//   timeout    setTimeout(callback, 0)
//   immediate  setImmediate(callback)
//   sendfrom   Promise.resolve().then(callback), and the callback sends one WebSocket message
const make = (kb) => {
  const out = [];
  for (let i = 0; i < kb * 16; i++) out.push({ a: i, b: [i, i + 1] });
  return out;
};
const DEFER = {
  plain: (callback) => callback(),
  promise: (callback) => Promise.resolve().then(callback),
  microtask: (callback) => queueMicrotask(callback),
  timeout: (callback) => setTimeout(callback, 0),
  immediate: (callback) => setImmediate(callback),
};
let sink = 0;

export default {
  async fetch(request) {
    const url = new URL(request.url);
    if (url.pathname === "/gc") {
      gc();
      return new Response("ok");
    }
    const iterations = Number(url.searchParams.get("iterations") ?? 300);
    const mode = url.searchParams.get("mode");
    const [client, server] = Object.values(new WebSocketPair());
    server.accept();
    client.accept();
    client.addEventListener("message", (event) => {
      sink += event.data.length;
    });
    for (let i = 0; i < iterations; i++) {
      const kept = { list: make(200) };
      make(600);
      if (mode === "sendfrom") {
        Promise.resolve().then(() => server.send(`${kept.list.length}`));
      } else {
        DEFER[mode](() => {
          sink += kept.list.length;
        });
      }
      await new Promise((resolve) => setTimeout(resolve, 50));
    }
    server.close();
    return new Response(`done ${sink}\n`);
  },
};

// Starts workerd with deferred.capnp, runs ITERATIONS iterations per mode between two forced
// full GCs, and prints what the scavenges promoted per iteration (--trace-gc-nvp "promoted")
// and the old generation before and after the full GC that ends the mode.
// Usage: WORKERD=path/to/workerd node run-deferred.mjs [iterations]
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 ITERATIONS = Number(process.argv[2] ?? 300);
const BASE = "127.0.0.1:8788";
const MODES = ["plain", "promise", "microtask", "timeout", "immediate", "sendfrom"];

const log = createWriteStream("deferred.log");
const workerd = spawn(WORKERD, ["serve", "deferred.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, timeout) =>
  new Promise((resolve, reject) => {
    const req = http.get(`http://${BASE}${path}`, { agent: false, timeout }, (res) => {
      let body = "";
      res.on("data", (chunk) => (body += chunk));
      res.on("end", () => resolve(body));
    });
    req.on("timeout", () => req.destroy(new Error(`timeout: ${path}`)));
    req.on("error", reject);
  });
for (let i = 0; ; i++) {
  try {
    await get("/gc", 10_000); // forced full GC 0
    break;
  } catch (error) {
    if (i === 50) throw error;
    await sleep(100);
  }
}

// Mode m runs between forced full GCs 2m+1 and 2m+2.
for (const mode of MODES) {
  await get(`/?mode=${mode}&iterations=20`, 60_000); // warm up
  await get("/gc", 10_000);
  await get(`/?mode=${mode}&iterations=${ITERATIONS}`, ITERATIONS * 1000);
  await get("/gc", 10_000);
}
// 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("deferred.log", "utf8").match(/"gc":"mc".*"reason":"testing".*\}\n/g) ?? []).length;
for (let i = 0; forcedInLog() < 2 + 2 * MODES.length; i++) {
  if (i === 300) throw new Error("the GC trace never reached the log");
  await get("/gc", 10_000);
  await sleep(100);
}
workerd.kill();
await new Promise((resolve) => log.end(resolve));

const records = [];
for (const line of readFileSync("deferred.log", "utf8").split("\n")) {
  const nvp = line.match(/ms: GC: (\{.*\})$/);
  if (nvp) records.push(JSON.parse(nvp[1]));
}
const forced = records.flatMap((r, index) => (r.gc === "mc" && r.reason === "testing" ? [index] : []));
if (forced.length < 1 + 2 * MODES.length) throw new Error(`only ${forced.length} forced full GCs in the trace`);

console.log(`${ITERATIONS} iterations 50 ms apart per mode, each keeping about 180 KB that one deferred callback reads`);
console.log("mode        scavenges  promoted KB/iteration  old gen before full GC MB  after MB");
MODES.forEach((mode, m) => {
  const [from, to] = [forced[2 * m + 1], forced[2 * m + 2]];
  const scavenges = records.slice(from + 1, to).filter((r) => r.gc === "s");
  const promoted = scavenges.reduce((sum, r) => sum + r.promoted, 0);
  const end = records[to];
  console.log(
    `${mode.padEnd(10)} ${String(scavenges.length).padStart(10)}  ${(promoted / 1024 / ITERATIONS).toFixed(1).padStart(21)}  ${
      (end.start_old_gen_consumed_size / 1048576).toFixed(1).padStart(25)
    }  ${(end.end_old_gen_consumed_size / 1048576).toFixed(1).padStart(8)}`,
  );
});

// The loop from deferred.js in plain Node, as a control for queueMicrotask: every 50 ms build the
// same structure, which the next iteration drops, plus the same garbage, and read the structure
// from one deferred callback. Runs each mode in a child Node with --trace-gc-nvp and
// a 16 MB semi-space, and prints what the scavenges promoted per iteration.
// Usage: node node-control.mjs
import { execFileSync } from "node:child_process";
import { setTimeout as sleep } from "node:timers/promises";

const ITERATIONS = 300;
const MODES = ["plain", "timeout", "microtask"];
const make = (kb) => {
  const out = [];
  for (let i = 0; i < kb * 16; i++) out.push({ a: i, b: [i, i + 1] });
  return out;
};
const DEFER = {
  plain: (callback) => callback(),
  timeout: (callback) => setTimeout(callback, 0),
  microtask: (callback) => queueMicrotask(callback),
};

const mode = process.argv[2];
if (mode) {
  let sink = 0;
  for (let i = 0; i < ITERATIONS; i++) {
    const kept = { list: make(200) };
    make(600);
    DEFER[mode](() => {
      sink += kept.list.length;
    });
    await sleep(50);
  }
} else {
  console.log(`node ${process.version} (V8 ${process.versions.v8}), ${ITERATIONS} iterations 50 ms apart per mode`);
  console.log("mode        scavenges  promoted KB/iteration");
  for (const m of MODES) {
    const trace = execFileSync(
      process.execPath,
      ["--trace-gc-nvp", "--min-semi-space-size=16", "--max-semi-space-size=16", import.meta.filename, m],
      { encoding: "utf8", maxBuffer: 64 * 1024 * 1024 },
    );
    const scavenges = [...trace.matchAll(/(\{"pause".*\})/g)].map((r) => JSON.parse(r[1])).filter((r) => r.gc === "s");
    const promoted = scavenges.reduce((sum, r) => sum + r.promoted, 0);
    console.log(
      `${m.padEnd(10)} ${String(scavenges.length).padStart(10)}  ${(promoted / 1024 / ITERATIONS).toFixed(1).padStart(21)}`,
    );
  }
}