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 _check_lease_timeout(self):
      now = time()
      semaphore_names = list(self.leases.keys())
      for name in semaphore_names:
          ids = list(self.leases[name])
          logger.debug(
              "Validating leases for %s at time %s. Currently known %s",
              name,
              now,
              self.leases[name],
          )
          for _id in ids:
              time_since_refresh = now - self.leases[name][_id]
              if time_since_refresh > self.lease_timeout:
                  logger.debug(
                      "Lease %s for %s timed out after %ss.",
                      _id,
                      name,
                      time_since_refresh,
                  )
                  self._release_value(name=name, lease_id=_id)
  
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 9 for calling the SemaphoreExtension obj's underlying member method code with example usage: obj._check_lease_timeout() without return types