code: def main(
      host,
      port,
      protocol,
      interface,
      show,
      dashboard,
      dashboard_prefix,
      use_xheaders,
      pid_file,
      tls_ca_file,
      tls_cert,
      tls_key,
      dashboard_address,
      jupyter,
      **kwargs,
  ):
      """Launch a Dask scheduler."""
  
      if "dask-scheduler" in sys.argv[0]:
          warnings.warn(
              "dask-scheduler is deprecated and will be removed in a future release; use `dask scheduler` 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()
  
      if interface and "," in interface:
          interface = interface.split(",")
  
      if protocol and "," in protocol:
          protocol = protocol.split(",")
  
      if port:
          if "," in port:
              port = [int(p) for p in port.split(",")]
          else:
              port = int(port)
  
      if port is None and (not host or not re.search(r":\d", host)):
          if isinstance(protocol, list):
              port = [8786] + [0] * (len(protocol) - 1)
          else:
              port = 8786
  
      if isinstance(protocol, list) or isinstance(port, list):
          if (
              not isinstance(protocol, list)
              or not isinstance(port, list)
          ) or len(port) != len(protocol):
              raise ValueError(
                  "--protocol and --port must both be lists of equal length"
              )
  
      sec = {
          k: v
          for k, v in [
              ("tls_ca_file", tls_ca_file),
              ("tls_scheduler_cert", tls_cert),
              ("tls_scheduler_key", tls_key),
          ]
          if v is not None
      }
  
      if not host and (tls_ca_file or tls_cert or tls_key):
          host = "tls://"
  
      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 sys.platform.startswith("linux"):
          import resource  # module fails importing on Windows
  
          soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
          limit = max(soft, hard // 2)
          resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))
  
      async def run():
          logger.info("-" * 47)
  
          scheduler = Scheduler(
              security=sec,
              host=host,
              port=port,
              protocol=protocol,
              interface=interface,
              dashboard=dashboard,
              dashboard_address=dashboard_address,
              http_prefix=dashboard_prefix,
              jupyter=jupyter,
              **kwargs,
          )
          logger.info("-" * 47)
  
          async def wait_for_scheduler_to_finish():
              """Wait for the scheduler to initialize and finish"""
              await scheduler
              await scheduler.finished()
  
          async def wait_for_signals_and_close():
              """Wait for SIGINT or SIGTERM and close the scheduler upon receiving one of those signals"""
              signum = await wait_for_signals()
              await scheduler.close(reason=f"signal-{signum}")
  
          wait_for_signals_and_close_task = asyncio.create_task(
              wait_for_signals_and_close()
          )
          wait_for_scheduler_to_finish_task = asyncio.create_task(
              wait_for_scheduler_to_finish()
          )
  
          done, _ = await asyncio.wait(
              [
                  wait_for_signals_and_close_task,
                  wait_for_scheduler_to_finish_task,
              ],
              return_when=asyncio.FIRST_COMPLETED,
          )
          # Re-raise exceptions from done tasks
          [task.result() for task in done]
          logger.info("Stopped scheduler at %r", scheduler.address)
  
      try:
          asyncio_run(run(), loop_factory=get_loop_factory())
      finally:
          logger.info("End scheduler")
  
file_path: distributed/distributed/cli/dask_scheduler.py
imports: from __future__ import annotations
  import asyncio
  import atexit
  import gc
  import logging
  import os
  import re
  import sys
  import warnings
  import click
  from distributed import Scheduler
  from distributed._signals import wait_for_signals
  from distributed.compatibility import asyncio_run
  from distributed.config import get_loop_factory
  from distributed.preloading import validate_preload_argv
  from distributed.proctitle import enable_proctitle_on_children, enable_proctitle_on_current
label_desc: Use image node_id 1 for calling a global function with example usage: main(host, port, protocol, interface, show, dashboard, dashboard_prefix, use_xheaders, pid_file, tls_ca_file, tls_cert, tls_key, dashboard_address, jupyter) without return types