centersl/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-03-27 09:21:35 +07:00
Code Issues Packages Projects Releases Wiki Activity

test: add local thread expansion policy

Browse Source
This commit is contained in:
Tak Hoffman
2026-03-25 23:27:28 -05:00
parent 1bc30b7fb9
commit 813059460f
7 changed files with 416 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
.github/workflows/thread-expansion-experiment.yml vendored Normal file
View File

@@ -0,0 +1,108 @@
name: Thread Expansion Experiment
on:
pull_request:
types: [opened, reopened, synchronize, ready_for_review]
workflow_dispatch:
permissions:
contents: read
concurrency:
group: ${{ github.event_name == 'pull_request' && format('{0}-{1}', github.workflow, github.event.pull_request.number) || format('{0}-{1}', github.workflow, github.run_id) }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
jobs:
compare:
if: github.event_name != 'pull_request' || !github.event.pull_request.draft
strategy:
fail-fast: false
matrix:
include:
- label: linux
runner: blacksmith-16vcpu-ubuntu-2404
mode: forks
- label: linux
runner: blacksmith-16vcpu-ubuntu-2404
mode: threads
- label: macos
runner: macos-latest
mode: forks
- label: macos
runner: macos-latest
mode: threads
name: compare (${{ matrix.label }}, ${{ matrix.mode }})
runs-on: ${{ matrix.runner }}
timeout-minutes: 20
env:
EXPERIMENT_MODE: ${{ matrix.mode }}
steps:
- name: Checkout
uses: actions/checkout@v6
with:
persist-credentials: false
submodules: false
- name: Setup Node environment
uses: ./.github/actions/setup-node-env
with:
install-bun: "false"
use-sticky-disk: "false"
- name: Show runtime info
run: |
node -v
pnpm -v
node -e 'const os=require("node:os"); console.log(JSON.stringify({ platform: process.platform, cpuCount: os.cpus().length, totalMemGiB: Math.round((os.totalmem() / 1024 / 1024 / 1024) * 10) / 10, loadavg: os.loadavg() }, null, 2));'
- name: Run thread expansion experiment
shell: bash
run: |
set -euo pipefail
case "$EXPERIMENT_MODE" in
forks)
export OPENCLAW_TEST_FORCE_FORKS=1
POOL="forks"
;;
threads)
export OPENCLAW_TEST_FORCE_THREADS=1
POOL="threads"
;;
*)
echo "Unsupported EXPERIMENT_MODE=$EXPERIMENT_MODE" >&2
exit 1
;;
esac
POLICY_FILES=(
src/commands/agents.test.ts
src/commands/text-format.test.ts
src/auto-reply/chunk.test.ts
)
EVIDENCE_FILES=(
src/commands/backup.test.ts
src/auto-reply/reply/commands-acp/install-hints.test.ts
src/agents/pi-extensions/compaction-safeguard.test.ts
)
echo "## ${RUNNER_OS} / ${EXPERIMENT_MODE}" >> "$GITHUB_STEP_SUMMARY"
echo "" >> "$GITHUB_STEP_SUMMARY"
echo "- Wrapper policy sample files: \`${POLICY_FILES[*]}\`" >> "$GITHUB_STEP_SUMMARY"
echo "- Direct evidence sample files: \`${EVIDENCE_FILES[*]}\`" >> "$GITHUB_STEP_SUMMARY"
echo "" >> "$GITHUB_STEP_SUMMARY"
echo "=== Wrapper policy sample (${EXPERIMENT_MODE}) ==="
SECONDS=0
OPENCLAW_TEST_SHOW_POOL_DECISION=1 node scripts/test-parallel.mjs "${POLICY_FILES[@]}"
wrapper_elapsed=$SECONDS
echo "- wrapper policy sample (${EXPERIMENT_MODE}): ${wrapper_elapsed}s" >> "$GITHUB_STEP_SUMMARY"
echo "=== Direct vitest evidence sample (${POOL}) ==="
SECONDS=0
pnpm vitest run --config vitest.config.ts "--pool=${POOL}" "${EVIDENCE_FILES[@]}"
direct_elapsed=$SECONDS
echo "- direct evidence sample (${POOL}): ${direct_elapsed}s" >> "$GITHUB_STEP_SUMMARY"

3
scripts/test-planner/executor.mjs
View File

@@ -109,7 +109,7 @@ const getShardLabel = (args) => {
export function formatPlanOutput(plan) {
return [
`runtime=${plan.runtimeCapabilities.runtimeProfileName} mode=${plan.runtimeCapabilities.mode} intent=${plan.runtimeCapabilities.intentProfile} memoryBand=${plan.runtimeCapabilities.memoryBand} loadBand=${plan.runtimeCapabilities.loadBand} vitestMaxWorkers=${String(plan.executionBudget.vitestMaxWorkers ?? "default")} topLevelParallel=${plan.topLevelParallelEnabled ? String(plan.topLevelParallelLimit) : "off"}`,
`runtime=${plan.runtimeCapabilities.runtimeProfileName} mode=${plan.runtimeCapabilities.mode} intent=${plan.runtimeCapabilities.intentProfile} memoryBand=${plan.runtimeCapabilities.memoryBand} loadBand=${plan.runtimeCapabilities.loadBand} vitestMaxWorkers=${String(plan.executionBudget.vitestMaxWorkers ?? "default")} topLevelParallel=${plan.topLevelParallelEnabled ? String(plan.topLevelParallelLimit) : "off"} unitPool=${plan.executionBudget.defaultUnitPool} basePool=${plan.executionBudget.defaultBasePool} threadExpansion=${String(plan.executionBudget.threadExpansionEnabled)} threadPolicy=${plan.executionBudget.threadPoolReason}`,
...plan.selectedUnits.map(
(unit) =>
`${unit.id} filters=${String(countExplicitEntryFilters(unit.args) ?? "all")} maxWorkers=${String(
@@ -123,6 +123,7 @@ export function formatExplanation(explanation) {
return [
`file=${explanation.file}`,
`runtime=${explanation.runtimeProfile} intent=${explanation.intentProfile} memoryBand=${explanation.memoryBand} loadBand=${explanation.loadBand}`,
`threadExpansion=${String(explanation.threadExpansionEnabled)} threadPolicy=${explanation.threadPoolReason}`,
`surface=${explanation.surface}`,
`isolate=${explanation.isolate ? "yes" : "no"}`,
`pool=${explanation.pool}`,

40
scripts/test-planner/planner.mjs
View File

@@ -265,12 +265,33 @@ const formatPerFileEntryName = (owner, file) => {
return `${owner}-${baseName}`;
};
const resolvePoolForUnit = (config) => {
if (config.pool) {
return config.pool;
}
const explicitPoolArg = config.args?.find(
(arg) => typeof arg === "string" && arg.startsWith("--pool="),
);
if (explicitPoolArg) {
return explicitPoolArg.slice("--pool=".length);
}
const configIndex = config.args?.findIndex((arg) => arg === "--config") ?? -1;
const vitestConfig = configIndex >= 0 ? (config.args?.[configIndex + 1] ?? "") : "";
if (
vitestConfig === "vitest.extensions.config.ts" ||
vitestConfig === "vitest.channels.config.ts"
) {
return "threads";
}
return "forks";
};
const createExecutionUnit = (context, config) => {
const unit = {
id: config.id,
surface: config.surface,
isolate: Boolean(config.isolate),
pool: config.pool ?? "forks",
pool: resolvePoolForUnit(config),
args: config.args,
env: config.env,
includeFiles: config.includeFiles,
@@ -437,7 +458,7 @@ const buildDefaultUnits = (context, request) => {
"run",
"--config",
"vitest.unit.config.ts",
"--pool=forks",
`--pool=${executionBudget.defaultUnitPool}`,
...noIsolateArgs,
],
reasons: ["unit-fast-shared"],
@@ -522,7 +543,7 @@ const buildDefaultUnits = (context, request) => {
"run",
"--config",
"vitest.unit.config.ts",
"--pool=forks",
"--pool=threads",
...noIsolateArgs,
...catalog.unitThreadPinnedFiles,
],
@@ -666,12 +687,15 @@ const createTargetedUnit = (context, classification, filters) => {
: owner;
const args = (() => {
if (owner === "unit") {
const pool = classification.reasons.includes("unit-pinned-manifest")
? "threads"
: context.executionBudget.defaultUnitPool;
return [
"vitest",
"run",
"--config",
"vitest.unit.config.ts",
"--pool=forks",
`--pool=${pool}`,
...context.noIsolateArgs,
...filters,
];
@@ -682,7 +706,7 @@ const createTargetedUnit = (context, classification, filters) => {
"run",
"--config",
"vitest.config.ts",
"--pool=forks",
"--pool=threads",
...context.noIsolateArgs,
...filters,
];
@@ -745,15 +769,15 @@ const createTargetedUnit = (context, classification, filters) => {
"run",
"--config",
"vitest.config.ts",
`--pool=${classification.isolated ? "forks" : context.executionBudget.defaultBasePool}`,
...context.noIsolateArgs,
...(classification.isolated ? ["--pool=forks"] : []),
...filters,
];
})();
return createExecutionUnit(context, {
id: unitId,
surface: classification.legacyBasePinned ? "base" : classification.surface,
isolate: classification.isolated || owner === "base-pinned",
isolate: classification.isolated,
args,
reasons: classification.reasons,
});
@@ -1191,6 +1215,8 @@ export function explainExecutionTarget(request, options = {}) {
intentProfile: context.runtime.intentProfile,
memoryBand: context.runtime.memoryBand,
loadBand: context.runtime.loadBand,
threadExpansionEnabled: context.executionBudget.threadExpansionEnabled,
threadPoolReason: context.executionBudget.threadPoolReason,
file: classification.file,
surface: classification.legacyBasePinned ? "base" : classification.surface,
isolate: targetedUnit.isolate,

127
scripts/test-planner/runtime-profile.mjs
View File

@@ -87,6 +87,14 @@ const scaleConcurrencyForLoad = (value, loadBand) => {
return Math.max(1, Math.floor(value * scale));
};
const parseTruthyEnv = (value) => {
if (typeof value !== "string") {
return false;
}
const normalized = value.trim().toLowerCase();
return normalized === "1" || normalized === "true";
};
const LOCAL_MEMORY_BUDGETS = {
constrained: {
vitestCap: 2,
@@ -197,6 +205,106 @@ const withIntentBudgetAdjustments = (budget, intentProfile, cpuCount) => {
return budget;
};
export function resolveThreadPoolPolicy(runtimeCapabilities) {
const runtime = runtimeCapabilities;
const forceThreads = runtime.forceThreads;
const forceForks = runtime.forceForks;
if (runtime.isCI) {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: runtime.isWindows ? 1 : 3,
reason: "ci-preserves-current-policy",
};
}
if (forceForks) {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "forced-forks",
};
}
if (forceThreads) {
return {
threadExpansionEnabled: true,
defaultUnitPool: "threads",
defaultBasePool: "threads",
unitFastLaneCount:
runtime.hostCpuCount >= 12 && runtime.hostMemoryGiB >= 96 && runtime.loadBand === "idle"
? 2
: 1,
reason: "forced-threads",
};
}
if (runtime.isWindows) {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "windows-local-conservative",
};
}
if (runtime.intentProfile === "serial") {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "profile-conservative",
};
}
if (runtime.hostMemoryGiB < 64) {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "memory-below-thread-threshold",
};
}
if (runtime.hostCpuCount < 10) {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "cpu-below-thread-threshold",
};
}
if (runtime.loadBand === "busy" || runtime.loadBand === "saturated") {
return {
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "host-under-load",
};
}
return {
threadExpansionEnabled: true,
defaultUnitPool: "threads",
defaultBasePool: "threads",
unitFastLaneCount:
runtime.hostCpuCount >= 12 && runtime.hostMemoryGiB >= 96 && runtime.loadBand === "idle"
? 2
: 1,
reason: "strong-local-host",
};
}
export function resolveRuntimeCapabilities(env = process.env, options = {}) {
const mode = resolveVitestMode(env, options.mode ?? null);
const isCI = mode === "ci";
@@ -219,6 +327,10 @@ export function resolveRuntimeCapabilities(env = process.env, options = {}) {
const loadAware = !isCI && platform !== "win32";
const memoryBand = resolveMemoryBand(hostMemoryGiB);
const loadBand = resolveLoadBand(loadAware, loadRatio);
const forceThreads = parseTruthyEnv(env.OPENCLAW_TEST_FORCE_THREADS);
const forceForks =
parseTruthyEnv(env.OPENCLAW_TEST_FORCE_FORKS) ||
parseTruthyEnv(env.OPENCLAW_TEST_DISABLE_THREAD_EXPANSION);
const runtimeProfileName = isCI
? isWindows
? "ci-windows"
@@ -247,12 +359,15 @@ export function resolveRuntimeCapabilities(env = process.env, options = {}) {
loadAware,
loadRatio,
loadBand,
forceThreads,
forceForks,
};
}
export function resolveExecutionBudget(runtimeCapabilities) {
const runtime = runtimeCapabilities;
const cpuCount = clamp(runtime.hostCpuCount, 1, 16);
const threadPoolPolicy = resolveThreadPoolPolicy(runtime);
if (runtime.isCI) {
const macCiWorkers = runtime.isMacOS ? 1 : null;
@@ -271,10 +386,14 @@ export function resolveExecutionBudget(runtimeCapabilities) {
heavyUnitFileLimit: 64,
heavyUnitLaneCount: 4,
memoryHeavyUnitFileLimit: 64,
unitFastLaneCount: runtime.isWindows ? 1 : 3,
unitFastLaneCount: threadPoolPolicy.unitFastLaneCount,
unitFastBatchTargetMs: runtime.isWindows ? 0 : 45_000,
channelsBatchTargetMs: runtime.isWindows ? 0 : 30_000,
extensionsBatchTargetMs: runtime.isWindows ? 0 : 30_000,
threadExpansionEnabled: threadPoolPolicy.threadExpansionEnabled,
defaultUnitPool: threadPoolPolicy.defaultUnitPool,
defaultBasePool: threadPoolPolicy.defaultBasePool,
threadPoolReason: threadPoolPolicy.reason,
};
}
@@ -294,10 +413,14 @@ export function resolveExecutionBudget(runtimeCapabilities) {
heavyUnitFileLimit: bandBudget.heavyFileLimit,
heavyUnitLaneCount: bandBudget.heavyLaneCount,
memoryHeavyUnitFileLimit: bandBudget.memoryHeavyFileLimit,
unitFastLaneCount: 1,
unitFastLaneCount: threadPoolPolicy.unitFastLaneCount,
unitFastBatchTargetMs: bandBudget.unitFastBatchTargetMs,
channelsBatchTargetMs: 0,
extensionsBatchTargetMs: 0,
threadExpansionEnabled: threadPoolPolicy.threadExpansionEnabled,
defaultUnitPool: threadPoolPolicy.defaultUnitPool,
defaultBasePool: threadPoolPolicy.defaultBasePool,
threadPoolReason: threadPoolPolicy.reason,
};
const loadAdjustedBudget = {

111
test/scripts/test-parallel-pool-policy.test.ts Normal file
View File

@@ -0,0 +1,111 @@
import { describe, expect, it } from "vitest";
import {
resolveExecutionBudget,
resolveRuntimeCapabilities,
resolveThreadPoolPolicy,
} from "../../scripts/test-planner/runtime-profile.mjs";
describe("thread pool policy", () => {
it("keeps constrained local hosts on forks", () => {
const runtime = resolveRuntimeCapabilities(
{},
{
mode: "local",
platform: "darwin",
cpuCount: 8,
totalMemoryBytes: 32 * 1024 ** 3,
loadAverage: [1.6, 1.6, 1.6],
},
);
expect(resolveThreadPoolPolicy(runtime, {})).toMatchObject({
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "memory-below-thread-threshold",
});
});
it("enables threads for strong idle local hosts", () => {
const runtime = resolveRuntimeCapabilities(
{},
{
mode: "local",
platform: "darwin",
cpuCount: 16,
totalMemoryBytes: 128 * 1024 ** 3,
loadAverage: [2, 2, 2],
},
);
expect(resolveThreadPoolPolicy(runtime, {})).toMatchObject({
threadExpansionEnabled: true,
defaultUnitPool: "threads",
defaultBasePool: "threads",
unitFastLaneCount: 2,
reason: "strong-local-host",
});
});
it("disables thread expansion for saturated local hosts", () => {
const runtime = resolveRuntimeCapabilities(
{},
{
mode: "local",
platform: "darwin",
cpuCount: 16,
totalMemoryBytes: 128 * 1024 ** 3,
loadAverage: [17, 17, 17],
},
);
expect(resolveThreadPoolPolicy(runtime, {})).toMatchObject({
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 1,
reason: "host-under-load",
});
});
it("honors explicit force-threads overrides", () => {
const runtime = resolveRuntimeCapabilities(
{ OPENCLAW_TEST_FORCE_THREADS: "1" },
{
mode: "local",
platform: "darwin",
cpuCount: 8,
totalMemoryBytes: 32 * 1024 ** 3,
loadAverage: [8, 8, 8],
},
);
expect(resolveExecutionBudget(runtime)).toMatchObject({
threadExpansionEnabled: true,
defaultUnitPool: "threads",
defaultBasePool: "threads",
threadPoolReason: "forced-threads",
});
});
it("keeps CI on the current unit/base policy", () => {
const runtime = resolveRuntimeCapabilities(
{ CI: "true", GITHUB_ACTIONS: "true" },
{
mode: "ci",
platform: "linux",
cpuCount: 32,
totalMemoryBytes: 128 * 1024 ** 3,
},
);
expect(resolveExecutionBudget(runtime)).toMatchObject({
threadExpansionEnabled: false,
defaultUnitPool: "forks",
defaultBasePool: "forks",
unitFastLaneCount: 3,
threadPoolReason: "ci-preserves-current-policy",
});
});
});

30
test/scripts/test-parallel.test.ts
View File

@@ -195,10 +195,37 @@ describe("scripts/test-parallel lane planning", () => {
);
expect(output).toContain("mode=local intent=normal memoryBand=mid");
expect(output).toContain("unitPool=threads");
expect(output).toContain("basePool=threads");
expect(output).toContain("unit-fast filters=all maxWorkers=");
expect(output).toContain("extensions filters=all maxWorkers=");
});
it("keeps constrained local hosts on forks in plan output", () => {
const repoRoot = path.resolve(import.meta.dirname, "../..");
const output = execFileSync(
"node",
["scripts/test-parallel.mjs", "--plan", "--surface", "unit"],
{
cwd: repoRoot,
env: {
...clearPlannerShardEnv(process.env),
CI: "",
GITHUB_ACTIONS: "",
RUNNER_OS: "macOS",
OPENCLAW_TEST_HOST_CPU_COUNT: "10",
OPENCLAW_TEST_HOST_MEMORY_GIB: "16",
OPENCLAW_TEST_LOAD_AWARE: "0",
},
encoding: "utf8",
},
);
expect(output).toContain("unitPool=forks");
expect(output).toContain("basePool=forks");
expect(output).toContain("threadPolicy=memory-below-thread-threshold");
});
it("explains targeted file ownership and execution policy", () => {
const repoRoot = path.resolve(import.meta.dirname, "../..");
const output = execFileSync(
@@ -213,7 +240,8 @@ describe("scripts/test-parallel lane planning", () => {
expect(output).toContain("surface=base");
expect(output).toContain("reasons=base-surface,base-pinned-manifest");
expect(output).toContain("pool=forks");
expect(output).toContain("pool=threads");
expect(output).toContain("threadPolicy=memory-below-thread-threshold");
});
it("prints the planner-backed CI manifest as JSON", () => {

9
test/scripts/test-planner.test.ts
View File

@@ -41,6 +41,8 @@ describe("test planner", () => {
expect(plan.runtimeCapabilities.runtimeProfileName).toBe("local-darwin");
expect(plan.runtimeCapabilities.memoryBand).toBe("mid");
expect(plan.executionBudget.unitSharedWorkers).toBe(4);
expect(plan.executionBudget.defaultUnitPool).toBe("threads");
expect(plan.executionBudget.defaultBasePool).toBe("threads");
expect(plan.executionBudget.topLevelParallelLimitNoIsolate).toBe(8);
expect(plan.executionBudget.topLevelParallelLimitIsolated).toBe(3);
expect(plan.selectedUnits.some((unit) => unit.id.startsWith("unit-fast"))).toBe(true);
@@ -76,6 +78,8 @@ describe("test planner", () => {
expect(plan.runtimeCapabilities.memoryBand).toBe("mid");
expect(plan.runtimeCapabilities.loadBand).toBe("saturated");
expect(plan.executionBudget.defaultUnitPool).toBe("forks");
expect(plan.executionBudget.defaultBasePool).toBe("forks");
expect(plan.executionBudget.unitSharedWorkers).toBe(2);
expect(plan.executionBudget.topLevelParallelLimitNoIsolate).toBe(4);
expect(plan.executionBudget.topLevelParallelLimitIsolated).toBe(1);
@@ -156,8 +160,11 @@ describe("test planner", () => {
);
expect(explanation.surface).toBe("base");
expect(explanation.pool).toBe("forks");
expect(explanation.pool).toBe("threads");
expect(explanation.isolate).toBe(false);
expect(explanation.reasons).toContain("base-pinned-manifest");
expect(explanation.threadExpansionEnabled).toBe(false);
expect(explanation.threadPoolReason).toBe("memory-below-thread-threshold");
expect(explanation.intentProfile).toBe("normal");
});