Skip to content

test_runner: do not reuse a worker ID held by a running file #65739

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vserpokryl wants to merge 1 commit into
base: main
Choose a base branch
from
Open

test_runner: do not reuse a worker ID held by a running file #65739

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion doc/api/test.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4575,7 +4575,9 @@ The unique identifier of the worker running the current test file. This value is
derived from the `NODE_TEST_WORKER_ID` environment variable. When running tests
with `--test-isolation=process` (the default), each test file runs in a separate
child process and is assigned a worker ID from 1 to N, where N is the number of
concurrent workers. When running with `--test-isolation=none`, all tests run in
concurrent workers. A worker ID is never shared by two test files running at the
same time. Once a test file finishes, its worker ID is reused by the next test
file that starts. When running with `--test-isolation=none`, all tests run in
the same process and the worker ID is always 1. This value is `undefined` when
not running in a test context.

Expand Down
202 changes: 100 additions & 102 deletions lib/internal/test_runner/runner.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ const {
ArrayPrototypeSlice,
ArrayPrototypeSome,
ArrayPrototypeSort,
MathMax,
ObjectAssign,
PromisePrototypeThen,
PromiseWithResolvers,
Expand All @@ -38,7 +37,6 @@ const {
const { spawn } = require('child_process');
const { statSync } = require('fs');
const { finished } = require('internal/streams/end-of-stream');
const { availableParallelism } = require('os');
const { resolve, sep, isAbsolute } = require('path');
const { DefaultDeserializer, DefaultSerializer } = require('v8');
const { getOptionValue, getOptionsAsFlagsFromBinding } = require('internal/options');
Expand Down Expand Up @@ -139,17 +137,22 @@ let kResistStopPropagation;

// Worker ID pool management for concurrent test execution
class WorkerIdPool {
#nextId = 0;
#maxConcurrency;

constructor(maxConcurrency) {
this.#maxConcurrency = maxConcurrency;
}
#acquiredIds = new SafeSet();

acquire() {
const id = (this.#nextId++ % this.#maxConcurrency) + 1;
let id = 1;

while (this.#acquiredIds.has(id)) {
id++;
}

this.#acquiredIds.add(id);
return id;
}

release(id) {
this.#acquiredIds.delete(id);
}
}

function createTestFileList(patterns, cwd) {
Expand Down Expand Up @@ -538,94 +541,102 @@ function runTestFile(path, filesWatcher, opts) {
debug('Assigned worker ID %d to test file: %s', workerId, path);
}

if (watchMode) {
stdio.push('ipc');
env.WATCH_REPORT_DEPENDENCIES = '1';
}
if (opts.root.harness.shouldColorizeTestFiles) {
env.FORCE_COLOR = '1';
}

const child = spawn(
process.execPath, args,
{
__proto__: null,
signal: t.signal,
encoding: 'utf8',
env,
stdio,
cwd: opts.cwd,
},
);
if (watchMode) {
filesWatcher.runningProcesses.set(path, child);
filesWatcher.watcher.watchChildProcessModules(child, path);
}

let err;
try {
if (watchMode) {
stdio.push('ipc');
env.WATCH_REPORT_DEPENDENCIES = '1';
}
if (opts.root.harness.shouldColorizeTestFiles) {
env.FORCE_COLOR = '1';
}

child.on('error', (error) => {
err = error;
});
const child = spawn(
process.execPath, args,
{
__proto__: null,
signal: t.signal,
encoding: 'utf8',
env,
stdio,
cwd: opts.cwd,
},
);
if (watchMode) {
filesWatcher.runningProcesses.set(path, child);
filesWatcher.watcher.watchChildProcessModules(child, path);
}

child.stdout.on('data', (data) => {
subtest.parseMessage(data);
});
let err;

const rl = new Interface({ __proto__: null, input: child.stderr });
rl.on('line', (line) => {
if (isInspectorMessage(line)) {
process.stderr.write(line + '\n');
return;
}
child.on('error', (error) => {
err = error;
});

// stderr cannot be treated as TAP, per the spec. However, we want to
// surface stderr lines to improve the DX. Inject each line into the
// test output as an unknown token as if it came from the TAP parser.
subtest.addToReport({
__proto__: null,
type: 'test:stderr',
data: { __proto__: null, file: path, message: line + '\n' },
child.stdout.on('data', (data) => {
subtest.parseMessage(data);
});
});

const { 0: { 0: code, 1: signal } } = await SafePromiseAll([
once(child, 'exit', { __proto__: null, signal: t.signal }),
finished(child.stdout, { __proto__: null, signal: t.signal }),
]);

// Close readline interface to prevent memory leak
rl.close();

if (watchMode) {
filesWatcher.runningProcesses.delete(path);
filesWatcher.runningSubtests.delete(path);
(async () => {
try {
await subTestEnded;
} finally {
if (filesWatcher.runningSubtests.size === 0) {
opts.root.reporter[kEmitMessage]('test:watch:drained');
opts.root.postRun();
}
const rl = new Interface({ __proto__: null, input: child.stderr });
rl.on('line', (line) => {
if (isInspectorMessage(line)) {
process.stderr.write(line + '\n');
return;
}
})();
}

if (code !== 0 || signal !== null) {
if (!err) {
const failureType = subtest.failedSubtests ? kSubtestsFailed : kTestCodeFailure;
err = ObjectAssign(new ERR_TEST_FAILURE('test failed', failureType), {
// stderr cannot be treated as TAP, per the spec. However, we want to
// surface stderr lines to improve the DX. Inject each line into the
// test output as an unknown token as if it came from the TAP parser.
subtest.addToReport({
__proto__: null,
exitCode: code,
signal: signal,
// The stack will not be useful since the failures came from tests
// in a child process.
stack: undefined,
type: 'test:stderr',
data: { __proto__: null, file: path, message: line + '\n' },
});
});

const { 0: { 0: code, 1: signal } } = await SafePromiseAll([
once(child, 'exit', { __proto__: null, signal: t.signal }),
finished(child.stdout, { __proto__: null, signal: t.signal }),
]);

// Close readline interface to prevent memory leak
rl.close();

if (watchMode) {
filesWatcher.runningProcesses.delete(path);
filesWatcher.runningSubtests.delete(path);
(async () => {
try {
await subTestEnded;
} finally {
if (filesWatcher.runningSubtests.size === 0) {
opts.root.reporter[kEmitMessage]('test:watch:drained');
opts.root.postRun();
}
}
})();
}

throw err;
if (code !== 0 || signal !== null) {
if (!err) {
const failureType = subtest.failedSubtests ? kSubtestsFailed : kTestCodeFailure;
err = ObjectAssign(new ERR_TEST_FAILURE('test failed', failureType), {
__proto__: null,
exitCode: code,
signal: signal,
// The stack will not be useful since the failures came from tests
// in a child process.
stack: undefined,
});
}

throw err;
}
} finally {
// Every exit path must return the ID, including abort and spawn failure.
if (opts.workerIdPool && workerId !== undefined) {
opts.workerIdPool.release(workerId);
debug('Released worker ID %d from test file: %s', workerId, path);
}
}
});
const subTestEnded = subtest.start();
Expand Down Expand Up @@ -1011,23 +1022,10 @@ function run(options = kEmptyObject) {
let filesWatcher;
let runFiles;

// Create worker ID pool for concurrent test execution.
// Use concurrency from globalOptions which has been processed by parseCommandLine().
const effectiveConcurrency = globalOptions.concurrency ?? concurrency;
let maxConcurrency = 1;
if (effectiveConcurrency === true) {
maxConcurrency = MathMax(availableParallelism() - 1, 1);
} else if (typeof effectiveConcurrency === 'number') {
maxConcurrency = effectiveConcurrency;
}
const workerIdPool = new WorkerIdPool(maxConcurrency);
debug(
'Created worker ID pool with max concurrency: %d, ' +
'effectiveConcurrency: %s, testFiles: %d',
maxConcurrency,
effectiveConcurrency,
testFiles.length,
);
// The pool tracks the IDs actually in use, so they stay exclusive and never
// exceed the number of files running concurrently.
const workerIdPool = new WorkerIdPool();
debug('Created worker ID pool, testFiles: %d', testFiles.length);

const opts = {
__proto__: null,
Expand Down
Loading
Loading