class_docstr: An extension for the scheduler to manage Semaphores
  
  This adds the following routes to the scheduler
  
  * semaphore_acquire
  * semaphore_release
  * semaphore_close
  * semaphore_refresh_leases
  * semaphore_register
class_docstr_tok: ["An","extension","for","the","scheduler","to","manage","Semaphores","This","adds","the","following","routes","to","the","scheduler","*","semaphore_acquire","*","semaphore_release","*","semaphore_close","*","semaphore_refresh_leases","*","semaphore_register"]
code: def __init__(self, scheduler):
      self.scheduler = scheduler
  
      # {semaphore_name: asyncio.Event}
      self.events = defaultdict(asyncio.Event)
      # {semaphore_name: max_leases}
      self.max_leases = dict()
      # {semaphore_name: {lease_id: lease_last_seen_timestamp}}
      self.leases = defaultdict(dict)
  
      self.scheduler.handlers.update(
          {
              "semaphore_register": self.create,
              "semaphore_acquire": self.acquire,
              "semaphore_release": self.release,
              "semaphore_close": self.close,
              "semaphore_refresh_leases": self.refresh_leases,
              "semaphore_value": self.get_value,
          }
      )
  
      # {metric_name: {semaphore_name: metric}}
      self.metrics = {
          "acquire_total": defaultdict(int),  # counter
          "release_total": defaultdict(int),  # counter
          "average_pending_lease_time": defaultdict(float),  # gauge
          "pending": defaultdict(int),  # gauge
      }
  
      validation_callback_time = parse_timedelta(
          dask.config.get(
              "distributed.scheduler.locks.lease-validation-interval"
          ),
          default="s",
      )
      self.scheduler.periodic_callbacks[
          "semaphore-lease-timeout"
      ] = pc = PeriodicCallback(
          self._check_lease_timeout, validation_callback_time * 1000
      )
      pc.start()
      self.lease_timeout = parse_timedelta(
          dask.config.get("distributed.scheduler.locks.lease-timeout"),
          default="s",
      )
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import uuid
  import warnings
  from asyncio import TimeoutError
  from collections import defaultdict, deque
  import dask
  from dask.utils import parse_timedelta
  from distributed.compatibility import PeriodicCallback
  from distributed.metrics import time
  from distributed.utils import Deadline, SyncMethodMixin, log_errors, wait_for
  from distributed.utils_comm import retry_operation
  from distributed.worker import get_client, get_worker
label_desc: Use image node_id 1 to create a new SemaphoreExtension object  with example: obj = SemaphoreExtension(scheduler)