text: import asyncio
  import functools
  import logging
  import os
  import platform
  import signal
  import socket
  import sys
  import threading
  import time
  from email.utils import formatdate
  
  import click
  
  HANDLED_SIGNALS = (
      signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.
      signal.SIGTERM,  # Unix signal 15. Sent by `kill <pid>`.
  )
  
  logger = logging.getLogger("uvicorn.error")
  
  
  class ServerState:
      """
      Shared servers state that is available between all protocol instances.
      """
  
      def __init__(self):
          self.total_requests = 0
          self.connections = set()
          self.tasks = set()
          self.default_headers = []
  
  
  class Server:
      def __init__(self, config):
          self.config = config
          self.server_state = ServerState()
  
          self.started = False
          self.should_exit = False
          self.force_exit = False
          self.last_notified = 0
  
      def run(self, sockets=None):
          self.config.setup_event_loop()
          loop = asyncio.get_event_loop()
          loop.run_until_complete(self.serve(sockets=sockets))
  
      async def serve(self, sockets=None):
          process_id = os.getpid()
  
          config = self.config
          if not config.loaded:
              config.load()
  
          self.lifespan = config.lifespan_class(config)
  
          self.install_signal_handlers()
  
          message = "Started server process [%d]"
          color_message = "Started server process [" + click.style("%d", fg="cyan") + "]"
          logger.info(message, process_id, extra={"color_message": color_message})
  
          await self.startup(sockets=sockets)
          if self.should_exit:
              return
          await self.main_loop()
          await self.shutdown(sockets=sockets)
  
          message = "Finished server process [%d]"
          color_message = "Finished server process [" + click.style("%d", fg="cyan") + "]"
          logger.info(
              "Finished server process [%d]",
              process_id,
              extra={"color_message": color_message},
          )
  
      async def startup(self, sockets=None):
          await self.lifespan.startup()
          if self.lifespan.should_exit:
              self.should_exit = True
              return
  
          config = self.config
  
          create_protocol = functools.partial(
              config.http_protocol_class, config=config, server_state=self.server_state
          )
  
          loop = asyncio.get_event_loop()
  
          if sockets is not None:
              # Explicitly passed a list of open sockets.
              # We use this when the server is run from a Gunicorn worker.
  
              def _share_socket(sock: socket) -> socket:
                  # Windows requires the socket be explicitly shared across
                  # multiple workers (processes).
                  from socket import fromshare  # type: ignore
  
                  sock_data = sock.share(os.getpid())  # type: ignore
                  return fromshare(sock_data)
  
              self.servers = []
              for sock in sockets:
                  if config.workers > 1 and platform.system() == "Windows":
                      sock = _share_socket(sock)
                  server = await loop.create_server(
                      create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog
                  )
                  self.servers.append(server)
  
          elif config.fd is not None:
              # Use an existing socket, from a file descriptor.
              sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
              server = await loop.create_server(
                  create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog
              )
              message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
              logger.info(message % str(sock.getsockname()))
              self.servers = [server]
  
          elif config.uds is not None:
              # Create a socket using UNIX domain socket.
              uds_perms = 0o666
              if os.path.exists(config.uds):
                  uds_perms = os.stat(config.uds).st_mode
              server = await loop.create_unix_server(
                  create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
              )
              os.chmod(config.uds, uds_perms)
              message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
              logger.info(message % config.uds)
              self.servers = [server]
  
          else:
              # Standard case. Create a socket from a host/port pair.
              addr_format = "%s://%s:%d"
              if config.host and ":" in config.host:
                  # It's an IPv6 address.
                  addr_format = "%s://[%s]:%d"
  
              try:
                  server = await loop.create_server(
                      create_protocol,
                      host=config.host,
                      port=config.port,
                      ssl=config.ssl,
                      backlog=config.backlog,
                  )
              except OSError as exc:
                  logger.error(exc)
                  await self.lifespan.shutdown()
                  sys.exit(1)
              port = config.port
              if port == 0:
                  port = server.sockets[0].getsockname()[1]
              protocol_name = "https" if config.ssl else "http"
              message = f"Uvicorn running on {addr_format} (Press CTRL+C to quit)"
              color_message = (
                  "Uvicorn running on "
                  + click.style(addr_format, bold=True)
                  + " (Press CTRL+C to quit)"
              )
              logger.info(
                  message,
                  protocol_name,
                  config.host,
                  port,
                  extra={"color_message": color_message},
              )
              self.servers = [server]
  
          self.started = True
  
      async def main_loop(self):
          counter = 0
          should_exit = await self.on_tick(counter)
          while not should_exit:
              counter += 1
              counter = counter % 864000
              await asyncio.sleep(0.1)
              should_exit = await self.on_tick(counter)
  
      async def on_tick(self, counter) -> bool:
          # Update the default headers, once per second.
          if counter % 10 == 0:
              current_time = time.time()
              current_date = formatdate(current_time, usegmt=True).encode()
              self.server_state.default_headers = [
                  (b"date", current_date)
              ] + self.config.encoded_headers
  
              # Callback to `callback_notify` once every `timeout_notify` seconds.
              if self.config.callback_notify is not None:
                  if current_time - self.last_notified > self.config.timeout_notify:
                      self.last_notified = current_time
                      await self.config.callback_notify()
  
          # Determine if we should exit.
          if self.should_exit:
              return True
          if self.config.limit_max_requests is not None:
              return self.server_state.total_requests >= self.config.limit_max_requests
          return False
  
      async def shutdown(self, sockets=None):
          logger.info("Shutting down")
  
          # Stop accepting new connections.
          for server in self.servers:
              server.close()
          for sock in sockets or []:
              sock.close()
          for server in self.servers:
              await server.wait_closed()
  
          # Request shutdown on all existing connections.
          for connection in list(self.server_state.connections):
              connection.shutdown()
          await asyncio.sleep(0.1)
  
          # Wait for existing connections to finish sending responses.
          if self.server_state.connections and not self.force_exit:
              msg = "Waiting for connections to close. (CTRL+C to force quit)"
              logger.info(msg)
              while self.server_state.connections and not self.force_exit:
                  await asyncio.sleep(0.1)
  
          # Wait for existing tasks to complete.
          if self.server_state.tasks and not self.force_exit:
              msg = "Waiting for background tasks to complete. (CTRL+C to force quit)"
              logger.info(msg)
              while self.server_state.tasks and not self.force_exit:
                  await asyncio.sleep(0.1)
  
          # Send the lifespan shutdown event, and wait for application shutdown.
          if not self.force_exit:
              await self.lifespan.shutdown()
  
      def install_signal_handlers(self) -> None:
          if threading.current_thread() is not threading.main_thread():
              # Signals can only be listened to from the main thread.
              return
  
          loop = asyncio.get_event_loop()
  
          try:
              for sig in HANDLED_SIGNALS:
                  loop.add_signal_handler(sig, self.handle_exit, sig, None)
          except NotImplementedError:
              # Windows
              for sig in HANDLED_SIGNALS:
                  signal.signal(sig, self.handle_exit)
  
      def handle_exit(self, sig, frame):
          if self.should_exit:
              self.force_exit = True
          else:
              self.should_exit = True
  