ext/web/locks.rs - main - githubmirror/deno

ext/web/locks.rs
475 строк13 KB
crowlbot
fix(ext/node): implement worker_threads.locks via Web Locks (#35963)
22 июл 2026, 16:15
22 июл 2026, 16:152731794
// Copyright 2018-2026 the Deno authors. MIT license.

use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::VecDeque;
use std::fmt::Display;
use std::rc::Rc;
use std::sync::LazyLock;
use std::sync::Mutex;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;

use deno_core::OpState;
use deno_core::Resource;
use deno_core::ResourceId;
use deno_core::op2;
use tokio::sync::oneshot;

#[derive(
  serde::Serialize, serde::Deserialize, Debug, PartialEq, Clone, Copy,
)]
#[serde(rename_all = "camelCase")]
enum LockMode {
  Shared,
  Exclusive,
}

struct HeldLock {
  name: String,
  mode: LockMode,
  id: u64,
  client_id: String,
  // Fires when the lock is stolen, so the holder's `request()` promise can
  // reject with an AbortError. `None` until the holder starts awaiting it.
  broken_tx: Option<oneshot::Sender<()>>,
}

struct PendingRequest {
  name: String,
  mode: LockMode,
  id: u64,
  client_id: String,
  tx: oneshot::Sender<bool>,
}

struct LockState {
  held: Vec<HeldLock>,
  queues: HashMap<String, VecDeque<PendingRequest>>,
  counter: u64,
}

static LOCK_STATE: LazyLock<Mutex<LockState>> = LazyLock::new(|| {
  Mutex::new(LockState {
    held: vec![],
    queues: HashMap::new(),
    counter: 0,
  })
});

static CLIENT_ID_COUNTER: AtomicU64 = AtomicU64::new(1);

struct LockClientId(String);

pub fn worker_lock_client_id(worker_id: impl Display) -> String {
  format!("worker-{worker_id}")
}

pub fn set_lock_client_id(state: &mut OpState, client_id: String) {
  state.put(LockClientId(client_id));
}

fn get_client_id(state: &mut OpState) -> String {
  if let Some(id) = state.try_borrow::<LockClientId>() {
    return id.0.clone();
  }
  let id = CLIENT_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
  let client_id = format!("{id}");
  state.put(LockClientId(client_id.clone()));
  client_id
}

fn grantable(held: &[HeldLock], name: &str, mode: LockMode) -> bool {
  match mode {
    LockMode::Exclusive => !held.iter().any(|h| h.name == name),
    LockMode::Shared => !held
      .iter()
      .any(|h| h.name == name && h.mode == LockMode::Exclusive),
  }
}

fn process_queue(state: &mut LockState, name: &str) {
  let queue = match state.queues.get_mut(name) {
    Some(q) => q,
    None => return,
  };

  while let Some(front) = queue.front() {
    if !grantable(&state.held, name, front.mode) {
      break;
    }
    let request = queue.pop_front().unwrap();
    if request.tx.send(true).is_ok() {
      state.held.push(HeldLock {
        name: request.name,
        mode: request.mode,
        id: request.id,
        client_id: request.client_id,
        broken_tx: None,
      });
    }
    // If send fails (receiver dropped / cancelled), skip this request
  }
}

fn release_lock(state: &mut LockState, id: u64) {
  if let Some(pos) = state.held.iter().position(|h| h.id == id) {
    let lock = state.held.remove(pos);
    process_queue(state, &lock.name);
  }
}

fn cancel_request(state: &mut LockState, id: u64) {
  for queue in state.queues.values_mut() {
    if let Some(pos) = queue.iter().position(|r| r.id == id) {
      let req = queue.remove(pos).unwrap();
      let _ = req.tx.send(false);
      return;
    }
  }
}

/// Whether `client_id` currently holds any lock. Used at worker teardown to
/// decide whether the worker's JS must be halted before its held locks are
/// handed off to other clients (see `WorkerThread::drop`). A worker that holds
/// nothing needs no halt, so we avoid interrupting it — halting an arbitrary
/// worker can abort an in-progress synthetic module instantiation during boot.
pub fn client_holds_lock(client_id: &str) -> bool {
  let state = LOCK_STATE
    .lock()
    .unwrap_or_else(|poisoned| poisoned.into_inner());
  state.held.iter().any(|lock| lock.client_id == client_id)
}

pub fn cleanup_locks_for_client_id(client_id: &str) {
  // Runs from `WorkerThread::drop`, so recover from a poisoned mutex instead
  // of `unwrap()`ing: a panic here would be a panic during unwinding, which
  // aborts the process.
  let mut state = LOCK_STATE
    .lock()
    .unwrap_or_else(|poisoned| poisoned.into_inner());

  // Most workers never touch Web Locks; skip the held/queue scan entirely when
  // there is nothing to clean up.
  if state.held.is_empty() && state.queues.is_empty() {
    return;
  }

  let mut affected_names = HashSet::new();

  state.held.retain(|lock| {
    if lock.client_id == client_id {
      affected_names.insert(lock.name.clone());
      false
    } else {
      true
    }
  });

  // Drop this client's pending requests, rejecting each (`send(false)`), and
  // keep the rest in original order. `VecDeque::remove` is O(n), so queues that
  // hold one of this client's requests are drained into a fresh queue in one
  // O(n) pass rather than removed by index; queues without any are left as-is.
  for (name, queue) in state.queues.iter_mut() {
    if !queue.iter().any(|request| request.client_id == client_id) {
      continue;
    }
    let mut kept = VecDeque::with_capacity(queue.len());
    while let Some(request) = queue.pop_front() {
      if request.client_id == client_id {
        let _ = request.tx.send(false);
        affected_names.insert(name.clone());
      } else {
        kept.push_back(request);
      }
    }
    *queue = kept;
  }

  for name in affected_names {
    process_queue(&mut state, &name);
  }
}

// Resource for a held lock — releases the lock on drop
struct HeldLockResource {
  id: u64,
  // Resolves when the lock is stolen (see `op_lock_manager_await_steal`).
  broken_rx: RefCell<Option<oneshot::Receiver<()>>>,
}

// Sets up the steal-notification channel for a freshly granted lock: the
// sender is stored on the held lock in the global state and the receiver is
// stored on the resource handed back to the holder.
fn make_held_lock_resource(state: &mut LockState, id: u64) -> HeldLockResource {
  let (tx, rx) = oneshot::channel();
  if let Some(held) = state.held.iter_mut().find(|h| h.id == id) {
    held.broken_tx = Some(tx);
  }
  HeldLockResource {
    id,
    broken_rx: RefCell::new(Some(rx)),
  }
}

impl Drop for HeldLockResource {
  fn drop(&mut self) {
    let mut state = LOCK_STATE.lock().unwrap();
    release_lock(&mut state, self.id);
  }
}

impl Resource for HeldLockResource {
  fn name(&self) -> Cow<'_, str> {
    "webLock".into()
  }
}

// Resource for a pending lock request — cancels the request on drop
struct PendingLockResource {
  rx: RefCell<Option<oneshot::Receiver<bool>>>,
  id: u64,
}

impl Drop for PendingLockResource {
  fn drop(&mut self) {
    let mut state = LOCK_STATE.lock().unwrap();
    cancel_request(&mut state, self.id);
  }
}

impl Resource for PendingLockResource {
  fn name(&self) -> Cow<'_, str> {
    "pendingWebLock".into()
  }
}

/// Result from op_lock_manager_request.
/// status: 0 = granted, 1 = pending, 2 = not available (ifAvailable)
#[derive(serde::Serialize)]
struct LockRequestResult {
  status: u8,
  rid: ResourceId,
}

/// Synchronous op: registers a lock request.
/// Returns immediately with either a granted lock, a pending handle, or
/// a not-available indicator.
#[op2]
#[serde]
pub fn op_lock_manager_request(
  state: &mut OpState,
  #[string] name: String,
  #[serde] mode: LockMode,
  if_available: bool,
  steal: bool,
) -> LockRequestResult {
  let client_id = get_client_id(state);
  let mut ls = LOCK_STATE.lock().unwrap();

  ls.counter += 1;
  let id = ls.counter;

  if steal {
    // Notify the current holders that their lock has been broken, then remove
    // all held locks for this name. The holders' `request()` promises reject
    // with an AbortError. Pending requests are left untouched: the stealing
    // request jumps to the front of the queue (below) and is granted ahead of
    // them, but they remain queued and are granted once the steal is released.
    for held in ls.held.iter_mut().filter(|h| h.name == name) {
      if let Some(tx) = held.broken_tx.take() {
        let _ = tx.send(());
      }
    }
    ls.held.retain(|h| h.name != name);
  } else if if_available && !grantable(&ls.held, &name, mode) {
    return LockRequestResult { status: 2, rid: 0 };
  }

  let (tx, mut rx) = oneshot::channel();
  let name_clone = name.clone();
  let queue = ls.queues.entry(name).or_default();

  if steal {
    queue.push_front(PendingRequest {
      name: name_clone.clone(),
      mode,
      id,
      client_id,
      tx,
    });
  } else {
    queue.push_back(PendingRequest {
      name: name_clone.clone(),
      mode,
      id,
      client_id,
      tx,
    });
  }

  process_queue(&mut ls, &name_clone);

  // Check if granted immediately (process_queue sends synchronously
  // through the oneshot before we check)
  match rx.try_recv() {
    Ok(true) => {
      let resource = make_held_lock_resource(&mut ls, id);
      drop(ls);
      let rid = state.resource_table.add(resource);
      LockRequestResult { status: 0, rid }
    }
    _ => {
      drop(ls);
      let rid = state.resource_table.add(PendingLockResource {
        rx: RefCell::new(Some(rx)),
        id,
      });
      LockRequestResult { status: 1, rid }
    }
  }
}

/// Async op: waits for a pending lock request to be granted.
/// Returns the held lock resource ID, or null if the request was
/// cancelled/stolen.
#[op2]
#[smi]
pub async fn op_lock_manager_await_lock(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
) -> Option<ResourceId> {
  let (rx, lock_id) = {
    let state = state.borrow();
    let pending = state.resource_table.get::<PendingLockResource>(rid).ok()?;
    let rx = pending.rx.borrow_mut().take()?;
    (rx, pending.id)
  };

  let granted = rx.await.unwrap_or(false);

  // Clean up the pending resource (Drop will no-op since request
  // is already resolved or cancelled in global state)
  let _ = state
    .borrow_mut()
    .resource_table
    .take::<PendingLockResource>(rid);

  if granted {
    let resource = {
      let mut ls = LOCK_STATE.lock().unwrap();
      make_held_lock_resource(&mut ls, lock_id)
    };
    let held_rid = state.borrow_mut().resource_table.add(resource);
    Some(held_rid)
  } else {
    None
  }
}

/// Async op: resolves while a held lock is alive. Returns `true` if the lock
/// was stolen (the holder's `request()` promise must reject with AbortError),
/// or `false` when the lock is released normally. Keeping this op pending also
/// keeps the event loop alive for as long as the lock is held.
#[op2]
pub async fn op_lock_manager_await_steal(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
) -> bool {
  let rx = {
    let state = state.borrow();
    let Ok(held) = state.resource_table.get::<HeldLockResource>(rid) else {
      return false;
    };
    let Some(rx) = held.broken_rx.borrow_mut().take() else {
      return false;
    };
    rx
  };
  // `Ok(())` means the lock was stolen; `Err(_)` means the sender was dropped
  // because the lock was released normally.
  rx.await.is_ok()
}

/// Synchronous op: reports whether a held lock has been stolen. A steal removes
/// the lock from `held` synchronously (see `op_lock_manager_request`), while the
/// steal notification that `op_lock_manager_await_steal` waits on is only
/// delivered a full event-loop turn later. A fast-returning callback can
/// therefore finish before that async notification arrives, so `request()`
/// re-checks synchronously via this op to guarantee a stolen lock still rejects.
#[op2(fast)]
pub fn op_lock_manager_is_stolen(
  state: &mut OpState,
  #[smi] rid: ResourceId,
) -> bool {
  let Ok(held) = state.resource_table.get::<HeldLockResource>(rid) else {
    return false;
  };
  let id = held.id;
  let ls = LOCK_STATE
    .lock()
    .unwrap_or_else(|poisoned| poisoned.into_inner());
  // Only a steal removes a still-held lock from `held` before its holder
  // releases it, so a missing id means the lock was stolen.
  !ls.held.iter().any(|h| h.id == id)
}

/// Cancels a pending lock request (used by AbortSignal).
#[op2(fast)]
pub fn op_lock_manager_cancel(state: &mut OpState, #[smi] rid: ResourceId) {
  if let Ok(pending) = state.resource_table.get::<PendingLockResource>(rid) {
    let id = pending.id;
    drop(pending);
    let mut ls = LOCK_STATE.lock().unwrap();
    cancel_request(&mut ls, id);
  }
}

/// Releases a held lock.
#[op2(fast)]
pub fn op_lock_manager_release(state: &mut OpState, #[smi] rid: ResourceId) {
  // Taking the resource drops it, which triggers release_lock via Drop
  let _ = state.resource_table.take::<HeldLockResource>(rid);
}

#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct QueryLock {
  name: String,
  mode: LockMode,
  client_id: String,
}

#[derive(serde::Serialize)]
struct Query {
  held: Vec<QueryLock>,
  pending: Vec<QueryLock>,
}

#[op2]
#[serde]
pub fn op_lock_manager_query() -> Query {
  let ls = LOCK_STATE.lock().unwrap();
  let held: Vec<QueryLock> = ls
    .held
    .iter()
    .map(|h| QueryLock {
      name: h.name.clone(),
      mode: h.mode,
      client_id: h.client_id.clone(),
    })
    .collect();
  let mut pending: Vec<QueryLock> = vec![];
  for queue in ls.queues.values() {
    for p in queue {
      pending.push(QueryLock {
        name: p.name.clone(),
        mode: p.mode,
        client_id: p.client_id.clone(),
      });
    }
  }
  Query { held, pending }
}