Files
paseo/scripts/benchmark-terminal-latency.ts
2026-06-16 17:35:16 +07:00

764 lines
23 KiB
TypeScript
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/**
* Reproducible terminal latency benchmark (Node-only, isolated daemon).
*
* Boots its OWN isolated daemon subprocess (fresh mkdtemp PASEO_HOME, random
* port) — it NEVER touches the developer daemon on port 6767 — and measures:
* A) terminal echo latency (single-byte input -> first echoed output frame)
* B) terminal output jitter (inter-frame gaps while a command drains ~2MB)
* C) RPC ping RTT at 10Hz (proxy for daemon main-loop delay)
* across load levels L0..L3 where load is generated by dev-only `mock` agents
* that stream agent updates over the same socket the benchmark client uses.
*
* Run:
* npx tsx scripts/benchmark-terminal-latency.ts
*
* Optional:
* BENCH_INCLUDE_L3=1 include the L3 level (L2 + a second noisy terminal)
*
* Output: pretty table to stdout + JSON to /tmp/paseo-terminal-bench/<ts>.json
*
* Requires built client/protocol dist (packages/client/dist). Build with:
* npm run build:client
*/
import { spawn, execSync, type ChildProcess } from "node:child_process";
import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises";
import net from "node:net";
import os from "node:os";
import path from "node:path";
import { performance } from "node:perf_hooks";
import { fileURLToPath, pathToFileURL } from "node:url";
import { WebSocket } from "ws";
const __dirname = path.dirname(fileURLToPath(import.meta.url));
const REPO_ROOT = path.resolve(__dirname, "..");
const SERVER_DIR = path.join(REPO_ROOT, "packages", "server");
// --- minimal structural view of the dist DaemonClient we depend on ----------
interface TerminalStreamEvent {
terminalId: string;
type: "output" | "snapshot" | "restore";
data?: Uint8Array;
}
interface DaemonClientLike {
connect(): Promise<void>;
close(): Promise<void>;
ping(params?: { timeoutMs?: number }): Promise<{ rttMs: number }>;
listTerminals(cwd?: string): Promise<unknown>;
openProject(cwd: string): Promise<{ workspace: { id: string } | null; error: string | null }>;
createTerminal(
cwd: string,
name?: string,
requestId?: string,
options?: { workspaceId?: string },
): Promise<{ terminal: { id: string } | null; error: string | null }>;
subscribeTerminal(terminalId: string): Promise<{ error: string | null }>;
killTerminal(terminalId: string): Promise<unknown>;
sendTerminalInput(
terminalId: string,
message: { type: "input"; data: string } | { type: "resize"; rows: number; cols: number },
): void;
onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void;
createAgent(options: {
provider: string;
cwd: string;
workspaceId?: string;
model?: string;
title?: string;
}): Promise<{ id: string }>;
sendAgentMessage(agentId: string, text: string): Promise<void>;
cancelAgent?(agentId: string): Promise<unknown>;
deleteAgent?(agentId: string): Promise<unknown>;
}
interface DaemonClientCtor {
new (config: {
url: string;
clientId: string;
clientType: "cli";
appVersion?: string;
webSocketFactory: (url: string, options?: { headers?: Record<string, string> }) => unknown;
}): DaemonClientLike;
}
// --- stats helpers ----------------------------------------------------------
interface Stats {
count: number;
min: number;
p50: number;
p95: number;
max: number;
avg: number;
}
function summarize(values: number[]): Stats {
if (values.length === 0) {
return { count: 0, min: 0, p50: 0, p95: 0, max: 0, avg: 0 };
}
const sorted = [...values].sort((a, b) => a - b);
const pick = (q: number) => sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))];
return {
count: sorted.length,
min: round(sorted[0]),
p50: round(pick(0.5)),
p95: round(pick(0.95)),
max: round(sorted[sorted.length - 1]),
avg: round(values.reduce((a, b) => a + b, 0) / values.length),
};
}
function round(n: number): number {
return Math.round(n * 100) / 100;
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
// --- daemon lifecycle -------------------------------------------------------
async function getFreePort(): Promise<number> {
return new Promise((resolve, reject) => {
const srv = net.createServer();
srv.on("error", reject);
srv.listen(0, "127.0.0.1", () => {
const addr = srv.address();
if (addr === null || typeof addr === "string") {
srv.close();
reject(new Error("Failed to acquire a free port"));
return;
}
const { port } = addr;
srv.close(() => resolve(port));
});
});
}
async function waitForPort(port: number, child: ChildProcess, timeoutMs: number): Promise<void> {
const start = Date.now();
let lastError: unknown = null;
while (Date.now() - start < timeoutMs) {
if (child.exitCode !== null) {
throw new Error(`Daemon exited before listening (code ${child.exitCode})`);
}
try {
await new Promise<void>((resolve, reject) => {
const socket = net.connect(port, "127.0.0.1", () => {
socket.end();
resolve();
});
socket.setTimeout(1000, () => {
socket.destroy();
reject(new Error("connect timeout"));
});
socket.on("error", reject);
});
return;
} catch (error) {
lastError = error;
await sleep(150);
}
}
throw new Error(
`Daemon did not listen on 127.0.0.1:${port} within ${timeoutMs}ms (${String(lastError)})`,
);
}
interface BootedDaemon {
child: ChildProcess;
port: number;
paseoHome: string;
pid: number;
}
async function bootDaemon(): Promise<BootedDaemon> {
const port = await getFreePort();
if (port === 6767) {
throw new Error("Refusing to use port 6767 (the developer daemon)");
}
const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-bench-home-"));
const tsxBin = execSync("which tsx").toString().trim();
const child = spawn(tsxBin, ["scripts/supervisor-entrypoint.ts", "--dev"], {
cwd: SERVER_DIR,
env: {
...process.env,
PASEO_HOME: paseoHome,
PASEO_SERVER_ID: "srv_terminal_bench",
PASEO_LISTEN: `127.0.0.1:${port}`,
PASEO_NODE_ENV: "development",
NODE_ENV: "development",
},
stdio: ["ignore", "pipe", "pipe"],
detached: false,
});
child.stdout?.on("data", (data: Buffer) => {
for (const line of data.toString().split("\n")) {
const trimmed = line.trim();
if (trimmed) console.log(`[daemon] ${trimmed}`);
}
});
child.stderr?.on("data", (data: Buffer) => {
for (const line of data.toString().split("\n")) {
const trimmed = line.trim();
if (trimmed) console.error(`[daemon] ${trimmed}`);
}
});
await waitForPort(port, child, 30_000);
return { child, port, paseoHome, pid: child.pid ?? -1 };
}
async function loadDaemonClientCtor(): Promise<DaemonClientCtor> {
const moduleUrl = pathToFileURL(
path.join(REPO_ROOT, "packages", "client", "dist", "daemon-client.js"),
).href;
const mod = (await import(moduleUrl)) as { DaemonClient: DaemonClientCtor };
return mod.DaemonClient;
}
async function connectClient(ctor: DaemonClientCtor, port: number): Promise<DaemonClientLike> {
const client = new ctor({
url: `ws://127.0.0.1:${port}/ws`,
clientId: `bench-${Math.random().toString(36).slice(2)}`,
clientType: "cli",
webSocketFactory: (url, options) => new WebSocket(url, { headers: options?.headers }),
});
await client.connect();
return client;
}
// --- measurement primitives -------------------------------------------------
const ECHO_SAMPLES = 100;
const ECHO_INTERVAL_MS = 50;
const ECHO_FRAME_TIMEOUT_MS = 2_000;
// Printable ASCII the shell echoes back verbatim (skip control chars/space).
const ECHO_ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
interface EchoResult {
samples: number;
hits: number;
latencyMs: Stats;
}
/**
* Sends single distinct bytes spaced out in time; each byte's latency is the
* gap between send and the first output frame whose decoded text contains it.
*/
async function measureEcho(client: DaemonClientLike, terminalId: string): Promise<EchoResult> {
const decoder = new TextDecoder();
// The stream handler records the arrival time of the awaited echo byte; the
// sender loop polls this between sends. A plain shared slot avoids juggling a
// promise resolver across the handler/timeout boundary.
// Mutated by the async stream handler and read by the sender loop; a holder
// object keeps the cross-callback mutation explicit.
const slot: { awaitedChar: string | null; arrivalAt: number | null } = {
awaitedChar: null,
arrivalAt: null,
};
const unsubscribe = client.onTerminalStreamEvent((event) => {
if (event.terminalId !== terminalId || event.type !== "output" || !event.data) {
return;
}
if (slot.awaitedChar === null || slot.arrivalAt !== null) {
return;
}
if (decoder.decode(event.data).includes(slot.awaitedChar)) {
slot.arrivalAt = performance.now();
}
});
const latencies: number[] = [];
let hits = 0;
try {
for (let i = 0; i < ECHO_SAMPLES; i += 1) {
const char = ECHO_ALPHABET[i % ECHO_ALPHABET.length];
slot.awaitedChar = char;
slot.arrivalAt = null;
const sentAt = performance.now();
client.sendTerminalInput(terminalId, { type: "input", data: char });
const deadline = sentAt + ECHO_FRAME_TIMEOUT_MS;
while (slot.arrivalAt === null && performance.now() < deadline) {
await sleep(1);
}
if (slot.arrivalAt !== null) {
latencies.push(slot.arrivalAt - sentAt);
hits += 1;
}
slot.awaitedChar = null;
await sleep(ECHO_INTERVAL_MS);
}
} finally {
unsubscribe();
}
return { samples: ECHO_SAMPLES, hits, latencyMs: summarize(latencies) };
}
const BURST_TARGET_BYTES = 2_000_000;
const BURST_CHUNK = 16_384;
const BURST_DRAIN_TIMEOUT_MS = 20_000;
const BURST_QUIET_MS = 400;
interface BurstResult {
bytesSeen: number;
outputFrames: number;
snapshotFrames: number;
restoreFrames: number;
drainMs: number;
interFrameGapMs: Stats;
}
/**
* Runs a node one-liner in the terminal that prints ~2MB rapidly, then records
* inter-output-frame gaps until the stream goes quiet. Snapshot/restore frames
* during the burst indicate the snapshot-churn path firing.
*/
async function measureBurst(client: DaemonClientLike, terminalId: string): Promise<BurstResult> {
let bytesSeen = 0;
let outputFrames = 0;
let snapshotFrames = 0;
let restoreFrames = 0;
const gaps: number[] = [];
let lastBurstFrameAt = 0;
let firstBurstFrameAt = 0;
let lastActivityAt = performance.now();
// Frames smaller than this are the echoed command / prompt noise, not burst
// payload — gap timing only starts once real drain data is flowing, so the
// node startup latency doesn't pollute the inter-frame jitter.
const BURST_FRAME_MIN_BYTES = 1_024;
const unsubscribe = client.onTerminalStreamEvent((event) => {
if (event.terminalId !== terminalId) {
return;
}
const now = performance.now();
lastActivityAt = now;
if (event.type === "snapshot") {
snapshotFrames += 1;
return;
}
if (event.type === "restore") {
restoreFrames += 1;
return;
}
if (event.type !== "output" || !event.data) {
return;
}
outputFrames += 1;
bytesSeen += event.data.byteLength;
if (event.data.byteLength < BURST_FRAME_MIN_BYTES && firstBurstFrameAt === 0) {
return;
}
if (firstBurstFrameAt === 0) {
firstBurstFrameAt = now;
} else {
gaps.push(now - lastBurstFrameAt);
}
lastBurstFrameAt = now;
});
// Clear whatever the echo measurement left on the shell input line (Ctrl-C),
// then let the fresh prompt settle before sending the burst command.
client.sendTerminalInput(terminalId, { type: "input", data: "" });
await sleep(300);
const startedAt = performance.now();
try {
const chunks = Math.ceil(BURST_TARGET_BYTES / BURST_CHUNK);
// Print BURST_TARGET_BYTES of 'x' in chunks, no trailing newline spam.
const oneLiner =
`const c='x'.repeat(${BURST_CHUNK});` +
`for(let i=0;i<${chunks};i++)process.stdout.write(c);` +
`process.stdout.write('\\nBURST_DONE\\n')`;
client.sendTerminalInput(terminalId, {
type: "input",
data: `node -e "${oneLiner}"\r`,
});
const deadline = performance.now() + BURST_DRAIN_TIMEOUT_MS;
while (performance.now() < deadline) {
await sleep(50);
const quietFor = performance.now() - lastActivityAt;
// The daemon coalesces a large pty write into a snapshot plus a handful of
// frames, so the client rarely sees the full BURST_TARGET_BYTES. Exit once
// real burst data has arrived and the stream has gone quiet.
if (firstBurstFrameAt > 0 && quietFor >= BURST_QUIET_MS) {
break;
}
}
} finally {
unsubscribe();
}
return {
bytesSeen,
outputFrames,
snapshotFrames,
restoreFrames,
drainMs: round((lastBurstFrameAt || startedAt) - startedAt),
interFrameGapMs: summarize(gaps),
};
}
const PING_INTERVAL_MS = 100;
/**
* Samples RPC RTT at 10Hz for `durationMs`. Uses `ping` if available, else a
* cheap listTerminals request. Returns RTT stats — a proxy for main-loop delay.
*/
async function measurePing(client: DaemonClientLike, durationMs: number): Promise<Stats> {
const rtts: number[] = [];
const deadline = Date.now() + durationMs;
while (Date.now() < deadline) {
const start = performance.now();
try {
const result = await client.ping({ timeoutMs: 5_000 });
rtts.push(typeof result?.rttMs === "number" ? result.rttMs : performance.now() - start);
} catch {
const fallbackStart = performance.now();
await client.listTerminals().catch(() => undefined);
rtts.push(performance.now() - fallbackStart);
}
await sleep(PING_INTERVAL_MS);
}
return summarize(rtts);
}
// --- load generation --------------------------------------------------------
interface LoadController {
agentIds: string[];
noisyTerminalId: string | null;
stop(): Promise<void>;
}
interface LevelSpec {
id: string;
description: string;
mockAgents: number;
bigDiffEvery2s: boolean;
secondNoisyTerminal: boolean;
}
const LEVELS: LevelSpec[] = [
{
id: "L0",
description: "idle",
mockAgents: 0,
bigDiffEvery2s: false,
secondNoisyTerminal: false,
},
{
id: "L1",
description: "2 mock agents streaming",
mockAgents: 2,
bigDiffEvery2s: false,
secondNoisyTerminal: false,
},
{
id: "L2",
description: "4 mock agents + 200KB diff bursts every 2s",
mockAgents: 4,
bigDiffEvery2s: true,
secondNoisyTerminal: false,
},
];
if (process.env.BENCH_INCLUDE_L3 === "1") {
LEVELS.push({
id: "L3",
description: "L2 + a second terminal continuously emitting",
mockAgents: 4,
bigDiffEvery2s: true,
secondNoisyTerminal: true,
});
}
async function startLoad(
client: DaemonClientLike,
cwd: string,
workspaceId: string,
spec: LevelSpec,
): Promise<LoadController> {
const agentIds: string[] = [];
for (let i = 0; i < spec.mockAgents; i += 1) {
const agent = await client.createAgent({
provider: "mock",
cwd,
workspaceId,
model: "five-minute-stream",
title: `${spec.id}-load-${i}`,
});
agentIds.push(agent.id);
// Kick off the sustained stream.
await client.sendAgentMessage(agent.id, "start streaming").catch(() => undefined);
}
let bigDiffTimer: ReturnType<typeof setInterval> | null = null;
if (spec.bigDiffEvery2s && agentIds.length > 0) {
bigDiffTimer = setInterval(() => {
const target = agentIds[0];
void client
.sendAgentMessage(target, "emit 200000 byte large diff agent stream update")
.catch(() => undefined);
}, 2_000);
bigDiffTimer.unref?.();
}
let noisyTerminalId: string | null = null;
if (spec.secondNoisyTerminal) {
const created = await client.createTerminal(cwd, `${spec.id}-noisy`, undefined, {
workspaceId,
});
if (created.terminal) {
noisyTerminalId = created.terminal.id;
await client.subscribeTerminal(noisyTerminalId);
// Continuous infinite emit loop in the shell.
client.sendTerminalInput(noisyTerminalId, {
type: "input",
data: `node -e "setInterval(()=>process.stdout.write('y'.repeat(4096)),20)"\r`,
});
}
}
// Let the streams warm up before we start measuring.
await sleep(500);
return {
agentIds,
noisyTerminalId,
async stop() {
if (bigDiffTimer) {
clearInterval(bigDiffTimer);
}
if (noisyTerminalId) {
// Ctrl-C then kill the terminal.
client.sendTerminalInput(noisyTerminalId, { type: "input", data: "" });
await client.killTerminal(noisyTerminalId).catch(() => undefined);
}
for (const agentId of agentIds) {
if (client.cancelAgent) {
await client.cancelAgent(agentId).catch(() => undefined);
}
if (client.deleteAgent) {
await client.deleteAgent(agentId).catch(() => undefined);
}
}
},
};
}
// --- per-level run ----------------------------------------------------------
interface LevelResult {
id: string;
description: string;
echo: EchoResult;
burst: BurstResult;
pingMs: Stats;
}
async function runLevel(
client: DaemonClientLike,
cwd: string,
workspaceId: string,
spec: LevelSpec,
measuredTerminalId: string,
): Promise<LevelResult> {
console.log(`\n=== ${spec.id} (${spec.description}) ===`);
const load = await startLoad(client, cwd, workspaceId, spec);
try {
// Ping sampling runs concurrently with the echo + burst measurements so it
// observes the daemon under the same load the latency primitives see.
let pingMs: Stats = summarize([]);
const pingDone = (async () => {
pingMs = await measurePing(client, 14_000);
})();
console.log(` measuring echo latency (${ECHO_SAMPLES} samples)...`);
const echo = await measureEcho(client, measuredTerminalId);
console.log(
` echo p50=${echo.latencyMs.p50}ms p95=${echo.latencyMs.p95}ms max=${echo.latencyMs.max}ms (${echo.hits}/${echo.samples} hits)`,
);
console.log(" measuring output burst/jitter (~2MB)...");
const burst = await measureBurst(client, measuredTerminalId);
console.log(
` burst drain=${burst.drainMs}ms frames=${burst.outputFrames} gap_p95=${burst.interFrameGapMs.p95}ms snap=${burst.snapshotFrames} restore=${burst.restoreFrames}`,
);
await pingDone;
console.log(` ping p50=${pingMs.p50}ms p95=${pingMs.p95}ms max=${pingMs.max}ms`);
return { id: spec.id, description: spec.description, echo, burst, pingMs };
} finally {
await load.stop();
}
}
// --- output -----------------------------------------------------------------
function pad(value: string | number, width: number): string {
return String(value).padStart(width);
}
function printTable(results: LevelResult[]): void {
console.log("\n========== TERMINAL LATENCY BENCHMARK ==========\n");
const header = [
pad("level", 6),
pad("echo_p50", 9),
pad("echo_p95", 9),
pad("echo_max", 9),
pad("ping_p50", 9),
pad("ping_p95", 9),
pad("gap_p95", 8),
pad("gap_max", 8),
pad("drain", 8),
pad("frames", 7),
pad("snap", 5),
pad("restore", 8),
].join(" ");
console.log(header);
console.log("-".repeat(header.length));
for (const r of results) {
console.log(
[
pad(r.id, 6),
pad(r.echo.latencyMs.p50, 9),
pad(r.echo.latencyMs.p95, 9),
pad(r.echo.latencyMs.max, 9),
pad(r.pingMs.p50, 9),
pad(r.pingMs.p95, 9),
pad(r.burst.interFrameGapMs.p95, 8),
pad(r.burst.interFrameGapMs.max, 8),
pad(r.burst.drainMs, 8),
pad(r.burst.outputFrames, 7),
pad(r.burst.snapshotFrames, 5),
pad(r.burst.restoreFrames, 8),
].join(" "),
);
}
console.log("");
for (const r of results) {
console.log(` ${r.id}: ${r.description}`);
}
console.log("");
}
function getCommitHash(): string {
try {
return execSync("git rev-parse --short HEAD", { cwd: REPO_ROOT }).toString().trim();
} catch {
return "unknown";
}
}
// --- main -------------------------------------------------------------------
async function main(): Promise<void> {
const commit = getCommitHash();
console.log(`Terminal latency benchmark — commit ${commit}`);
console.log("Booting isolated daemon (random port, fresh PASEO_HOME)...");
let daemon: BootedDaemon | null = null;
let client: DaemonClientLike | null = null;
let workspaceDir: string | null = null;
let measuredTerminalId: string | null = null;
try {
daemon = await bootDaemon();
console.log(`Daemon ready: pid=${daemon.pid} port=${daemon.port} home=${daemon.paseoHome}`);
const ctor = await loadDaemonClientCtor();
client = await connectClient(ctor, daemon.port);
workspaceDir = await mkdtemp(path.join(os.tmpdir(), "paseo-bench-ws-"));
const opened = await client.openProject(workspaceDir);
if (!opened.workspace) {
throw new Error(`Failed to open project: ${opened.error}`);
}
const workspaceId = opened.workspace.id;
const created = await client.createTerminal(workspaceDir, "measured", undefined, {
workspaceId,
});
if (!created.terminal) {
throw new Error(`Failed to create terminal: ${created.error}`);
}
measuredTerminalId = created.terminal.id;
await client.subscribeTerminal(measuredTerminalId);
// Let the shell print its prompt and settle before measuring.
await sleep(750);
const results: LevelResult[] = [];
for (const spec of LEVELS) {
const result = await runLevel(client, workspaceDir, workspaceId, spec, measuredTerminalId);
results.push(result);
// Settle between levels so a previous level's load fully drains.
await sleep(2_000);
}
printTable(results);
const outDir = "/tmp/paseo-terminal-bench";
await mkdir(outDir, { recursive: true });
const timestamp = new Date().toISOString().replace(/[:.]/g, "-");
const outPath = path.join(outDir, `${timestamp}.json`);
await writeFile(
outPath,
JSON.stringify(
{
commit,
daemonPid: daemon.pid,
port: daemon.port,
paseoHome: daemon.paseoHome,
node: process.version,
platform: `${os.platform()} ${os.arch()}`,
createdAt: new Date().toISOString(),
levels: results,
},
null,
2,
),
);
console.log(`Wrote results to ${outPath}`);
} finally {
if (client && measuredTerminalId) {
await client.killTerminal(measuredTerminalId).catch(() => undefined);
}
if (client) {
await client.close().catch(() => undefined);
}
if (daemon) {
daemon.child.kill("SIGTERM");
await Promise.race([
new Promise<void>((resolve) => daemon!.child.once("exit", () => resolve())),
sleep(5_000),
]);
if (daemon.child.exitCode === null && daemon.child.signalCode === null) {
daemon.child.kill("SIGKILL");
}
await rm(daemon.paseoHome, { recursive: true, force: true }).catch(() => undefined);
}
if (workspaceDir) {
await rm(workspaceDir, { recursive: true, force: true }).catch(() => undefined);
}
}
}
main().catch((error) => {
console.error(error);
process.exit(1);
});