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 create(self, name=None, max_leases=None):
      # We use `self.max_leases` as the point of truth to find out if a semaphore with a specific
      # `name` has been created.
      if name not in self.max_leases:
          assert isinstance(max_leases, int), max_leases
          self.max_leases[name] = max_leases
      else:
          if max_leases != self.max_leases[name]:
              raise ValueError(
                  "Inconsistent max leases: %s, expected: %s"
                  % (max_leases, self.max_leases[name])
              )
  
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 3 for calling the SemaphoreExtension obj's underlying member method code with example usage: obj.create(name, max_leases) without return types