class_docstr: An extension for the scheduler to manage MultiLocks
  
  This adds the following routes to the scheduler
  
  *  multi_lock_acquire
  *  multi_lock_release
  
  The approach is to maintain `self.locks` that maps a lock (unique name given to
  `MultiLock(names=, ...)` at creation) to a list of users (instances of `MultiLock`)
  that "requests" the lock. Additionally, `self.requests` maps a user to its requested
  locks and `self.requests_left` maps a user to the number of locks still need.
  
  Every time a user `x` gets to the front in `self.locks[name] = [x, ...]` it means
  that `x` now holds the lock `name` and when it holds all the requested locks
  `acquire()` can return.
  
  Finally, `self.events` contains all the events users are waiting on to finish.
docstr: Request locks
  
  Parameters
  ----------
  locks: List[str]
      Names of the locks to request.
  id: Hashable
      Identifier of the `MultiLock` instance requesting the locks.
  num_locks: int
      Number of locks in `locks` requesting
  
  Return
  ------
  result: bool
      Whether `num_locks` requested locks are free immediately or not.
docstr_tok: ["Request","locks","Parameters","--","--","--","--","--","locks",":","List","[","str","]","Names","of","the","locks","to","request",".","id",":","Hashable","Identifier","of","the","`","MultiLock","`","instance","requesting","the","locks",".","num_locks",":","int","Number","of","locks","in","`","locks","`","requesting","Return","--","--","--","result",":","bool","Whether","`","num_locks","`","requested","locks","are","free","immediately","or","not","."]
code: def _request_locks(
      self, locks: list[str], id: Hashable, num_locks: int
  ) -> bool:
      """Request locks
  
      Parameters
      ----------
      locks: List[str]
          Names of the locks to request.
      id: Hashable
          Identifier of the `MultiLock` instance requesting the locks.
      num_locks: int
          Number of locks in `locks` requesting
  
      Return
      ------
      result: bool
          Whether `num_locks` requested locks are free immediately or not.
      """
      assert id not in self.requests
      self.requests[id] = set(locks)
      assert len(locks) >= num_locks and num_locks > 0
      self.requests_left[id] = num_locks
  
      locks = sorted(locks, key=lambda x: len(self.locks[x]))
      for i, lock in enumerate(locks):
          self.locks[lock].append(id)
          if len(self.locks[lock]) == 1:  # The lock was free
              self.requests_left[id] -= 1
              if self.requests_left[id] == 0:  # Got all locks needed
                  # Since we got all locks need, we can remove the rest of the requests
                  self.requests[id] -= set(locks[i + 1 :])
                  return True
      return False
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import uuid
  from collections import defaultdict
  from collections.abc import Hashable
  from dask.utils import parse_timedelta
  from distributed.client import Client
  from distributed.utils import TimeoutError, log_errors, wait_for
  from distributed.worker import get_worker
label_desc: Use image node_id 2 for calling the MultiLockExtension obj's underlying member method code with example usage: obj._request_locks(locks, id, num_locks) and returns: False, True