code: def main(  # type: ignore[no-untyped-def]
      scheduler,
      host,
      worker_port: str | None,
      listen_address,
      contact_address,
      nanny_port: str | None,
      nthreads,
      n_workers,
      nanny,
      name,
      pid_file,
      resources,
      dashboard,
      scheduler_file,
      dashboard_prefix,
      tls_ca_file,
      tls_cert,
      tls_key,
      dashboard_address,
      worker_class,
      preload_nanny,
      **kwargs,
  ):
      """Launch a Dask worker attached to an existing scheduler"""
  
      if "dask-worker" in sys.argv[0]:
          warnings.warn(
              "dask-worker is deprecated and will be removed in a future release; use `dask worker` instead",
              FutureWarning,
              stacklevel=1,
          )
  
      (
          g0,
          g1,
          g2,
      ) = (
          gc.get_threshold()
      )  # https://github.com/dask/distributed/issues/1653
      gc.set_threshold(g0 * 3, g1 * 3, g2 * 3)
  
      enable_proctitle_on_current()
      enable_proctitle_on_children()
  
      sec = {
          k: v
          for k, v in [
              ("tls_ca_file", tls_ca_file),
              ("tls_worker_cert", tls_cert),
              ("tls_worker_key", tls_key),
          ]
          if v is not None
      }
  
      if n_workers == "auto":
          n_workers, nthreads = nprocesses_nthreads()
      elif n_workers is None:
          n_workers = 1
      else:
          n_workers = int(n_workers)
  
      if n_workers < 0:
          n_workers = CPU_COUNT + 1 + n_workers
  
      if n_workers <= 0:
          logger.error(
              "Failed to launch worker. Must specify --nworkers so that there's at least one process."
          )
          sys.exit(1)
  
      if n_workers > 1 and not nanny:
          logger.error(
              "Failed to launch worker.  You cannot use the --no-nanny argument when n_workers > 1."
          )
          sys.exit(1)
  
      if contact_address and not listen_address:
          logger.error(
              "Failed to launch worker. "
              "Must specify --listen-address when --contact-address is given"
          )
          sys.exit(1)
  
      if n_workers > 1 and listen_address:
          logger.error(
              "Failed to launch worker. "
              "You cannot specify --listen-address when n_workers > 1."
          )
          sys.exit(1)
  
      if (worker_port or host) and listen_address:
          logger.error(
              "Failed to launch worker. "
              "You cannot specify --listen-address when --worker-port or --host is given."
          )
          sys.exit(1)
  
      try:
          if listen_address:
              host, _ = get_address_host_port(
                  listen_address, strict=True
              )
              worker_port = str(_)
              if ":" in host:
                  # IPv6 -- bracket to pass as user args
                  host = f"[{host}]"
  
          if contact_address:
              # we only need this to verify it is getting parsed
              (_, _) = get_address_host_port(
                  contact_address, strict=True
              )
          else:
              # if contact address is not present we use the listen_address for contact
              contact_address = listen_address
      except ValueError as e:  # pragma: no cover
          logger.error("Failed to launch worker. " + str(e))
          sys.exit(1)
  
      if not nthreads:
          nthreads = CPU_COUNT // n_workers
  
      if pid_file:
          with open(pid_file, "w") as f:
              f.write(str(os.getpid()))
  
          def del_pid_file():
              if os.path.exists(pid_file):
                  os.remove(pid_file)
  
          atexit.register(del_pid_file)
  
      if resources:
          resources = resources.replace(",", " ").split()
          resources = dict(pair.split("=") for pair in resources)
          resources = valmap(float, resources)
      else:
          resources = None
  
      worker_class = import_term(worker_class)
  
      port_kwargs = _apportion_ports(
          worker_port, nanny_port, n_workers, nanny
      )
      assert len(port_kwargs) == n_workers
  
      if nanny:
          kwargs["worker_class"] = worker_class
          kwargs["preload_nanny"] = preload_nanny
          kwargs["listen_address"] = listen_address
          t = Nanny
      else:
          t = worker_class
  
      if (
          not scheduler
          and not scheduler_file
          and dask.config.get("scheduler-address", None) is None
      ):
          raise ValueError(
              "Need to provide scheduler address like\n"
              "dask worker SCHEDULER_ADDRESS:8786"
          )
  
      with suppress(TypeError, ValueError):
          name = int(name)
  
      signal_fired = False
  
      async def run():
          nannies = [
              t(
                  scheduler,
                  scheduler_file=scheduler_file,
                  nthreads=nthreads,
                  resources=resources,
                  security=sec,
                  contact_address=contact_address,
                  host=host,
                  dashboard=dashboard,
                  dashboard_address=dashboard_address,
                  name=name
                  if n_workers == 1 or name is None or name == ""
                  else str(name) + "-" + str(i),
                  **kwargs,
                  **port_kwargs_i,
              )
              for i, port_kwargs_i in enumerate(port_kwargs)
          ]
  
          async def wait_for_nannies_to_finish():
              """Wait for all nannies to initialize and finish"""
              await asyncio.gather(*nannies)
              await asyncio.gather(*(n.finished() for n in nannies))
  
          async def wait_for_signals_and_close():
              """Wait for SIGINT or SIGTERM and close all nannies upon receiving one of those signals"""
              nonlocal signal_fired
              signum = await wait_for_signals()
  
              signal_fired = True
              if nanny:
                  # Unregister all workers from scheduler
                  await asyncio.gather(
                      *(
                          n.close(timeout=10, reason=f"signal-{signum}")
                          for n in nannies
                      )
                  )
  
          wait_for_signals_and_close_task = asyncio.create_task(
              wait_for_signals_and_close()
          )
          wait_for_nannies_to_finish_task = asyncio.create_task(
              wait_for_nannies_to_finish()
          )
  
          done, _ = await asyncio.wait(
              [
                  wait_for_signals_and_close_task,
                  wait_for_nannies_to_finish_task,
              ],
              return_when=asyncio.FIRST_COMPLETED,
          )
          # Re-raise exceptions from done tasks
          [task.result() for task in done]
  
      try:
          asyncio_run(run(), loop_factory=get_loop_factory())
      except (TimeoutError, asyncio.TimeoutError):
          # We already log the exception in nanny / worker. Don't do it again.
          if not signal_fired:
              logger.info("Timed out starting worker")
          sys.exit(1)
      finally:
          logger.info("End worker")
  
imports: from __future__ import annotations
  import asyncio
  import atexit
  import gc
  import logging
  import os
  import sys
  import warnings
  from collections.abc import Iterator
  from contextlib import suppress
  from typing import Any
  import click
  from tlz import valmap
  from tornado.ioloop import TimeoutError
  import dask
  from dask.system import CPU_COUNT
  from distributed import Nanny
  from distributed._signals import wait_for_signals
  from distributed.comm import get_address_host_port
  from distributed.compatibility import asyncio_run
  from distributed.config import get_loop_factory
  from distributed.deploy.utils import nprocesses_nthreads
  from distributed.preloading import validate_preload_argv
  from distributed.proctitle import enable_proctitle_on_children, enable_proctitle_on_current
  from distributed.utils import import_term, parse_ports
label_desc: Use image node_id 1 for calling a global function with example usage: main(scheduler, host, worker_port, listen_address, contact_address, nanny_port, nthreads, n_workers, nanny, name, pid_file, resources, dashboard, scheduler_file, dashboard_prefix, tls_ca_file, tls_cert, tls_key, dashboard_address, worker_class, preload_nanny) without return types