src: wake pending web locks when workers exit

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<Lock>(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<LockRequest> grantable_request;
     std::unique_ptr<LockRequest> if_available_request;
+    std::shared_ptr<Lock> granted_lock;
     std::unordered_set<Environment*> 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<Lock>(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<Object> 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<Environment*> 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

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);
+});