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.
code: def __init__(self, scheduler):
      self.scheduler = scheduler
      self.locks = defaultdict(list)  # lock -> users
      self.requests = {}  # user -> locks
      self.requests_left = {}  # user -> locks still needed
      self.events = {}
  
      self.scheduler.handlers.update(
          {
              "multi_lock_acquire": self.acquire,
              "multi_lock_release": self.release,
          }
      )
  
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 1 to create a new MultiLockExtension object  with example: obj = MultiLockExtension(scheduler)