From e91b6d2074f29052370dc300a5d59cd3d26d40b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fatih=20G=C3=BCzel?= Date: Sat, 7 Nov 2026 00:23:29 +0300 Subject: [PATCH] src: wake pending web locks when workers exit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wake requests left pending after worker cleanup. register granted locks under the queue mutex to prevent concurrent exclusive grants, and keep target environments alive until wakeups are queued. add regression tests for worker exits and exclusive lock ownership. Fixes: https://github.com/nodejs/node/issues/67216 Signed-off-by: Fatih Güzel --- src/node_locks.cc | 58 ++++--- test/parallel/test-web-locks-worker-exit.js | 181 ++++++++++++++++++++ 2 files changed, 216 insertions(+), 23 deletions(-) create mode 100644 test/parallel/test-web-locks-worker-exit.js diff --git a/src/node_locks.cc b/src/node_locks.cc index 4fca3a7df94e..225abb7e78b5 100644 --- a/src/node_locks.cc +++ b/src/node_locks.cc @@ -228,9 +228,22 @@ void LockManager::ProcessQueue(Environment* env) { // Remove locks that were stolen from this Environment first CleanupStolenLocks(env); + // The caller must hold mutex_ when recording a granted lock. + auto grant_lock = [this, env](LockRequest& request) { + auto lock = std::make_shared(env, + request.name(), + request.mode(), + request.client_id(), + request.waiting_promise(), + request.released_promise()); + held_locks_[request.name()].push_back(lock); + return lock; + }; + while (true) { std::unique_ptr grantable_request; std::unique_ptr if_available_request; + std::shared_ptr granted_lock; std::unordered_set other_envs_to_wake; /** @@ -306,13 +319,17 @@ void LockManager::ProcessQueue(Environment* env) { // Found a request that can be granted normally grantable_request = std::move(*queue_iter); pending_queue_.erase(queue_iter); + if (!grantable_request->steal()) { + // Reserve the lock before another Environment can process the queue. + granted_lock = grant_lock(*grantable_request); + } break; } - } - // Wake each environment only once - for (Environment* target_env : other_envs_to_wake) { - WakeEnvironment(target_env); + // Keep target Environments alive until their wakeups are queued. + for (Environment* target_env : other_envs_to_wake) { + WakeEnvironment(target_env); + } } /** @@ -441,30 +458,18 @@ void LockManager::ProcessQueue(Environment* env) { held_locks_.erase(held_locks_iter); } } - } + granted_lock = grant_lock(*grantable_request); - // Wake other environments - for (Environment* target_env : envs_to_notify) { - if (target_env != env) { - WakeEnvironment(target_env); + // Keep target Environments alive until their wakeups are queued. + for (Environment* target_env : envs_to_notify) { + if (target_env != env) { + WakeEnvironment(target_env); + } } } } - // Create and store the new granted lock - auto granted_lock = - std::make_shared(env, - grantable_request->name(), - grantable_request->mode(), - grantable_request->client_id(), - grantable_request->waiting_promise(), - grantable_request->released_promise()); - { - Mutex::ScopedLock scoped_lock(mutex_); - held_locks_[grantable_request->name()].push_back(granted_lock); - } - - // Create and store the new granted lock + // Create the lock info object. Local lock_info; if (!CreateLockInfoObject(env, *grantable_request).ToLocal(&lock_info)) { return; @@ -794,17 +799,24 @@ void LockManager::CleanupEnvironment(Environment* env_to_cleanup) { } // Remove every pending request submitted by this Environment. + std::unordered_set envs_to_wake; for (auto request_iter = pending_queue_.begin(); request_iter != pending_queue_.end();) { if ((*request_iter)->env() == env_to_cleanup) { request_iter = pending_queue_.erase(request_iter); } else { + envs_to_wake.insert((*request_iter)->env()); ++request_iter; } } // Finally, remove it from registered_envs_ registered_envs_.erase(env_to_cleanup); + + // Queue wakeups under mutex_ so target Environments cannot be destroyed. + for (Environment* env : envs_to_wake) { + WakeEnvironment(env); + } } // Cleanup hook wrapper diff --git a/test/parallel/test-web-locks-worker-exit.js b/test/parallel/test-web-locks-worker-exit.js new file mode 100644 index 000000000000..92de78eb2c81 --- /dev/null +++ b/test/parallel/test-web-locks-worker-exit.js @@ -0,0 +1,181 @@ +'use strict'; + +const common = require('../common'); +const assert = require('node:assert'); +const { once } = require('node:events'); +const { test } = require('node:test'); +const { Worker } = require('node:worker_threads'); + +async function waitForGrant(request) { + let timeout; + try { + return await Promise.race([ + request, + new Promise((resolve, reject) => { + timeout = setTimeout(() => { + reject(new Error('Lock request was not granted after worker exit')); + }, common.platformTimeout(5000)); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + +for (const exitType of ['exit', 'throw', 'resolve-exit', 'release']) { + for (const waiterThread of ['main', 'worker']) { + test(`${exitType} wakes a pending lock request in ${waiterThread}`, async (t) => { + const name = `${exitType}-${waiterThread}`; + const holder = new Worker(` + const { parentPort, workerData } = require('node:worker_threads'); + navigator.locks.request(workerData.name, () => { + return new Promise((resolve) => { + parentPort.once('message', () => { + switch (workerData.exitType) { + case 'throw': + throw new Error('lock holder failed'); + case 'resolve-exit': + resolve(); + process.exit(); + break; + case 'release': + resolve(); + break; + default: + process.exit(); + } + }); + parentPort.postMessage('held'); + }); + }); + `, { eval: true, workerData: { name, exitType } }); + t.after(() => holder.terminate()); + let workerError; + holder.on('error', (err) => { workerError = err; }); + assert.deepStrictEqual(await once(holder, 'message'), ['held']); + + let request; + if (waiterThread === 'worker') { + const waiter = new Worker(` + const { parentPort, workerData } = require('node:worker_threads'); + // A pending lock request alone does not keep the worker alive. + parentPort.on('message', () => {}); + navigator.locks.request(workerData, () => { + parentPort.postMessage('granted'); + }); + parentPort.postMessage('queued'); + `, { eval: true, workerData: name }); + t.after(() => waiter.terminate()); + assert.deepStrictEqual(await once(waiter, 'message'), ['queued']); + request = once(waiter, 'message').then(([message]) => message); + } else { + request = navigator.locks.request(name, common.mustCall(() => 'granted')); + const { pending } = await navigator.locks.query(); + assert.strictEqual(pending.length, 1); + assert.strictEqual(pending[0].name, name); + } + + const exited = new Promise((resolve) => holder.once('exit', resolve)); + holder.postMessage('exit'); + assert.strictEqual(await exited, exitType === 'throw' ? 1 : 0); + if (exitType === 'throw') { + assert.strictEqual(workerError.message, 'lock holder failed'); + } else { + assert.strictEqual(workerError, undefined); + } + assert.strictEqual(await waitForGrant(request), 'granted'); + }); + } +} + +test('exiting with a pending exclusive request unblocks a shared request', async (t) => { + const name = 'pending-worker-exit'; + await navigator.locks.request(name, { mode: 'shared' }, common.mustCall(async () => { + const worker = new Worker(` + const { parentPort, workerData } = require('node:worker_threads'); + parentPort.once('message', () => process.exit()); + navigator.locks.request(workerData, () => { + throw new Error('Exclusive lock must stay blocked'); + }); + parentPort.postMessage('queued'); + `, { eval: true, workerData: name }); + t.after(() => worker.terminate()); + assert.deepStrictEqual(await once(worker, 'message'), ['queued']); + + const request = navigator.locks.request( + name, { mode: 'shared' }, common.mustCall(() => 'granted')); + const { pending } = await navigator.locks.query(); + assert.strictEqual(pending.length, 1); + assert.strictEqual(pending[0].mode, 'shared'); + + const exited = once(worker, 'exit'); + worker.postMessage('exit'); + assert.deepStrictEqual(await exited, [0]); + assert.strictEqual(await waitForGrant(request), 'granted'); + })); +}); + +test('concurrent holder exits wake multiple requests and workers', async (t) => { + const name = 'concurrent-worker-exit'; + const active = new Int32Array(new SharedArrayBuffer(4)); + const holders = Array.from({ length: 2 }, () => new Worker(` + const { parentPort, workerData } = require('node:worker_threads'); + navigator.locks.request(workerData, { mode: 'shared' }, () => { + parentPort.once('message', () => process.exit()); + parentPort.postMessage('held'); + return new Promise(() => {}); + }); + `, { eval: true, workerData: name })); + t.after(() => Promise.all(holders.map((worker) => worker.terminate()))); + assert.deepStrictEqual( + await Promise.all(holders.map((worker) => once(worker, 'message'))), + [['held'], ['held']]); + + // Queue worker waiters first so cleanup wakes competing Environments. + const waiters = Array.from({ length: 8 }, () => new Worker(` + const assert = require('node:assert'); + const { parentPort, workerData } = require('node:worker_threads'); + const active = new Int32Array(workerData.counter); + parentPort.on('message', () => {}); + navigator.locks.request(workerData.name, () => { + assert.strictEqual(Atomics.add(active, 0, 1), 0); + return new Promise(resolve => { + parentPort.once('message', () => { + assert.strictEqual(Atomics.sub(active, 0, 1), 1); + resolve(); + }); + parentPort.postMessage('granted'); + }); + }).then(() => parentPort.postMessage('released')); + parentPort.postMessage('queued'); + `, { eval: true, workerData: { name, counter: active.buffer } })); + t.after(() => Promise.all(waiters.map((worker) => worker.terminate()))); + assert.deepStrictEqual( + await Promise.all(waiters.map((worker) => once(worker, 'message'))), + Array.from({ length: 8 }, () => ['queued'])); + const granted = waiters.map((worker) => new Promise((resolve, reject) => { + worker.on('message', (message) => { + if (message === 'granted') worker.postMessage('release'); + if (message === 'released') resolve('granted'); + }); + worker.once('error', reject); + })); + + const requests = []; + for (const value of [0, 1]) { + requests.push(navigator.locks.request(name, common.mustCall(() => { + assert.strictEqual(Atomics.add(active, 0, 1), 0); + assert.strictEqual(Atomics.sub(active, 0, 1), 1); + return value; + }))); + } + const { pending } = await navigator.locks.query(); + assert.strictEqual(pending.length, 2); + + const exited = holders.map((worker) => once(worker, 'exit')); + const completed = Promise.all([...exited, ...requests, ...granted]); + for (const worker of holders) worker.postMessage('exit'); + assert.deepStrictEqual(await waitForGrant(completed), + [[0], [0], 0, 1, ...Array(8).fill('granted')]); + assert.strictEqual(Atomics.load(active, 0), 0); +});