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: Cancel/release previously requested/acquired locks
  
  Parameters
  ----------
  locks: List[str]
      Names of the locks to refain.
  id: Hashable
      Identifier of the `MultiLock` instance refraining the locks.
docstr_tok: ["Cancel\/release","previously","requested\/acquired","locks","Parameters","--","--","--","--","--","locks",":","List","[","str","]","Names","of","the","locks","to","refain",".","id",":","Hashable","Identifier","of","the","`","MultiLock","`","instance","refraining","the","locks","."]
code: def _refain_locks(self, locks, id):
      """Cancel/release previously requested/acquired locks
  
      Parameters
      ----------
      locks: List[str]
          Names of the locks to refain.
      id: Hashable
          Identifier of the `MultiLock` instance refraining the locks.
      """
      waiters_ready = set()
      for lock in locks:
          if self.locks[lock][0] == id:
              self.locks[lock].pop(0)
              if self.locks[lock]:
                  new_first = self.locks[lock][0]
                  self.requests_left[new_first] -= 1
                  if self.requests_left[new_first] <= 0:
                      # Notice, `self.requests_left[new_first]` might go below zero
                      # if more locks are freed than requested.
                      self.requests_left[new_first] = 0
                      waiters_ready.add(new_first)
          else:
              self.locks[lock].remove(id)
          assert id not in self.locks[lock]
      del self.requests[id]
      del self.requests_left[id]
  
      for waiter in waiters_ready:
          self.scheduler.loop.add_callback(self.events[waiter].set)
  
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 3 for calling the MultiLockExtension obj's underlying member method code with example usage: obj._refain_locks(locks, id) without return types