mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
perf(agents): isolate concurrency benchmark workers (#119749)
This commit is contained in:
Vincent Koc
@@ -117,6 +117,8 @@ const rootEntries = [
|
||||
"scripts/print-cli-backend-live-metadata.ts!",
|
||||
// Workflow/package-script entrypoints are not imported from production modules.
|
||||
"scripts/openclaw-cross-os-release-checks.ts!",
|
||||
// Spawned by the agent concurrency benchmark; no static import edge exists.
|
||||
"scripts/bench-agent-concurrency-worker.ts!",
|
||||
"scripts/bench-sqlite-reliability.ts!",
|
||||
// Docker/manual E2E executables and their nested assertion/probe entrypoints.
|
||||
"scripts/e2e/*.{js,mjs,ts}!",
|
||||
|
||||
@@ -0,0 +1,810 @@
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import type { SubagentRunRecord } from "../src/agents/subagent-registry.types.js";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../src/state/openclaw-state-db.generated.js";
|
||||
import {
|
||||
WORKER_RESULT_SENTINEL,
|
||||
type WorkerResult,
|
||||
type WorkerScenario,
|
||||
} from "./bench-agent-concurrency.js";
|
||||
|
||||
type WorkerOptions = {
|
||||
scenario: WorkerScenario;
|
||||
size: number;
|
||||
runs: number;
|
||||
warmup: number;
|
||||
};
|
||||
|
||||
type Sample = {
|
||||
durationMs: number;
|
||||
invariant: Record<string, number | boolean>;
|
||||
};
|
||||
|
||||
const SCENARIOS = new Set<WorkerScenario>([
|
||||
"spawnPipelineInMemory",
|
||||
"spawnPipelineDurable",
|
||||
"admission",
|
||||
"recoverySweep",
|
||||
"duplicateSuppression",
|
||||
]);
|
||||
|
||||
function parseInteger(raw: string | undefined, flag: string, min: number, max: number): number {
|
||||
if (!raw || !/^\d+$/u.test(raw)) {
|
||||
throw new Error(`${flag} must be an integer`);
|
||||
}
|
||||
const value = Number(raw);
|
||||
if (value < min || value > max) {
|
||||
throw new Error(`${flag} must be between ${min} and ${max}`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function parseOptions(argv: string[]): WorkerOptions {
|
||||
const values = new Map<string, string>();
|
||||
for (let index = 0; index < argv.length; index += 2) {
|
||||
const flag = argv[index];
|
||||
const value = argv[index + 1];
|
||||
if (!flag?.startsWith("--") || !value || value.startsWith("--")) {
|
||||
throw new Error(`invalid worker argument near ${flag ?? "end"}`);
|
||||
}
|
||||
if (values.has(flag)) {
|
||||
throw new Error(`${flag} was provided more than once`);
|
||||
}
|
||||
values.set(flag, value);
|
||||
}
|
||||
const scenario = values.get("--scenario") as WorkerScenario | undefined;
|
||||
if (!scenario || !SCENARIOS.has(scenario)) {
|
||||
throw new Error(`unknown worker scenario: ${scenario ?? "missing"}`);
|
||||
}
|
||||
return {
|
||||
scenario,
|
||||
size: parseInteger(values.get("--size"), "--size", 1, 4096),
|
||||
runs: parseInteger(values.get("--runs"), "--runs", 1, 100),
|
||||
warmup: parseInteger(values.get("--warmup"), "--warmup", 0, 20),
|
||||
};
|
||||
}
|
||||
|
||||
function processMaxRssBytes(): number {
|
||||
return Math.max(0, Math.round(process.resourceUsage().maxRSS * 1024));
|
||||
}
|
||||
|
||||
function toError(error: unknown): Error {
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
|
||||
async function waitForCondition(check: () => boolean): Promise<boolean> {
|
||||
const deadline = Date.now() + 30_000;
|
||||
while (Date.now() < deadline) {
|
||||
if (check()) {
|
||||
return true;
|
||||
}
|
||||
await new Promise<void>((resolve) => {
|
||||
setImmediate(resolve);
|
||||
});
|
||||
}
|
||||
return check();
|
||||
}
|
||||
|
||||
async function resetRuntime(persist: boolean): Promise<void> {
|
||||
const [subagents, tasks, stateDb, agentDb] = await Promise.all([
|
||||
import("../src/agents/subagent-registry.test-helpers.js"),
|
||||
import("../src/tasks/task-runtime.test-helpers.js"),
|
||||
import("../src/state/openclaw-state-db.js"),
|
||||
import("../src/state/openclaw-agent-db.js"),
|
||||
]);
|
||||
subagents.resetSubagentRegistryForTests({ persist });
|
||||
subagents.testing.setDepsForTest();
|
||||
tasks.resetTaskRegistryControlRuntimeForTests();
|
||||
tasks.resetTaskRegistryDeliveryRuntimeForTests();
|
||||
tasks.resetDetachedTaskLifecycleRuntimeForTests();
|
||||
tasks.resetTaskRegistryForTests({ persist });
|
||||
tasks.resetTaskFlowRegistryForTests({ persist });
|
||||
stateDb.closeOpenClawStateDatabaseForTest();
|
||||
agentDb.closeOpenClawAgentDatabasesForTest();
|
||||
}
|
||||
|
||||
function createTerminalWaitBarrier() {
|
||||
const waiters = new Map<string, (value: unknown) => void>();
|
||||
let outstanding = 0;
|
||||
let releasesStarted = false;
|
||||
const callGateway = (async <T>(request: { method?: string; params?: unknown }) => {
|
||||
if (request.method !== "agent.wait") {
|
||||
return {} as T;
|
||||
}
|
||||
if (releasesStarted) {
|
||||
throw new Error("agent.wait registered after the benchmark barrier released");
|
||||
}
|
||||
const runId =
|
||||
request.params &&
|
||||
typeof request.params === "object" &&
|
||||
!Array.isArray(request.params) &&
|
||||
typeof (request.params as { runId?: unknown }).runId === "string"
|
||||
? (request.params as { runId: string }).runId
|
||||
: undefined;
|
||||
if (!runId || waiters.has(runId)) {
|
||||
throw new Error(`invalid or duplicate benchmark agent.wait run id: ${runId ?? "missing"}`);
|
||||
}
|
||||
outstanding += 1;
|
||||
return await new Promise<T>((resolve) => {
|
||||
waiters.set(runId, (value) => {
|
||||
outstanding -= 1;
|
||||
resolve(value as T);
|
||||
});
|
||||
});
|
||||
}) as typeof import("../src/gateway/call.js").callGateway;
|
||||
return {
|
||||
callGateway,
|
||||
get outstanding() {
|
||||
return outstanding;
|
||||
},
|
||||
async waitUntilBlocked(expected: number) {
|
||||
return await waitForCondition(() => outstanding === expected);
|
||||
},
|
||||
release(runId: string) {
|
||||
releasesStarted = true;
|
||||
const resolve = waiters.get(runId);
|
||||
if (!resolve) {
|
||||
throw new Error(`missing blocked benchmark agent.wait for ${runId}`);
|
||||
}
|
||||
waiters.delete(runId);
|
||||
const endedAt = Date.now();
|
||||
resolve({
|
||||
status: "ok",
|
||||
startedAt: endedAt - 1,
|
||||
endedAt,
|
||||
stopReason: "stop",
|
||||
});
|
||||
},
|
||||
releaseAll() {
|
||||
releasesStarted = true;
|
||||
for (const runId of waiters.keys()) {
|
||||
this.release(runId);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function configureSpawnRuntime(
|
||||
mode: "memory" | "durable",
|
||||
callGateway: typeof import("../src/gateway/call.js").callGateway,
|
||||
): Promise<void> {
|
||||
const [subagents, registry, taskStore, flowStore] = await Promise.all([
|
||||
import("../src/agents/subagent-registry.test-helpers.js"),
|
||||
import("../src/agents/subagent-registry-memory.js"),
|
||||
import("../src/tasks/task-registry.store.js"),
|
||||
import("../src/tasks/task-flow-registry.store.test-support.js"),
|
||||
]);
|
||||
const sharedDeps = {
|
||||
callGateway,
|
||||
getRuntimeConfig: () => ({}),
|
||||
onAgentEvent: () => () => {},
|
||||
resolveAgentTimeoutMs: () => 1_000,
|
||||
captureSubagentCompletionReply: async (childSessionKey: string) => {
|
||||
const entry = [...registry.subagentRuns.values()].find(
|
||||
(candidate) => candidate.childSessionKey === childSessionKey,
|
||||
);
|
||||
if (entry) {
|
||||
// Completion already owns the row at this awaited seam. Suppress only
|
||||
// its unrelated session projection, not the terminal registry/task transition.
|
||||
entry.execution.suppressSessionEffects = true;
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
cleanupBrowserSessionsForLifecycleEnd: async () => {},
|
||||
runSubagentAnnounceFlow: async () => false,
|
||||
maybeWakeRequesterAfterAllChildrenSettled: async () => false,
|
||||
ensureContextEnginesInitialized: () => {},
|
||||
loadAgentRuntimePluginRegistryHandle: () => undefined,
|
||||
resolveContextEngine: async () =>
|
||||
({
|
||||
info: { id: "bench", name: "bench", version: "1" },
|
||||
ingest: async () => ({ ok: true }),
|
||||
assemble: async () => ({ messages: [] }),
|
||||
onSubagentEnded: async () => {},
|
||||
}) as unknown as import("../src/context-engine/types.js").ContextEngine,
|
||||
};
|
||||
if (mode === "memory") {
|
||||
subagents.testing.setDepsForTest({
|
||||
...sharedDeps,
|
||||
persistSubagentRunsToDisk: () => {},
|
||||
persistSubagentRunsToDiskOrThrow: () => {},
|
||||
});
|
||||
taskStore.configureTaskRegistryRuntime({
|
||||
store: {
|
||||
loadSnapshot: () => ({ tasks: new Map(), deliveryStates: new Map() }),
|
||||
saveSnapshot: () => {},
|
||||
upsertTaskWithDeliveryState: () => {},
|
||||
upsertTask: () => {},
|
||||
deleteTaskWithDeliveryState: () => {},
|
||||
deleteTask: () => {},
|
||||
upsertDeliveryState: () => {},
|
||||
deleteDeliveryState: () => {},
|
||||
close: () => {},
|
||||
},
|
||||
});
|
||||
flowStore.configureTaskFlowRegistryRuntime({
|
||||
store: {
|
||||
loadSnapshot: () => ({ flows: new Map() }),
|
||||
saveSnapshot: () => {},
|
||||
upsertFlow: () => {},
|
||||
deleteFlow: () => {},
|
||||
close: () => {},
|
||||
},
|
||||
});
|
||||
return;
|
||||
}
|
||||
subagents.testing.setDepsForTest(sharedDeps);
|
||||
}
|
||||
|
||||
type BenchmarkStateDatabase = Pick<OpenClawStateKyselyDatabase, "subagent_runs" | "task_runs">;
|
||||
|
||||
async function readDurableRows() {
|
||||
const [{ executeSqliteQuerySync, getNodeSqliteKysely }, stateDb] = await Promise.all([
|
||||
import("../src/infra/kysely-sync.js"),
|
||||
import("../src/state/openclaw-state-db.js"),
|
||||
]);
|
||||
const database = stateDb.openOpenClawStateDatabase();
|
||||
const db = getNodeSqliteKysely<BenchmarkStateDatabase>(database.db);
|
||||
return {
|
||||
path: database.path,
|
||||
subagentRows: executeSqliteQuerySync(
|
||||
database.db,
|
||||
db.selectFrom("subagent_runs").select(["run_id", "ended_at"]).orderBy("run_id"),
|
||||
).rows,
|
||||
taskRows: executeSqliteQuerySync(
|
||||
database.db,
|
||||
db
|
||||
.selectFrom("task_runs")
|
||||
.select(["run_id", "status", "ended_at"])
|
||||
.where("runtime", "=", "subagent")
|
||||
.orderBy("run_id"),
|
||||
).rows,
|
||||
};
|
||||
}
|
||||
|
||||
function listSqliteFiles(root: string): string[] {
|
||||
return fs
|
||||
.readdirSync(root, { recursive: true })
|
||||
.map(String)
|
||||
.filter((entry) => entry.includes(".sqlite"))
|
||||
.toSorted();
|
||||
}
|
||||
|
||||
async function listBenchmarkTasks() {
|
||||
const tasks = await import("../src/tasks/task-registry.js");
|
||||
return tasks.listTaskRecords().filter((task) => task.runtime === "subagent");
|
||||
}
|
||||
|
||||
async function listBenchmarkTaskMemory() {
|
||||
const state = await import("../src/tasks/task-registry-state.js");
|
||||
return [...state.tasks.values()].filter((task) => task.runtime === "subagent");
|
||||
}
|
||||
|
||||
function assertExactRunIds(actual: Array<string | null>, expected: string[], label: string): void {
|
||||
const normalized = actual
|
||||
.filter((value): value is string => typeof value === "string")
|
||||
.toSorted();
|
||||
const wanted = expected.toSorted();
|
||||
if (JSON.stringify(normalized) !== JSON.stringify(wanted)) {
|
||||
throw new Error(
|
||||
`${label} mismatch: ${JSON.stringify({ actual: normalized, expected: wanted })}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function runSpawnSample(
|
||||
fanout: number,
|
||||
serial: number,
|
||||
mode: "memory" | "durable",
|
||||
stateDir: string,
|
||||
): Promise<Sample> {
|
||||
const [pipeline, registry] = await Promise.all([
|
||||
import("../src/agents/spawn-pipeline.js"),
|
||||
import("../src/agents/subagent-registry-memory.js"),
|
||||
]);
|
||||
await resetRuntime(mode === "durable");
|
||||
const barrier = createTerminalWaitBarrier();
|
||||
await configureSpawnRuntime(mode, barrier.callGateway);
|
||||
const taskRegistry = await import("../src/tasks/task-registry.js");
|
||||
let releases = 0;
|
||||
const runIds: string[] = [];
|
||||
const params = Array.from({ length: fanout }, (_, index) => {
|
||||
const runId = `bench-${mode}-${serial}-${index}`;
|
||||
runIds.push(runId);
|
||||
let released = false;
|
||||
return {
|
||||
adapter: {
|
||||
initialize: async () => ({ index }),
|
||||
dispatchTurn: async () => ({ runId }),
|
||||
cleanupOnFailure: async () => {},
|
||||
},
|
||||
admissionReservation: {
|
||||
release: () => {
|
||||
if (!released) {
|
||||
released = true;
|
||||
releases += 1;
|
||||
}
|
||||
},
|
||||
},
|
||||
buildRegistration: () => ({
|
||||
runId,
|
||||
childSessionKey: `agent:bench:subagent:${mode}:${serial}:${index}`,
|
||||
requesterSessionKey: "agent:bench:main",
|
||||
requesterDisplayKey: "bench",
|
||||
task: `benchmark child ${index}`,
|
||||
cleanup: "keep" as const,
|
||||
expectsCompletionMessage: false,
|
||||
}),
|
||||
progressSessionKey: "agent:bench:main",
|
||||
};
|
||||
});
|
||||
let result: Sample | undefined;
|
||||
let failure: unknown;
|
||||
try {
|
||||
const startedAt = performance.now();
|
||||
const pipelineResults = await Promise.all(
|
||||
params.map((entry) => pipeline.runSpawnPipeline(entry)),
|
||||
);
|
||||
const durationMs = performance.now() - startedAt;
|
||||
const blocked = await barrier.waitUntilBlocked(fanout);
|
||||
const successful = pipelineResults.filter((pipelineResult) => pipelineResult.ok);
|
||||
const uniqueRuns = new Set(successful.map((pipelineResult) => pipelineResult.runId)).size;
|
||||
const registeredRuns = registry.subagentRuns.size;
|
||||
const tasksWhileBlocked = await listBenchmarkTasks();
|
||||
assertExactRunIds(
|
||||
tasksWhileBlocked.map((task) => task.runId ?? null),
|
||||
runIds,
|
||||
`${mode} in-memory task rows`,
|
||||
);
|
||||
if (!blocked || barrier.outstanding !== fanout) {
|
||||
throw new Error(`spawn ${mode} did not block ${fanout} agent.wait calls`);
|
||||
}
|
||||
if (
|
||||
successful.length !== fanout ||
|
||||
uniqueRuns !== fanout ||
|
||||
registeredRuns !== fanout ||
|
||||
releases !== fanout
|
||||
) {
|
||||
throw new Error(
|
||||
`spawn ${mode} registration invariant failed: ${JSON.stringify({ fanout, uniqueRuns, registeredRuns, releases })}`,
|
||||
);
|
||||
}
|
||||
let durableSubagentRows = 0;
|
||||
let durableTaskRows = 0;
|
||||
let durableStateFile = listSqliteFiles(stateDir).length > 0;
|
||||
if (mode === "durable") {
|
||||
const durable = await readDurableRows();
|
||||
assertExactRunIds(
|
||||
durable.subagentRows.map((row) => row.run_id),
|
||||
runIds,
|
||||
"durable subagent rows",
|
||||
);
|
||||
assertExactRunIds(
|
||||
durable.taskRows.map((row) => row.run_id),
|
||||
runIds,
|
||||
"durable task rows",
|
||||
);
|
||||
if (
|
||||
durable.subagentRows.some((row) => row.ended_at !== null) ||
|
||||
durable.taskRows.some((row) => row.status !== "running" || row.ended_at !== null)
|
||||
) {
|
||||
throw new Error("durable rows settled before the benchmark barrier released");
|
||||
}
|
||||
durableSubagentRows = durable.subagentRows.length;
|
||||
durableTaskRows = durable.taskRows.length;
|
||||
durableStateFile = fs.existsSync(durable.path);
|
||||
if (!durableStateFile) {
|
||||
throw new Error(`durable spawn pipeline database is missing at ${durable.path}`);
|
||||
}
|
||||
} else if (durableStateFile) {
|
||||
throw new Error("in-memory spawn pipeline created durable SQLite state");
|
||||
}
|
||||
for (const runId of runIds) {
|
||||
barrier.release(runId);
|
||||
const runSettled = await waitForCondition(() => {
|
||||
const run = registry.subagentRuns.get(runId);
|
||||
const task = taskRegistry.findTaskByRunId(runId);
|
||||
return (
|
||||
run?.execution.status === "terminal" &&
|
||||
typeof run.execution.endedAt === "number" &&
|
||||
typeof run.cleanupCompletedAt === "number" &&
|
||||
task?.status === "succeeded"
|
||||
);
|
||||
});
|
||||
if (!runSettled) {
|
||||
throw new Error(`spawn ${mode} did not settle released run ${runId}`);
|
||||
}
|
||||
}
|
||||
const settledTasks = await listBenchmarkTasks();
|
||||
const settledRuns = [...registry.subagentRuns.values()].filter(
|
||||
(entry) =>
|
||||
entry.execution.status === "terminal" &&
|
||||
typeof entry.execution.endedAt === "number" &&
|
||||
typeof entry.cleanupCompletedAt === "number",
|
||||
).length;
|
||||
const succeededTasks = settledTasks.filter((task) => task.status === "succeeded").length;
|
||||
if (settledRuns !== fanout || succeededTasks !== fanout || barrier.outstanding !== 0) {
|
||||
throw new Error(
|
||||
`spawn ${mode} settlement invariant failed: ${JSON.stringify({ fanout, settledRuns, succeededTasks, outstandingWaits: barrier.outstanding })}`,
|
||||
);
|
||||
}
|
||||
result = {
|
||||
durationMs,
|
||||
invariant: {
|
||||
ok: true,
|
||||
registeredRuns,
|
||||
reservationsReleased: releases,
|
||||
blockedWaits: fanout,
|
||||
settledRuns,
|
||||
settledTasks: succeededTasks,
|
||||
outstandingWaits: barrier.outstanding,
|
||||
durableSubagentRows,
|
||||
durableTaskRows,
|
||||
durableStateFile,
|
||||
postTeardownRegistryRows: -1,
|
||||
postTeardownTaskRows: -1,
|
||||
postTeardownDurableSubagentRows: -1,
|
||||
postTeardownDurableTaskRows: -1,
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
} finally {
|
||||
try {
|
||||
barrier.releaseAll();
|
||||
} catch (error) {
|
||||
failure ??= error;
|
||||
}
|
||||
try {
|
||||
await resetRuntime(mode === "durable");
|
||||
} catch (error) {
|
||||
failure ??= error;
|
||||
}
|
||||
}
|
||||
if (failure) {
|
||||
throw toError(failure);
|
||||
}
|
||||
const postTeardownTasks = await listBenchmarkTaskMemory();
|
||||
const postTeardownRegistryRows = registry.subagentRuns.size;
|
||||
const postTeardownTaskRows = postTeardownTasks.length;
|
||||
let postTeardownDurableSubagentRows = 0;
|
||||
let postTeardownDurableTaskRows = 0;
|
||||
if (mode === "durable") {
|
||||
const durable = await readDurableRows();
|
||||
postTeardownDurableSubagentRows = durable.subagentRows.length;
|
||||
postTeardownDurableTaskRows = durable.taskRows.length;
|
||||
await resetRuntime(false);
|
||||
}
|
||||
if (
|
||||
postTeardownRegistryRows !== 0 ||
|
||||
postTeardownTaskRows !== 0 ||
|
||||
postTeardownDurableSubagentRows !== 0 ||
|
||||
postTeardownDurableTaskRows !== 0 ||
|
||||
barrier.outstanding !== 0
|
||||
) {
|
||||
throw new Error(
|
||||
`spawn ${mode} teardown invariant failed: ${JSON.stringify({ postTeardownRegistryRows, postTeardownTaskRows, postTeardownDurableSubagentRows, postTeardownDurableTaskRows, outstandingWaits: barrier.outstanding })}`,
|
||||
);
|
||||
}
|
||||
if (mode === "memory" && listSqliteFiles(stateDir).length > 0) {
|
||||
throw new Error("in-memory spawn pipeline created SQLite state during teardown");
|
||||
}
|
||||
if (!result) {
|
||||
throw new Error(`spawn ${mode} did not produce a benchmark result`);
|
||||
}
|
||||
result.invariant.postTeardownRegistryRows = postTeardownRegistryRows;
|
||||
result.invariant.postTeardownTaskRows = postTeardownTaskRows;
|
||||
result.invariant.postTeardownDurableSubagentRows = postTeardownDurableSubagentRows;
|
||||
result.invariant.postTeardownDurableTaskRows = postTeardownDurableTaskRows;
|
||||
return result;
|
||||
}
|
||||
|
||||
async function runAdmissionSample(fanout: number, serial: number): Promise<Sample> {
|
||||
const { reserveChildAdmissionSlot } = await import("../src/agents/child-admission.js");
|
||||
const controllerSessionKey = `agent:bench:admission:${serial}`;
|
||||
const reservations: Array<{ release: () => void }> = [];
|
||||
const reserve = () =>
|
||||
reserveChildAdmissionSlot({
|
||||
controllerSessionKey,
|
||||
resolveAdmission: (pendingChildren) =>
|
||||
pendingChildren < fanout
|
||||
? { ok: true as const }
|
||||
: { ok: false as const, governingCap: "benchmark" },
|
||||
});
|
||||
const startedAt = performance.now();
|
||||
for (let index = 0; index < fanout; index += 1) {
|
||||
const reservation = reserve();
|
||||
if (!reservation.ok) {
|
||||
throw new Error(`admission rejected slot ${index + 1}/${fanout}`);
|
||||
}
|
||||
reservations.push(reservation);
|
||||
}
|
||||
const overflow = reserve();
|
||||
for (const reservation of reservations) {
|
||||
reservation.release();
|
||||
reservation.release();
|
||||
}
|
||||
const replacement = reserve();
|
||||
if (replacement.ok) {
|
||||
replacement.release();
|
||||
}
|
||||
const durationMs = performance.now() - startedAt;
|
||||
const ok = !overflow.ok && replacement.ok;
|
||||
if (!ok) {
|
||||
throw new Error(`admission invariant failed at fanout ${fanout}`);
|
||||
}
|
||||
return {
|
||||
durationMs,
|
||||
invariant: { ok, admissionCap: reservations.length, overflowRejected: true, released: true },
|
||||
};
|
||||
}
|
||||
|
||||
function sweepRow(child: number, generation: number, now: number): SubagentRunRecord {
|
||||
const current = generation === 3;
|
||||
return {
|
||||
runId: `bench-sweep-${child}-${generation}`,
|
||||
childSessionKey: `agent:bench:subagent:sweep-${child}`,
|
||||
requesterSessionKey: "agent:bench:main",
|
||||
requesterDisplayKey: "bench",
|
||||
task: `sweep child ${child}`,
|
||||
cleanup: current ? "keep" : "delete",
|
||||
generation,
|
||||
createdAt: now - generation,
|
||||
archiveAtMs: current ? undefined : now - 1,
|
||||
terminalOwner: current ? "interrupted-recovery" : undefined,
|
||||
endedReason: current ? "subagent-error" : undefined,
|
||||
execution: current
|
||||
? {
|
||||
status: "terminal",
|
||||
startedAt: now - 2_000,
|
||||
endedAt: now - 1_000,
|
||||
outcome: { status: "error", error: "interrupted recovery replay" },
|
||||
suppressSessionEffects: true,
|
||||
}
|
||||
: {
|
||||
status: "terminal",
|
||||
startedAt: now - 2_000,
|
||||
endedAt: now - 1_000,
|
||||
outcome: { status: "error", error: "retired recovery generation" },
|
||||
suppressSessionEffects: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function runSweepSample(childCount: number): Promise<Sample> {
|
||||
const { createSubagentRegistrySweeper } =
|
||||
await import("../src/agents/subagent-registry-sweeper.js");
|
||||
const now = Date.now();
|
||||
const runs = new Map<string, SubagentRunRecord>();
|
||||
for (let child = 0; child < childCount; child += 1) {
|
||||
for (const generation of [3, 2, 1]) {
|
||||
const entry = sweepRow(child, generation, now);
|
||||
runs.set(entry.runId, entry);
|
||||
}
|
||||
}
|
||||
let sessionEffects = 0;
|
||||
let recoveryProjections = 0;
|
||||
let lostContextCompletions = 0;
|
||||
const sweeper = createSubagentRegistrySweeper({
|
||||
runs,
|
||||
resumedRuns: new Set(),
|
||||
persist: () => {},
|
||||
clearPendingLifecycleError: () => {},
|
||||
clearPendingLifecycleTimeout: () => {},
|
||||
sweepPendingLifecycle: () => {},
|
||||
completeSubagentRunWithRecovery: async () => {
|
||||
lostContextCompletions += 1;
|
||||
},
|
||||
getGatewayRecoveryRuntime: () => undefined,
|
||||
abandonSubagentRestartRecoveryLaunch: () => true,
|
||||
clearAcceptedSubagentRestartRecovery: () => true,
|
||||
resumeSettledSubagentRestartRecovery: () => true,
|
||||
replaceSubagentRunAfterSteer: () => true,
|
||||
markSubagentRestartRecoveryLaunchAttempted: () => undefined,
|
||||
markSubagentRestartRecoveryLaunchAccepted: () => undefined,
|
||||
markSubagentRestartRecoveryLaunchConsumed: () => undefined,
|
||||
reserveSubagentRestartRecoveryLaunch: () => undefined,
|
||||
resetSubagentRestartRecoveryLaunchAttempt: () => true,
|
||||
finalizeInterruptedSubagentRun: async ({ runId, expectedEntry }) => {
|
||||
if (runs.get(runId) !== expectedEntry || expectedEntry?.generation !== 3) {
|
||||
throw new Error(`unexpected recovery projection owner: ${runId}`);
|
||||
}
|
||||
recoveryProjections += 1;
|
||||
return 1;
|
||||
},
|
||||
resumeRequesterSettleWake: () => {},
|
||||
startSubagentAnnounceCleanupFlow: () => true,
|
||||
completeCleanupBookkeeping: () => {},
|
||||
shouldEmitEndedHookForRun: () => false,
|
||||
emitSubagentEndedHookForRun: async () => {},
|
||||
callGateway: (async <T>() => {
|
||||
sessionEffects += 1;
|
||||
return {} as T;
|
||||
}) as typeof import("../src/gateway/call.js").callGateway,
|
||||
cleanupCollectorLaunchResources: async () => true,
|
||||
runContextEngineSubagentEnded: async () => {
|
||||
sessionEffects += 1;
|
||||
},
|
||||
notifyContextEngineSubagentEnded: async () => {
|
||||
sessionEffects += 1;
|
||||
},
|
||||
retireSupersededRun: async () => {},
|
||||
getRunsForChildSession: (childSessionKey) =>
|
||||
[...runs.values()].filter((entry) => entry.childSessionKey === childSessionKey),
|
||||
getRunsForCollectorGroup: () => [],
|
||||
warn: () => {},
|
||||
});
|
||||
try {
|
||||
const startedAt = performance.now();
|
||||
await sweeper.sweepOnce();
|
||||
const durationMs = performance.now() - startedAt;
|
||||
const retainedCurrent = [...runs.values()].filter((entry) => entry.generation === 3).length;
|
||||
const removedRows = childCount * 3 - runs.size;
|
||||
const ok =
|
||||
removedRows === childCount * 2 &&
|
||||
retainedCurrent === childCount &&
|
||||
sessionEffects === 0 &&
|
||||
recoveryProjections === childCount &&
|
||||
lostContextCompletions === 0;
|
||||
if (!ok) {
|
||||
throw new Error(
|
||||
`registry sweep invariant failed: ${JSON.stringify({ childCount, removedRows, retainedCurrent, sessionEffects, recoveryProjections, lostContextCompletions })}`,
|
||||
);
|
||||
}
|
||||
return {
|
||||
durationMs,
|
||||
invariant: {
|
||||
ok,
|
||||
seededRows: childCount * 3,
|
||||
removedRows,
|
||||
retainedCurrent,
|
||||
sessionEffects,
|
||||
recoveryProjections,
|
||||
lostContextCompletions,
|
||||
},
|
||||
};
|
||||
} finally {
|
||||
sweeper.reset();
|
||||
}
|
||||
}
|
||||
|
||||
async function runDedupeSample(childCount: number): Promise<Sample> {
|
||||
const { dedupeLatestChildCompletionRows } =
|
||||
await import("../src/agents/subagent-announce-output.js");
|
||||
const rowsForOrder = (generations: number[]) =>
|
||||
Array.from({ length: childCount }, (_, child) =>
|
||||
generations.map((generation) => ({
|
||||
runId: `bench-dedupe-${child}-${generation}`,
|
||||
childSessionKey: `agent:bench:subagent:dedupe-${child}`,
|
||||
task: `dedupe child ${child}`,
|
||||
generation,
|
||||
createdAt: generation,
|
||||
execution: { status: "terminal" as const, endedAt: generation },
|
||||
})),
|
||||
).flat();
|
||||
const newestFirstRows = rowsForOrder([3, 2, 1]);
|
||||
const oldestFirstRows = rowsForOrder([1, 2, 3]);
|
||||
const startedAt = performance.now();
|
||||
const newestFirst = dedupeLatestChildCompletionRows(newestFirstRows);
|
||||
const oldestFirst = dedupeLatestChildCompletionRows(oldestFirstRows);
|
||||
const durationMs = performance.now() - startedAt;
|
||||
const newestFirstSelectedNewest = newestFirst.every((row) => row.generation === 3);
|
||||
const oldestFirstSelectedNewest = oldestFirst.every((row) => row.generation === 3);
|
||||
const ok =
|
||||
newestFirst.length === childCount &&
|
||||
oldestFirst.length === childCount &&
|
||||
newestFirstSelectedNewest &&
|
||||
oldestFirstSelectedNewest;
|
||||
if (!ok) {
|
||||
throw new Error(`completion dedupe invariant failed for ${childCount} children`);
|
||||
}
|
||||
return {
|
||||
durationMs,
|
||||
invariant: {
|
||||
ok,
|
||||
inputRowsPerOrdering: newestFirstRows.length,
|
||||
newestFirstSelectedRows: newestFirst.length,
|
||||
oldestFirstSelectedRows: oldestFirst.length,
|
||||
newestFirstSelectedNewest,
|
||||
oldestFirstSelectedNewest,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function runScenario(options: WorkerOptions, stateDir: string): Promise<WorkerResult> {
|
||||
const timingsMs: number[] = [];
|
||||
let invariant: Record<string, number | boolean> = {};
|
||||
const rssStartBytes = process.memoryUsage().rss;
|
||||
for (let index = 0; index < options.warmup + options.runs; index += 1) {
|
||||
let sample: Sample;
|
||||
if (options.scenario === "spawnPipelineInMemory") {
|
||||
sample = await runSpawnSample(options.size, index, "memory", stateDir);
|
||||
} else if (options.scenario === "spawnPipelineDurable") {
|
||||
sample = await runSpawnSample(options.size, index, "durable", stateDir);
|
||||
} else if (options.scenario === "admission") {
|
||||
sample = await runAdmissionSample(options.size, index);
|
||||
} else if (options.scenario === "recoverySweep") {
|
||||
sample = await runSweepSample(options.size);
|
||||
} else {
|
||||
sample = await runDedupeSample(options.size);
|
||||
}
|
||||
invariant = sample.invariant;
|
||||
if (index >= options.warmup) {
|
||||
timingsMs.push(sample.durationMs);
|
||||
}
|
||||
}
|
||||
return {
|
||||
scenario: options.scenario,
|
||||
size: options.size,
|
||||
timingsMs,
|
||||
memory: {
|
||||
rssStartBytes,
|
||||
rssEndBytes: process.memoryUsage().rss,
|
||||
processMaxRssBytes: processMaxRssBytes(),
|
||||
},
|
||||
invariant,
|
||||
};
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const options = parseOptions(process.argv.slice(2));
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-agent-concurrency-worker-"));
|
||||
const previousStateDir = process.env.OPENCLAW_STATE_DIR;
|
||||
const previousNodeEnv = process.env.NODE_ENV;
|
||||
let result: WorkerResult | undefined;
|
||||
let failure: unknown;
|
||||
process.env.OPENCLAW_STATE_DIR = stateDir;
|
||||
process.env.NODE_ENV = "test";
|
||||
try {
|
||||
const { pinRuntimePaths } = await import("../src/config/paths.js");
|
||||
pinRuntimePaths();
|
||||
result = await runScenario(options, stateDir);
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
} finally {
|
||||
try {
|
||||
await resetRuntime(false);
|
||||
} catch (error) {
|
||||
failure ??= error;
|
||||
} finally {
|
||||
if (previousStateDir === undefined) {
|
||||
delete process.env.OPENCLAW_STATE_DIR;
|
||||
} else {
|
||||
process.env.OPENCLAW_STATE_DIR = previousStateDir;
|
||||
}
|
||||
if (previousNodeEnv === undefined) {
|
||||
delete process.env.NODE_ENV;
|
||||
} else {
|
||||
process.env.NODE_ENV = previousNodeEnv;
|
||||
}
|
||||
try {
|
||||
fs.rmSync(stateDir, { recursive: true, force: true });
|
||||
} catch (error) {
|
||||
failure ??= error;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (failure) {
|
||||
throw toError(failure);
|
||||
}
|
||||
if (!result) {
|
||||
throw new Error("benchmark worker completed without a result");
|
||||
}
|
||||
process.stdout.write(`${WORKER_RESULT_SENTINEL}${JSON.stringify(result)}\n`);
|
||||
}
|
||||
|
||||
if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) {
|
||||
try {
|
||||
await main();
|
||||
} catch (error) {
|
||||
console.error(error instanceof Error ? error.stack : String(error));
|
||||
process.exitCode = 1;
|
||||
} finally {
|
||||
if (process.exitCode && process.exitCode !== 0) {
|
||||
console.error(`[bench-agent-concurrency-worker] FAILED (exit ${process.exitCode})`);
|
||||
}
|
||||
}
|
||||
}
|
||||
+347
-379
@@ -1,12 +1,32 @@
|
||||
import { spawnSync } from "node:child_process";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import type { SubagentRunRecord } from "../src/agents/subagent-registry.types.js";
|
||||
import { isRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
|
||||
const DEFAULT_FANOUT = [1, 8, 32, 64],
|
||||
DEFAULT_SWEEP_ROWS = [32, 128, 512];
|
||||
const DEFAULT_FANOUT = [1, 8, 32, 64];
|
||||
const DEFAULT_SWEEP_ROWS = [32, 128, 512];
|
||||
const WORKER_TIMEOUT_MS = 300_000;
|
||||
export const WORKER_RESULT_SENTINEL = "[bench-agent-concurrency-result] ";
|
||||
|
||||
export type WorkerScenario =
|
||||
| "spawnPipelineInMemory"
|
||||
| "spawnPipelineDurable"
|
||||
| "admission"
|
||||
| "recoverySweep"
|
||||
| "duplicateSuppression";
|
||||
|
||||
export type WorkerResult = {
|
||||
scenario: WorkerScenario;
|
||||
size: number;
|
||||
timingsMs: number[];
|
||||
memory: {
|
||||
rssStartBytes: number;
|
||||
rssEndBytes: number;
|
||||
processMaxRssBytes: number;
|
||||
};
|
||||
invariant: Record<string, number | boolean>;
|
||||
};
|
||||
|
||||
type Options = {
|
||||
runs: number;
|
||||
@@ -18,12 +38,90 @@ type Options = {
|
||||
help: boolean;
|
||||
};
|
||||
|
||||
type TimingSummary = Record<"count" | "min" | "p50" | "p95" | "p99" | "max", number>;
|
||||
type TimingSummary = {
|
||||
count: number;
|
||||
min: number;
|
||||
p50: number;
|
||||
max: number;
|
||||
p95?: number;
|
||||
p99?: number;
|
||||
};
|
||||
|
||||
type ScenarioResult = {
|
||||
size: number;
|
||||
timingsMs: TimingSummary;
|
||||
invariant: Record<string, number | boolean>;
|
||||
const SCENARIO_SPECS: ReadonlyArray<{
|
||||
scenario: WorkerScenario;
|
||||
sizes: "fanout" | "sweepRows";
|
||||
}> = [
|
||||
{ scenario: "spawnPipelineInMemory", sizes: "fanout" },
|
||||
{ scenario: "spawnPipelineDurable", sizes: "fanout" },
|
||||
{ scenario: "admission", sizes: "fanout" },
|
||||
{ scenario: "recoverySweep", sizes: "sweepRows" },
|
||||
{ scenario: "duplicateSuppression", sizes: "sweepRows" },
|
||||
];
|
||||
|
||||
const REQUIRED_INVARIANT_FIELDS: Record<WorkerScenario, readonly string[]> = {
|
||||
spawnPipelineInMemory: [
|
||||
"ok",
|
||||
"registeredRuns",
|
||||
"reservationsReleased",
|
||||
"blockedWaits",
|
||||
"settledRuns",
|
||||
"settledTasks",
|
||||
"outstandingWaits",
|
||||
"durableSubagentRows",
|
||||
"durableTaskRows",
|
||||
"durableStateFile",
|
||||
"postTeardownRegistryRows",
|
||||
"postTeardownTaskRows",
|
||||
"postTeardownDurableSubagentRows",
|
||||
"postTeardownDurableTaskRows",
|
||||
],
|
||||
spawnPipelineDurable: [
|
||||
"ok",
|
||||
"registeredRuns",
|
||||
"reservationsReleased",
|
||||
"blockedWaits",
|
||||
"settledRuns",
|
||||
"settledTasks",
|
||||
"outstandingWaits",
|
||||
"durableSubagentRows",
|
||||
"durableTaskRows",
|
||||
"durableStateFile",
|
||||
"postTeardownRegistryRows",
|
||||
"postTeardownTaskRows",
|
||||
"postTeardownDurableSubagentRows",
|
||||
"postTeardownDurableTaskRows",
|
||||
],
|
||||
admission: ["ok", "admissionCap", "overflowRejected", "released"],
|
||||
recoverySweep: [
|
||||
"ok",
|
||||
"seededRows",
|
||||
"removedRows",
|
||||
"retainedCurrent",
|
||||
"sessionEffects",
|
||||
"recoveryProjections",
|
||||
"lostContextCompletions",
|
||||
],
|
||||
duplicateSuppression: [
|
||||
"ok",
|
||||
"inputRowsPerOrdering",
|
||||
"newestFirstSelectedRows",
|
||||
"oldestFirstSelectedRows",
|
||||
"newestFirstSelectedNewest",
|
||||
"oldestFirstSelectedNewest",
|
||||
],
|
||||
};
|
||||
|
||||
type WorkerProcessResult = {
|
||||
status: number | null;
|
||||
stdout: string;
|
||||
stderr: string;
|
||||
error?: Error & { code?: string };
|
||||
};
|
||||
|
||||
type BenchmarkRuntime = {
|
||||
runWorker?: typeof runWorker;
|
||||
writeProgress?: (line: string) => void;
|
||||
now?: () => number;
|
||||
};
|
||||
|
||||
function usage(): string {
|
||||
@@ -124,390 +222,252 @@ function summarizeTimings(values: number[]): TimingSummary {
|
||||
throw new Error("cannot summarize an empty timing set");
|
||||
}
|
||||
const sorted = values.toSorted((left, right) => left - right);
|
||||
return {
|
||||
const summary: TimingSummary = {
|
||||
count: sorted.length,
|
||||
min: sorted[0] ?? 0,
|
||||
p50: percentile(sorted, 0.5),
|
||||
p95: percentile(sorted, 0.95),
|
||||
p99: percentile(sorted, 0.99),
|
||||
max: sorted.at(-1) ?? 0,
|
||||
};
|
||||
if (sorted.length >= 20) {
|
||||
summary.p95 = percentile(sorted, 0.95);
|
||||
summary.p99 = percentile(sorted, 0.99);
|
||||
}
|
||||
return summary;
|
||||
}
|
||||
|
||||
async function sampleScenario<T extends { durationMs: number }>(
|
||||
function expectedWorkerKeys(options: Options): string[] {
|
||||
return SCENARIO_SPECS.flatMap(({ scenario, sizes }) =>
|
||||
options[sizes].map((size) => `${scenario}:${size}`),
|
||||
);
|
||||
}
|
||||
|
||||
function aggregateWorkerResults(
|
||||
options: Options,
|
||||
sample: () => Promise<T>,
|
||||
validate: (result: T) => Record<string, number | boolean>,
|
||||
): Promise<{ timingsMs: TimingSummary; invariant: Record<string, number | boolean> }> {
|
||||
const timings: number[] = [];
|
||||
let invariant: Record<string, number | boolean> = {};
|
||||
for (let index = 0; index < options.warmup + options.runs; index += 1) {
|
||||
const result = await sample();
|
||||
invariant = validate(result);
|
||||
if (index >= options.warmup) {
|
||||
timings.push(result.durationMs);
|
||||
}
|
||||
workers: WorkerResult[],
|
||||
parentMemory = {
|
||||
rssStartBytes: process.memoryUsage().rss,
|
||||
rssEndBytes: process.memoryUsage().rss,
|
||||
},
|
||||
) {
|
||||
const expected = expectedWorkerKeys(options);
|
||||
const byKey = new Map(workers.map((worker) => [`${worker.scenario}:${worker.size}`, worker]));
|
||||
if (byKey.size !== workers.length) {
|
||||
throw new Error("worker results contain duplicate scenario/size pairs");
|
||||
}
|
||||
const missing = expected.filter((key) => !byKey.has(key));
|
||||
const unexpected = [...byKey.keys()].filter((key) => !expected.includes(key));
|
||||
if (missing.length > 0 || unexpected.length > 0) {
|
||||
throw new Error(
|
||||
`worker result mismatch: missing=${missing.join(",") || "none"} unexpected=${unexpected.join(",") || "none"}`,
|
||||
);
|
||||
}
|
||||
return { timingsMs: summarizeTimings(timings), invariant };
|
||||
}
|
||||
|
||||
async function runSpawnPipelineSample(fanout: number, serial: number) {
|
||||
const [{ runSpawnPipeline }, registry, helpers] = await Promise.all([
|
||||
import("../src/agents/spawn-pipeline.js"),
|
||||
import("../src/agents/subagent-registry-memory.js"),
|
||||
import("../src/agents/subagent-registry.test-helpers.js"),
|
||||
]);
|
||||
helpers.resetSubagentRegistryForTests({ persist: false });
|
||||
helpers.testing.setDepsForTest({
|
||||
getRuntimeConfig: () => ({}),
|
||||
onAgentEvent: () => () => {},
|
||||
persistSubagentRunsToDisk: () => {},
|
||||
persistSubagentRunsToDiskOrThrow: () => {},
|
||||
});
|
||||
let releases = 0;
|
||||
const pipelineParams = Array.from({ length: fanout }, (_, index) => {
|
||||
const runId = `bench-pipeline-${serial}-${index}`;
|
||||
let released = false;
|
||||
return {
|
||||
adapter: {
|
||||
initialize: async () => ({ index }),
|
||||
dispatchTurn: async () => ({ runId }),
|
||||
cleanupOnFailure: async () => {},
|
||||
},
|
||||
admissionReservation: {
|
||||
release: () => {
|
||||
if (!released) {
|
||||
released = true;
|
||||
releases += 1;
|
||||
}
|
||||
},
|
||||
},
|
||||
buildRegistration: () => ({
|
||||
runId,
|
||||
childSessionKey: `agent:bench:subagent:${serial}:${index}`,
|
||||
requesterSessionKey: "agent:bench:main",
|
||||
requesterDisplayKey: "bench",
|
||||
task: `benchmark child ${index}`,
|
||||
cleanup: "keep" as const,
|
||||
expectsCompletionMessage: false,
|
||||
}),
|
||||
progressSessionKey: "agent:bench:main",
|
||||
};
|
||||
});
|
||||
const startedAt = performance.now();
|
||||
const results = await Promise.all(pipelineParams.map((params) => runSpawnPipeline(params)));
|
||||
const durationMs = performance.now() - startedAt;
|
||||
const registered = results.filter((result) => result.ok).map((result) => result.runId);
|
||||
const uniqueRegistered = new Set(registered).size;
|
||||
const registrySize = registry.subagentRuns.size;
|
||||
helpers.resetSubagentRegistryForTests({ persist: false });
|
||||
helpers.testing.setDepsForTest();
|
||||
return { durationMs, expected: fanout, uniqueRegistered, registrySize, releases };
|
||||
}
|
||||
|
||||
function validateSpawnPipeline(result: Awaited<ReturnType<typeof runSpawnPipelineSample>>) {
|
||||
const ok =
|
||||
result.uniqueRegistered === result.expected &&
|
||||
result.registrySize === result.expected &&
|
||||
result.releases === result.expected;
|
||||
if (!ok) {
|
||||
throw new Error(`spawn pipeline invariant failed: ${JSON.stringify(result)}`);
|
||||
}
|
||||
return { ok, registeredRuns: result.registrySize, reservationsReleased: result.releases };
|
||||
}
|
||||
|
||||
async function runAdmissionSample(fanout: number, serial: number) {
|
||||
const { reserveChildAdmissionSlot } = await import("../src/agents/child-admission.js");
|
||||
const controllerSessionKey = `agent:bench:admission:${serial}`;
|
||||
const reservations: Array<{ release: () => void }> = [];
|
||||
const reserve = () =>
|
||||
reserveChildAdmissionSlot({
|
||||
controllerSessionKey,
|
||||
resolveAdmission: (pendingChildren) =>
|
||||
pendingChildren < fanout
|
||||
? { ok: true as const }
|
||||
: { ok: false as const, governingCap: "benchmark" },
|
||||
});
|
||||
const startedAt = performance.now();
|
||||
for (let index = 0; index < fanout; index += 1) {
|
||||
const reservation = reserve();
|
||||
if (!reservation.ok) {
|
||||
throw new Error(`admission rejected slot ${index + 1}/${fanout}`);
|
||||
}
|
||||
reservations.push(reservation);
|
||||
}
|
||||
const overflow = reserve();
|
||||
for (const reservation of reservations) {
|
||||
reservation.release();
|
||||
reservation.release();
|
||||
}
|
||||
const replacement = reserve();
|
||||
if (replacement.ok) {
|
||||
replacement.release();
|
||||
}
|
||||
return {
|
||||
durationMs: performance.now() - startedAt,
|
||||
admitted: reservations.length,
|
||||
overflowRejected: !overflow.ok,
|
||||
replacement: replacement.ok,
|
||||
};
|
||||
}
|
||||
|
||||
function validateAdmission(result: Awaited<ReturnType<typeof runAdmissionSample>>) {
|
||||
const ok = result.overflowRejected && result.replacement;
|
||||
if (!ok) {
|
||||
throw new Error(`admission invariant failed: ${JSON.stringify(result)}`);
|
||||
}
|
||||
return { ok, admissionCap: result.admitted, overflowRejected: true, released: true };
|
||||
}
|
||||
|
||||
function recoveryRow(child: number, generation: number, now: number): SubagentRunRecord {
|
||||
const current = generation === 3;
|
||||
return {
|
||||
runId: `bench-sweep-${child}-${generation}`,
|
||||
childSessionKey: `agent:bench:subagent:sweep-${child}`,
|
||||
requesterSessionKey: "agent:bench:main",
|
||||
requesterDisplayKey: "bench",
|
||||
task: `sweep child ${child}`,
|
||||
cleanup: current ? "keep" : "delete",
|
||||
generation,
|
||||
createdAt: now - generation,
|
||||
archiveAtMs: current ? undefined : now - 1,
|
||||
terminalOwner: current ? "interrupted-recovery" : undefined,
|
||||
endedReason: current ? "subagent-error" : undefined,
|
||||
execution: current
|
||||
? {
|
||||
status: "terminal",
|
||||
startedAt: now - 2_000,
|
||||
endedAt: now - 1_000,
|
||||
outcome: { status: "error", error: "interrupted recovery replay" },
|
||||
suppressSessionEffects: true,
|
||||
const scenarios = Object.fromEntries(
|
||||
SCENARIO_SPECS.map(({ scenario, sizes }) => [
|
||||
scenario,
|
||||
options[sizes].map((size) => {
|
||||
const worker = byKey.get(`${scenario}:${size}`);
|
||||
if (!worker) {
|
||||
throw new Error(`missing worker result for ${scenario}:${size}`);
|
||||
}
|
||||
: {
|
||||
status: "terminal",
|
||||
startedAt: now - 2_000,
|
||||
endedAt: now - 1_000,
|
||||
outcome: { status: "error", error: "retired recovery generation" },
|
||||
suppressSessionEffects: true,
|
||||
},
|
||||
return {
|
||||
size,
|
||||
timingsMs: summarizeTimings(worker.timingsMs),
|
||||
memory: worker.memory,
|
||||
invariant: worker.invariant,
|
||||
};
|
||||
}),
|
||||
]),
|
||||
) as Record<
|
||||
WorkerScenario,
|
||||
Array<{
|
||||
size: number;
|
||||
timingsMs: TimingSummary;
|
||||
memory: WorkerResult["memory"];
|
||||
invariant: WorkerResult["invariant"];
|
||||
}>
|
||||
>;
|
||||
|
||||
const checks = {
|
||||
spawnPipelineInMemory: scenarios.spawnPipelineInMemory.every(
|
||||
(entry) => entry.invariant.ok === true,
|
||||
),
|
||||
spawnPipelineDurable: scenarios.spawnPipelineDurable.every(
|
||||
(entry) => entry.invariant.ok === true,
|
||||
),
|
||||
admissionCapOverflowRelease: scenarios.admission.every((entry) => entry.invariant.ok === true),
|
||||
sweepRecoveryRowsWithoutSessionEffects: scenarios.recoverySweep.every(
|
||||
(entry) => entry.invariant.ok === true,
|
||||
),
|
||||
dedupeNewestPerChild: scenarios.duplicateSuppression.every(
|
||||
(entry) => entry.invariant.ok === true,
|
||||
),
|
||||
};
|
||||
const failures = Object.entries(checks)
|
||||
.filter(([, ok]) => !ok)
|
||||
.map(([name]) => name);
|
||||
|
||||
return {
|
||||
schemaVersion: 2,
|
||||
generatedAt: new Date().toISOString(),
|
||||
runtime: { node: process.version, platform: process.platform, arch: process.arch },
|
||||
options: {
|
||||
runs: options.runs,
|
||||
warmup: options.warmup,
|
||||
fanout: options.fanout,
|
||||
sweepRows: options.sweepRows,
|
||||
},
|
||||
memory: {
|
||||
...parentMemory,
|
||||
workerProcessMaxRssBytes: Math.max(
|
||||
...workers.map((worker) => worker.memory.processMaxRssBytes),
|
||||
),
|
||||
},
|
||||
scenarios,
|
||||
invariants: { ok: failures.length === 0, failures, ...checks },
|
||||
};
|
||||
}
|
||||
|
||||
async function runSweepSample(childCount: number) {
|
||||
const { createSubagentRegistrySweeper } =
|
||||
await import("../src/agents/subagent-registry-sweeper.js");
|
||||
const now = Date.now();
|
||||
const runs = new Map<string, SubagentRunRecord>();
|
||||
for (let child = 0; child < childCount; child += 1) {
|
||||
for (const generation of [3, 2, 1]) {
|
||||
const entry = recoveryRow(child, generation, now);
|
||||
runs.set(entry.runId, entry);
|
||||
function assertFiniteNonNegative(value: unknown, field: string): asserts value is number {
|
||||
if (typeof value !== "number" || !Number.isFinite(value) || value < 0) {
|
||||
throw new Error(`worker result field ${field} must be a finite nonnegative number`);
|
||||
}
|
||||
}
|
||||
|
||||
function validateWorkerResult(
|
||||
value: unknown,
|
||||
expected: { scenario: WorkerScenario; size: number; runs: number },
|
||||
): WorkerResult {
|
||||
if (!isRecord(value)) {
|
||||
throw new Error("worker result must be an object");
|
||||
}
|
||||
if (value.scenario !== expected.scenario || value.size !== expected.size) {
|
||||
throw new Error(`worker ${expected.scenario}:${expected.size} returned mismatched identity`);
|
||||
}
|
||||
if (!Array.isArray(value.timingsMs) || value.timingsMs.length !== expected.runs) {
|
||||
throw new Error(
|
||||
`worker ${expected.scenario}:${expected.size} returned ${Array.isArray(value.timingsMs) ? value.timingsMs.length : "invalid"} samples; expected ${expected.runs}`,
|
||||
);
|
||||
}
|
||||
value.timingsMs.forEach((timing, index) =>
|
||||
assertFiniteNonNegative(timing, `timingsMs[${index}]`),
|
||||
);
|
||||
if (!isRecord(value.memory)) {
|
||||
throw new Error("worker result memory must be an object");
|
||||
}
|
||||
assertFiniteNonNegative(value.memory.rssStartBytes, "memory.rssStartBytes");
|
||||
assertFiniteNonNegative(value.memory.rssEndBytes, "memory.rssEndBytes");
|
||||
assertFiniteNonNegative(value.memory.processMaxRssBytes, "memory.processMaxRssBytes");
|
||||
if (!isRecord(value.invariant)) {
|
||||
throw new Error("worker result invariant must be an object");
|
||||
}
|
||||
for (const field of REQUIRED_INVARIANT_FIELDS[expected.scenario]) {
|
||||
const invariantValue = value.invariant[field];
|
||||
if (typeof invariantValue !== "number" && typeof invariantValue !== "boolean") {
|
||||
throw new Error(`worker result invariant.${field} is missing or invalid`);
|
||||
}
|
||||
}
|
||||
let sessionEffects = 0;
|
||||
let recoveryProjections = 0;
|
||||
let lostContextCompletions = 0;
|
||||
const sweeper = createSubagentRegistrySweeper({
|
||||
runs,
|
||||
resumedRuns: new Set(),
|
||||
persist: () => {},
|
||||
clearPendingLifecycleError: () => {},
|
||||
clearPendingLifecycleTimeout: () => {},
|
||||
sweepPendingLifecycle: () => {},
|
||||
completeSubagentRunWithRecovery: async () => {
|
||||
lostContextCompletions += 1;
|
||||
if (value.invariant.ok !== true) {
|
||||
throw new Error(`worker ${expected.scenario}:${expected.size} reported a failed invariant`);
|
||||
}
|
||||
return value as WorkerResult;
|
||||
}
|
||||
|
||||
function parseWorkerProcessResult(
|
||||
result: WorkerProcessResult,
|
||||
expected: { scenario: WorkerScenario; size: number; runs: number },
|
||||
): WorkerResult {
|
||||
if (result.error) {
|
||||
const detail =
|
||||
"code" in result.error && result.error.code === "ETIMEDOUT"
|
||||
? `timed out after ${WORKER_TIMEOUT_MS}ms`
|
||||
: result.error.message;
|
||||
throw new Error(`worker ${expected.scenario}:${expected.size} failed: ${detail}`);
|
||||
}
|
||||
if (result.status !== 0) {
|
||||
throw new Error(
|
||||
`worker ${expected.scenario}:${expected.size} failed (${result.status ?? "signal"}): ${result.stderr.trim() || result.stdout.trim()}`,
|
||||
);
|
||||
}
|
||||
const payloads = result.stdout
|
||||
.split(/\r?\n/u)
|
||||
.filter((line) => line.startsWith(WORKER_RESULT_SENTINEL));
|
||||
if (payloads.length !== 1) {
|
||||
throw new Error(
|
||||
`worker ${expected.scenario}:${expected.size} returned ${payloads.length} result payloads`,
|
||||
);
|
||||
}
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(payloads[0]!.slice(WORKER_RESULT_SENTINEL.length));
|
||||
} catch {
|
||||
throw new Error(`worker ${expected.scenario}:${expected.size} returned invalid JSON`);
|
||||
}
|
||||
return validateWorkerResult(parsed, expected);
|
||||
}
|
||||
|
||||
function runWorker(options: Options, scenario: WorkerScenario, size: number): WorkerResult {
|
||||
const result = spawnSync(
|
||||
process.execPath,
|
||||
[
|
||||
"--import",
|
||||
"tsx",
|
||||
"scripts/bench-agent-concurrency-worker.ts",
|
||||
"--scenario",
|
||||
scenario,
|
||||
"--size",
|
||||
String(size),
|
||||
"--runs",
|
||||
String(options.runs),
|
||||
"--warmup",
|
||||
String(options.warmup),
|
||||
],
|
||||
{
|
||||
cwd: process.cwd(),
|
||||
encoding: "utf8",
|
||||
env: { ...process.env, NODE_NO_WARNINGS: "1" },
|
||||
timeout: WORKER_TIMEOUT_MS,
|
||||
killSignal: "SIGTERM",
|
||||
maxBuffer: 16 * 1024 * 1024,
|
||||
},
|
||||
getGatewayRecoveryRuntime: () => undefined,
|
||||
abandonSubagentRestartRecoveryLaunch: () => true,
|
||||
clearAcceptedSubagentRestartRecovery: () => true,
|
||||
resumeSettledSubagentRestartRecovery: () => true,
|
||||
replaceSubagentRunAfterSteer: () => true,
|
||||
markSubagentRestartRecoveryLaunchAttempted: () => undefined,
|
||||
markSubagentRestartRecoveryLaunchAccepted: () => undefined,
|
||||
markSubagentRestartRecoveryLaunchConsumed: () => undefined,
|
||||
reserveSubagentRestartRecoveryLaunch: () => undefined,
|
||||
resetSubagentRestartRecoveryLaunchAttempt: () => true,
|
||||
finalizeInterruptedSubagentRun: async ({ runId, expectedEntry }) => {
|
||||
if (runs.get(runId) !== expectedEntry || expectedEntry?.generation !== 3) {
|
||||
throw new Error(`unexpected recovery projection owner: ${runId}`);
|
||||
}
|
||||
recoveryProjections += 1;
|
||||
return 1;
|
||||
},
|
||||
resumeRequesterSettleWake: () => {},
|
||||
startSubagentAnnounceCleanupFlow: () => true,
|
||||
completeCleanupBookkeeping: () => {},
|
||||
shouldEmitEndedHookForRun: () => false,
|
||||
emitSubagentEndedHookForRun: async () => {},
|
||||
callGateway: (async <T>() => {
|
||||
sessionEffects += 1;
|
||||
return {} as T;
|
||||
}) as typeof import("../src/gateway/call.js").callGateway,
|
||||
cleanupCollectorLaunchResources: async () => true,
|
||||
runContextEngineSubagentEnded: async () => {
|
||||
sessionEffects += 1;
|
||||
},
|
||||
notifyContextEngineSubagentEnded: async () => {
|
||||
sessionEffects += 1;
|
||||
},
|
||||
retireSupersededRun: async () => {},
|
||||
getRunsForChildSession: (childSessionKey) =>
|
||||
[...runs.values()].filter((entry) => entry.childSessionKey === childSessionKey),
|
||||
getRunsForCollectorGroup: () => [],
|
||||
warn: () => {},
|
||||
);
|
||||
return parseWorkerProcessResult(result, { scenario, size, runs: options.runs });
|
||||
}
|
||||
|
||||
function benchmark(options: Options, runtime: BenchmarkRuntime = {}) {
|
||||
const rssStartBytes = process.memoryUsage().rss;
|
||||
const jobs = SCENARIO_SPECS.flatMap(({ scenario, sizes }) =>
|
||||
options[sizes].map((size) => ({ scenario, size })),
|
||||
);
|
||||
const run = runtime.runWorker ?? runWorker;
|
||||
const writeProgress =
|
||||
runtime.writeProgress ?? ((line: string) => process.stderr.write(`${line}\n`));
|
||||
const now = runtime.now ?? Date.now;
|
||||
const workers = jobs.map(({ scenario, size }, index) => {
|
||||
const ordinal = index + 1;
|
||||
writeProgress(
|
||||
`[bench-agent-concurrency] worker ${ordinal}/${jobs.length} start scenario=${scenario} size=${size}`,
|
||||
);
|
||||
const startedAt = now();
|
||||
try {
|
||||
const worker = run(options, scenario, size);
|
||||
const elapsedMs = Math.max(0, now() - startedAt);
|
||||
writeProgress(
|
||||
`[bench-agent-concurrency] worker ${ordinal}/${jobs.length} complete scenario=${scenario} size=${size} elapsed=${(elapsedMs / 1_000).toFixed(3)}s`,
|
||||
);
|
||||
return worker;
|
||||
} catch (error) {
|
||||
const elapsedMs = Math.max(0, now() - startedAt);
|
||||
writeProgress(
|
||||
`[bench-agent-concurrency] worker ${ordinal}/${jobs.length} failed scenario=${scenario} size=${size} elapsed=${(elapsedMs / 1_000).toFixed(3)}s`,
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
return aggregateWorkerResults(options, workers, {
|
||||
rssStartBytes,
|
||||
rssEndBytes: process.memoryUsage().rss,
|
||||
});
|
||||
const startedAt = performance.now();
|
||||
let durationMs: number;
|
||||
try {
|
||||
await sweeper.sweepOnce();
|
||||
durationMs = performance.now() - startedAt;
|
||||
} finally {
|
||||
sweeper.reset();
|
||||
}
|
||||
const retainedCurrent = [...runs.values()].filter((entry) => entry.generation === 3).length;
|
||||
return {
|
||||
durationMs,
|
||||
seededRows: childCount * 3,
|
||||
removedRows: childCount * 3 - runs.size,
|
||||
retainedCurrent,
|
||||
sessionEffects,
|
||||
recoveryProjections,
|
||||
lostContextCompletions,
|
||||
};
|
||||
}
|
||||
|
||||
function validateSweep(result: Awaited<ReturnType<typeof runSweepSample>>) {
|
||||
const expectedChildren = result.seededRows / 3;
|
||||
const ok =
|
||||
result.removedRows === expectedChildren * 2 &&
|
||||
result.retainedCurrent === expectedChildren &&
|
||||
result.sessionEffects === 0 &&
|
||||
result.recoveryProjections === expectedChildren &&
|
||||
result.lostContextCompletions === 0;
|
||||
if (!ok) {
|
||||
throw new Error(`registry sweep invariant failed: ${JSON.stringify(result)}`);
|
||||
}
|
||||
return { ok, ...result };
|
||||
}
|
||||
|
||||
async function runDedupeSample(childCount: number) {
|
||||
const { dedupeLatestChildCompletionRows } =
|
||||
await import("../src/agents/subagent-announce-output.js");
|
||||
const rows = Array.from({ length: childCount }, (_, child) =>
|
||||
[3, 2, 1].map((generation) => ({
|
||||
runId: `bench-dedupe-${child}-${generation}`,
|
||||
childSessionKey: `agent:bench:subagent:dedupe-${child}`,
|
||||
task: `dedupe child ${child}`,
|
||||
generation,
|
||||
createdAt: generation,
|
||||
execution: { status: "terminal" as const, endedAt: generation },
|
||||
})),
|
||||
).flat();
|
||||
const startedAt = performance.now();
|
||||
const deduped = dedupeLatestChildCompletionRows(rows);
|
||||
return {
|
||||
durationMs: performance.now() - startedAt,
|
||||
inputRows: rows.length,
|
||||
selectedRows: deduped.length,
|
||||
newestSelected: deduped.every((row) => row.generation === 3),
|
||||
};
|
||||
}
|
||||
|
||||
function validateDedupe(result: Awaited<ReturnType<typeof runDedupeSample>>) {
|
||||
const ok = result.selectedRows === result.inputRows / 3 && result.newestSelected;
|
||||
if (!ok) {
|
||||
throw new Error(`completion dedupe invariant failed: ${JSON.stringify(result)}`);
|
||||
}
|
||||
return { ok, ...result };
|
||||
}
|
||||
|
||||
async function benchmark(options: Options) {
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-agent-concurrency-"));
|
||||
const previousStateDir = process.env.OPENCLAW_STATE_DIR;
|
||||
const previousNodeEnv = process.env.NODE_ENV;
|
||||
process.env.OPENCLAW_STATE_DIR = stateDir;
|
||||
process.env.NODE_ENV = "test";
|
||||
let serial = 0;
|
||||
let peakRss = process.memoryUsage().rss;
|
||||
const beforeRss = peakRss;
|
||||
const runSized = async <T extends { durationMs: number }>(
|
||||
sizes: number[],
|
||||
sample: (size: number, serial: number) => Promise<T>,
|
||||
validate: (result: T) => Record<string, number | boolean>,
|
||||
): Promise<ScenarioResult[]> => {
|
||||
const results: ScenarioResult[] = [];
|
||||
for (const size of sizes) {
|
||||
const measured = await sampleScenario(options, () => sample(size, serial++), validate);
|
||||
peakRss = Math.max(peakRss, process.memoryUsage().rss);
|
||||
results.push({ size, ...measured });
|
||||
}
|
||||
return results;
|
||||
};
|
||||
try {
|
||||
const scenarios = {
|
||||
spawnPipeline: await runSized(options.fanout, runSpawnPipelineSample, validateSpawnPipeline),
|
||||
admission: await runSized(options.fanout, runAdmissionSample, validateAdmission),
|
||||
recoverySweep: await runSized(
|
||||
options.sweepRows,
|
||||
(size) => runSweepSample(size),
|
||||
validateSweep,
|
||||
),
|
||||
duplicateSuppression: await runSized(
|
||||
options.sweepRows,
|
||||
(size) => runDedupeSample(size),
|
||||
validateDedupe,
|
||||
),
|
||||
};
|
||||
const afterRss = process.memoryUsage().rss;
|
||||
const checks = {
|
||||
admissionCapOverflowRelease: scenarios.admission.every((entry) => entry.invariant.ok),
|
||||
uniqueRegisteredRunsAndReleasedReservations: scenarios.spawnPipeline.every(
|
||||
(entry) => entry.invariant.ok,
|
||||
),
|
||||
sweepRecoveryRowsWithoutSessionEffects: scenarios.recoverySweep.every(
|
||||
(entry) => entry.invariant.ok,
|
||||
),
|
||||
dedupeNewestPerChild: scenarios.duplicateSuppression.every((entry) => entry.invariant.ok),
|
||||
};
|
||||
const failures = Object.entries(checks)
|
||||
.filter(([, ok]) => !ok)
|
||||
.map(([name]) => name);
|
||||
return {
|
||||
schemaVersion: 1,
|
||||
generatedAt: new Date().toISOString(),
|
||||
runtime: { node: process.version, platform: process.platform, arch: process.arch },
|
||||
options: {
|
||||
runs: options.runs,
|
||||
warmup: options.warmup,
|
||||
fanout: options.fanout,
|
||||
sweepRows: options.sweepRows,
|
||||
},
|
||||
memory: {
|
||||
rssStartBytes: beforeRss,
|
||||
rssPeakBytes: Math.max(peakRss, afterRss),
|
||||
rssEndBytes: afterRss,
|
||||
rssDeltaBytes: afterRss - beforeRss,
|
||||
},
|
||||
scenarios,
|
||||
invariants: {
|
||||
ok: failures.length === 0,
|
||||
failures,
|
||||
...checks,
|
||||
},
|
||||
};
|
||||
} finally {
|
||||
if (previousStateDir === undefined) {
|
||||
delete process.env.OPENCLAW_STATE_DIR;
|
||||
} else {
|
||||
process.env.OPENCLAW_STATE_DIR = previousStateDir;
|
||||
}
|
||||
if (previousNodeEnv === undefined) {
|
||||
delete process.env.NODE_ENV;
|
||||
} else {
|
||||
process.env.NODE_ENV = previousNodeEnv;
|
||||
}
|
||||
fs.rmSync(stateDir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
async function main(argv = process.argv.slice(2)): Promise<void> {
|
||||
@@ -516,7 +476,7 @@ async function main(argv = process.argv.slice(2)): Promise<void> {
|
||||
process.stdout.write(usage());
|
||||
return;
|
||||
}
|
||||
const report = await benchmark(options);
|
||||
const report = benchmark(options);
|
||||
const json = `${JSON.stringify(report, null, 2)}\n`;
|
||||
if (options.output) {
|
||||
fs.mkdirSync(path.dirname(path.resolve(options.output)), { recursive: true });
|
||||
@@ -528,17 +488,25 @@ async function main(argv = process.argv.slice(2)): Promise<void> {
|
||||
}
|
||||
for (const [name, scenarios] of Object.entries(report.scenarios)) {
|
||||
for (const scenario of scenarios) {
|
||||
const tail =
|
||||
scenario.timingsMs.p95 === undefined
|
||||
? ""
|
||||
: ` p95=${scenario.timingsMs.p95.toFixed(3)}ms p99=${scenario.timingsMs.p99?.toFixed(3)}ms`;
|
||||
console.log(
|
||||
`${name} size=${scenario.size} p50=${scenario.timingsMs.p50.toFixed(3)}ms p95=${scenario.timingsMs.p95.toFixed(3)}ms`,
|
||||
`${name} size=${scenario.size} p50=${scenario.timingsMs.p50.toFixed(3)}ms max=${scenario.timingsMs.max.toFixed(3)}ms${tail}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
console.log(`peak RSS ${(report.memory.rssPeakBytes / 1024 / 1024).toFixed(1)} MiB`);
|
||||
console.log(
|
||||
`max worker RSS ${(report.memory.workerProcessMaxRssBytes / 1024 / 1024).toFixed(1)} MiB`,
|
||||
);
|
||||
}
|
||||
|
||||
export const testing = {
|
||||
aggregateWorkerResults,
|
||||
benchmark,
|
||||
parseOptions,
|
||||
parseWorkerProcessResult,
|
||||
summarizeTimings,
|
||||
};
|
||||
|
||||
|
||||
@@ -1,6 +1,67 @@
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { testing } from "../../scripts/bench-agent-concurrency.ts";
|
||||
import {
|
||||
testing,
|
||||
WORKER_RESULT_SENTINEL,
|
||||
type WorkerResult,
|
||||
type WorkerScenario,
|
||||
} from "../../scripts/bench-agent-concurrency.ts";
|
||||
|
||||
function workerResult(scenario: WorkerScenario, size: number, timingsMs = [1, 2, 3]): WorkerResult {
|
||||
const invariant: Record<string, number | boolean> =
|
||||
scenario === "spawnPipelineInMemory" || scenario === "spawnPipelineDurable"
|
||||
? {
|
||||
ok: true,
|
||||
registeredRuns: size,
|
||||
reservationsReleased: size,
|
||||
blockedWaits: size,
|
||||
settledRuns: size,
|
||||
settledTasks: size,
|
||||
outstandingWaits: 0,
|
||||
durableSubagentRows: scenario === "spawnPipelineDurable" ? size : 0,
|
||||
durableTaskRows: scenario === "spawnPipelineDurable" ? size : 0,
|
||||
durableStateFile: scenario === "spawnPipelineDurable",
|
||||
postTeardownRegistryRows: 0,
|
||||
postTeardownTaskRows: 0,
|
||||
postTeardownDurableSubagentRows: 0,
|
||||
postTeardownDurableTaskRows: 0,
|
||||
}
|
||||
: scenario === "admission"
|
||||
? { ok: true, admissionCap: size, overflowRejected: true, released: true }
|
||||
: scenario === "recoverySweep"
|
||||
? {
|
||||
ok: true,
|
||||
seededRows: size * 3,
|
||||
removedRows: size * 2,
|
||||
retainedCurrent: size,
|
||||
sessionEffects: 0,
|
||||
recoveryProjections: size,
|
||||
lostContextCompletions: 0,
|
||||
}
|
||||
: {
|
||||
ok: true,
|
||||
inputRowsPerOrdering: size * 3,
|
||||
newestFirstSelectedRows: size,
|
||||
oldestFirstSelectedRows: size,
|
||||
newestFirstSelectedNewest: true,
|
||||
oldestFirstSelectedNewest: true,
|
||||
};
|
||||
return {
|
||||
scenario,
|
||||
size,
|
||||
timingsMs,
|
||||
memory: {
|
||||
rssStartBytes: 100,
|
||||
rssEndBytes: 120,
|
||||
processMaxRssBytes: 150,
|
||||
},
|
||||
invariant,
|
||||
};
|
||||
}
|
||||
|
||||
function workerStdout(result: WorkerResult): string {
|
||||
return `${WORKER_RESULT_SENTINEL}${JSON.stringify(result)}\n`;
|
||||
}
|
||||
|
||||
describe("agent concurrency benchmark", () => {
|
||||
it("parses bounded options and rejects ambiguous arguments", () => {
|
||||
@@ -36,60 +97,189 @@ describe("agent concurrency benchmark", () => {
|
||||
expect(() => testing.parseOptions(["--wat"])).toThrow("Unknown argument: --wat");
|
||||
});
|
||||
|
||||
it("summarizes min and nearest-rank percentiles", () => {
|
||||
it("emits tail percentiles only when the sample supports them", () => {
|
||||
expect(testing.summarizeTimings([100, 1, 4, 2, 3])).toEqual({
|
||||
count: 5,
|
||||
min: 1,
|
||||
p50: 3,
|
||||
p95: 100,
|
||||
p99: 100,
|
||||
max: 100,
|
||||
});
|
||||
const summary = testing.summarizeTimings(Array.from({ length: 20 }, (_, index) => index + 1));
|
||||
expect(summary).toMatchObject({ count: 20, p50: 10, p95: 19, p99: 20, max: 20 });
|
||||
});
|
||||
|
||||
it("emits schema version 1, RSS, scenarios, and invariants in a tiny real smoke", async () => {
|
||||
const previousNodeEnv = process.env.NODE_ENV;
|
||||
const report = await testing.benchmark(
|
||||
testing.parseOptions(["--runs", "1", "--warmup", "0", "--fanout", "2", "--sweep-rows", "2"]),
|
||||
it("aggregates synthetic worker results into schema version 2", () => {
|
||||
const options = testing.parseOptions([
|
||||
"--runs",
|
||||
"3",
|
||||
"--warmup",
|
||||
"1",
|
||||
"--fanout",
|
||||
"2",
|
||||
"--sweep-rows",
|
||||
"4",
|
||||
]);
|
||||
const report = testing.aggregateWorkerResults(
|
||||
options,
|
||||
[
|
||||
workerResult("spawnPipelineInMemory", 2),
|
||||
workerResult("spawnPipelineDurable", 2),
|
||||
workerResult("admission", 2),
|
||||
workerResult("recoverySweep", 4),
|
||||
workerResult("duplicateSuppression", 4),
|
||||
],
|
||||
{ rssStartBytes: 10, rssEndBytes: 20 },
|
||||
);
|
||||
|
||||
expect(report).toMatchObject({
|
||||
schemaVersion: 1,
|
||||
runtime: { node: process.version },
|
||||
options: { runs: 1, warmup: 0, fanout: [2], sweepRows: [2] },
|
||||
schemaVersion: 2,
|
||||
options: { runs: 3, warmup: 1, fanout: [2], sweepRows: [4] },
|
||||
memory: {
|
||||
rssStartBytes: expect.any(Number),
|
||||
rssPeakBytes: expect.any(Number),
|
||||
rssEndBytes: expect.any(Number),
|
||||
rssDeltaBytes: expect.any(Number),
|
||||
rssStartBytes: 10,
|
||||
rssEndBytes: 20,
|
||||
workerProcessMaxRssBytes: 150,
|
||||
},
|
||||
invariants: {
|
||||
ok: true,
|
||||
failures: [],
|
||||
spawnPipelineInMemory: true,
|
||||
spawnPipelineDurable: true,
|
||||
admissionCapOverflowRelease: true,
|
||||
uniqueRegisteredRunsAndReleasedReservations: true,
|
||||
sweepRecoveryRowsWithoutSessionEffects: true,
|
||||
dedupeNewestPerChild: true,
|
||||
},
|
||||
});
|
||||
expect(process.env.NODE_ENV).toBe(previousNodeEnv);
|
||||
expect(report.scenarios.spawnPipelineDurable[0]?.timingsMs).toEqual({
|
||||
count: 3,
|
||||
min: 1,
|
||||
p50: 2,
|
||||
max: 3,
|
||||
});
|
||||
expect(report.generatedAt).toEqual(expect.any(String));
|
||||
expect(report.scenarios.spawnPipeline[0]?.timingsMs.count).toBe(1);
|
||||
expect(report.scenarios.recoverySweep[0]?.invariant).toMatchObject({
|
||||
seededRows: 6,
|
||||
removedRows: 4,
|
||||
retainedCurrent: 2,
|
||||
sessionEffects: 0,
|
||||
recoveryProjections: 2,
|
||||
lostContextCompletions: 0,
|
||||
});
|
||||
expect(report.scenarios.duplicateSuppression[0]?.invariant).toMatchObject({
|
||||
selectedRows: 2,
|
||||
newestSelected: true,
|
||||
});
|
||||
});
|
||||
|
||||
it("supports native Node TypeScript help and ends failures with the marker", () => {
|
||||
it("reports deterministic parent progress around every worker", () => {
|
||||
const options = testing.parseOptions([
|
||||
"--runs",
|
||||
"3",
|
||||
"--warmup",
|
||||
"0",
|
||||
"--fanout",
|
||||
"2",
|
||||
"--sweep-rows",
|
||||
"4",
|
||||
]);
|
||||
const progress: string[] = [];
|
||||
let now = 0;
|
||||
const report = testing.benchmark(options, {
|
||||
runWorker: (_options, scenario, size) => workerResult(scenario, size),
|
||||
writeProgress: (line) => progress.push(line),
|
||||
now: () => {
|
||||
const value = now;
|
||||
now += 250;
|
||||
return value;
|
||||
},
|
||||
});
|
||||
|
||||
expect(report.invariants.ok).toBe(true);
|
||||
expect(progress).toHaveLength(10);
|
||||
expect(progress[0]).toBe(
|
||||
"[bench-agent-concurrency] worker 1/5 start scenario=spawnPipelineInMemory size=2",
|
||||
);
|
||||
expect(progress[1]).toBe(
|
||||
"[bench-agent-concurrency] worker 1/5 complete scenario=spawnPipelineInMemory size=2 elapsed=0.250s",
|
||||
);
|
||||
expect(progress.at(-2)).toBe(
|
||||
"[bench-agent-concurrency] worker 5/5 start scenario=duplicateSuppression size=4",
|
||||
);
|
||||
expect(progress.at(-1)).toBe(
|
||||
"[bench-agent-concurrency] worker 5/5 complete scenario=duplicateSuppression size=4 elapsed=0.250s",
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects incomplete synthetic worker sets", () => {
|
||||
const options = testing.parseOptions(["--fanout", "1", "--sweep-rows", "1"]);
|
||||
expect(() =>
|
||||
testing.aggregateWorkerResults(options, [workerResult("spawnPipelineInMemory", 1)]),
|
||||
).toThrow("worker result mismatch");
|
||||
});
|
||||
|
||||
it("rejects malformed, duplicate, mismatched, partial, and timed-out worker results", () => {
|
||||
const expected = { scenario: "admission" as const, size: 2, runs: 3 };
|
||||
const valid = workerResult("admission", 2);
|
||||
expect(
|
||||
testing.parseWorkerProcessResult(
|
||||
{ status: 0, stdout: workerStdout(valid), stderr: "" },
|
||||
expected,
|
||||
),
|
||||
).toEqual(valid);
|
||||
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{ status: 0, stdout: `${WORKER_RESULT_SENTINEL}{\n`, stderr: "" },
|
||||
expected,
|
||||
),
|
||||
).toThrow("returned invalid JSON");
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{ status: 0, stdout: `${workerStdout(valid)}${workerStdout(valid)}`, stderr: "" },
|
||||
expected,
|
||||
),
|
||||
).toThrow("returned 2 result payloads");
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{ status: 0, stdout: workerStdout(workerResult("admission", 3)), stderr: "" },
|
||||
expected,
|
||||
),
|
||||
).toThrow("mismatched identity");
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{
|
||||
status: 0,
|
||||
stdout: `${WORKER_RESULT_SENTINEL}${JSON.stringify({ ...valid, memory: {} })}\n`,
|
||||
stderr: "",
|
||||
},
|
||||
expected,
|
||||
),
|
||||
).toThrow("memory.rssStartBytes");
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{
|
||||
status: 0,
|
||||
stdout: workerStdout({ ...valid, timingsMs: [1, 2] }),
|
||||
stderr: "",
|
||||
},
|
||||
expected,
|
||||
),
|
||||
).toThrow("returned 2 samples; expected 3");
|
||||
const missingInvariant = structuredClone(valid);
|
||||
delete missingInvariant.invariant.admissionCap;
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{ status: 0, stdout: workerStdout(missingInvariant), stderr: "" },
|
||||
expected,
|
||||
),
|
||||
).toThrow("invariant.admissionCap is missing or invalid");
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{ status: 7, stdout: "", stderr: "worker exploded" },
|
||||
expected,
|
||||
),
|
||||
).toThrow("failed (7): worker exploded");
|
||||
expect(() =>
|
||||
testing.parseWorkerProcessResult(
|
||||
{
|
||||
status: null,
|
||||
stdout: "",
|
||||
stderr: "",
|
||||
error: Object.assign(new Error("spawnSync timed out"), { code: "ETIMEDOUT" }),
|
||||
},
|
||||
expected,
|
||||
),
|
||||
).toThrow("timed out after 300000ms");
|
||||
});
|
||||
|
||||
it("supports help and ends failures with the marker", () => {
|
||||
const help = spawnSync(
|
||||
process.execPath,
|
||||
["--import", "tsx", "scripts/bench-agent-concurrency.ts", "--help"],
|
||||
|
||||
Reference in New Issue
Block a user