code: def test_threadpoolworkers_pick_correct_ioloop(cleanup, loop):
      # gh4057
  
      # About picking appropriate values for the various timings
      # * Sleep time in `access_limited` impacts test runtime but is arbitrary
      # * `lease-timeout` should be smaller than the sleep time. This is what the
      #   test builds on. assuming the leases cannot be refreshed, e.g. wrong
      #   event loop picked / PeriodicCallback never scheduled, the semaphore
      #   would become oversubscribed and the len(protected_resources) becomes
      #   non zero. This should also trigger a log message about "unknown leases"
      #   and fails the test.
      # * `lease-validation-interval` interval should be the smallest quantity.
      #   How often leases are checked for staleness is hard coded atm and a fifth
      #   of the `lease-timeout`. Accounting for this and some jitter, this should
      #   be sufficiently small to ensure smooth operation.
  
      with dask.config.set(
          {
              "distributed.scheduler.locks.lease-validation-interval": 0.01,
              "distributed.scheduler.locks.lease-timeout": 0.1,
          }
      ):
          with Client(
              processes=False,
              dashboard_address=":0",
              threads_per_worker=4,
              loop=loop,
          ) as client:
              sem = Semaphore(max_leases=1, name="database", loop=None)
              assert sem.loop is loop
              protected_resource = []
  
              def access_limited(val, sem):
                  import time
  
                  with sem:
                      assert len(protected_resource) == 0
                      protected_resource.append(val)
                      # Interact with the DB
                      time.sleep(0.01)
                      protected_resource.remove(val)
  
              client.gather(
                  client.map(access_limited, range(10), sem=sem)
              )
              assert sem.refresh_callback.io_loop is loop
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import pickle
  import signal
  from datetime import timedelta
  from time import sleep
  import pytest
  import dask
  from dask.distributed import Client
  from distributed import Semaphore, fire_and_forget
  from distributed.core import ConnectionPool
  from distributed.metrics import time
  from distributed.utils_test import BrokenComm, captured_logger, cluster, gen_cluster, slowidentity
label_desc: Use image node_id 4 for calling a global function with example usage: test_threadpoolworkers_pick_correct_ioloop(cleanup, loop) without return types