diff --git a/src/node_locks.cc b/src/node_locks.cc index 4fca3a7df94..225abb7e78b 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 00000000000..92de78eb2c8 --- /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); +});