def bind_unix_socket(
        file: str, mode: int = 0o600, backlog: int = _DEFAULT_BACKLOG
    ) -> socket.socket:
        """Creates a listening unix socket.

        If a socket with the given name already exists, it will be deleted.
        If any other file with that name exists, an exception will be
        raised.

        Returns a socket object (not a list of socket objects like
        `bind_sockets`)
        """
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        set_close_exec(sock.fileno())
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        except socket.error as e:
            if errno_from_exception(e) != errno.ENOPROTOOPT:
                # Hurd doesn't support SO_REUSEADDR
                raise
        sock.setblocking(False)
        try:
            st = os.stat(file)
        except OSError as err:
            if errno_from_exception(err) != errno.ENOENT:
                raise
        else:
            if stat.S_ISSOCK(st.st_mode):
                os.remove(file)
            else:
                raise ValueError("File %s exists and is not a socket", file)
        sock.bind(file)
        os.chmod(file, mode)
        sock.listen(backlog)
        return sock

def GetSecretFile(secret):
  """Fetches the named secret into a temporary file for use with
  modules requiring the contents be accessible via a named file (e.g.
  Python SSL for keys and certificates).
  """
  if sys.platform.startswith('linux'):
    # Linux-specific implementation:  use an unnamed tempfile, which
    # will cease to exist when this process does.  Use /dev/fd to get
    # a name for the file.
    # Note that other platforms (including Mac) have /dev/fd as well,
    # but its semantics are different (all copies of a /dev/fd
    # file share one seek position, and that position is not reset on
    # open), so it's only safe to use on linux.
    if secret not in _tempfile_map:
      f = tempfile.TemporaryFile()
      set_close_exec(f.fileno())
      f.write(GetSecret(secret))
      f.flush()
      _tempfile_map[secret] = f

    return '/dev/fd/%d' % _tempfile_map[secret].fileno()
  else:
    # Default implementation: use a normal named tempfile, and delete
    # it when possible with atexit.
    if secret not in _tempfile_map:
      _, name = tempfile.mkstemp()
      with open(name, 'w') as f:
        f.write(GetSecret(secret))
      _tempfile_map[secret] = name
      atexit.register(os.remove, name)

    return _tempfile_map[secret]

    def initialize(self, impl, time_func=None):
        super(PollIOLoop, self).initialize()
        self._impl = impl
        if hasattr(self._impl, 'fileno'):
            set_close_exec(self._impl.fileno())
        self.time_func = time_func or time.time
        self._handlers = {}
        self._events = {}
        self._callbacks = []
        self._callback_lock = threading.Lock()
        self._timeouts = []
        self._cancellations = 0
        self._running = False
        self._stopped = False
        self._closing = False
        self._thread_ident = None
        self._blocking_signal_threshold = None
        self._timeout_counter = itertools.count()

        # Create a pipe that we send bogus data to when we want to wake
        # the I/O loop when it is idle
        self._waker = Waker()
        self.add_handler(self._waker.fileno(),
                         lambda fd, events: self._waker.consume(),
                         self.READ)

    def bind_unix_socket(file, mode=0o600, backlog=128):
        """Creates a listening unix socket.

        If a socket with the given name already exists, it will be deleted.
        If any other file with that name exists, an exception will be
        raised.

        Returns a socket object (not a list of socket objects like
        `bind_sockets`)
        """
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        set_close_exec(sock.fileno())
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.setblocking(0)
        try:
            st = os.stat(file)
        except OSError as err:
            if err.errno != errno.ENOENT:
                raise
        else:
            if stat.S_ISSOCK(st.st_mode):
                os.remove(file)
            else:
                raise ValueError("File %s exists and is not a socket", file)
        sock.bind(file)
        os.chmod(file, mode)
        sock.listen(backlog)
        return sock

def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128, flags=None):
    """Creates listening sockets bound to the given port and address.

    Returns a list of socket objects (multiple sockets are returned if
    the given address maps to multiple IP addresses, which is most common
    for mixed IPv4 and IPv6 use).

    Address may be either an IP address or hostname.  If it's a hostname,
    the server will listen on all IP addresses associated with the
    name.  Address may be an empty string or None to listen on all
    available interfaces.  Family may be set to either `socket.AF_INET`
    or `socket.AF_INET6` to restrict to IPv4 or IPv6 addresses, otherwise
    both will be used if available.

    The ``backlog`` argument has the same meaning as for
    `socket.listen() <socket.socket.listen>`.

    ``flags`` is a bitmask of AI_* flags to `~socket.getaddrinfo`, like
    ``socket.AI_PASSIVE | socket.AI_NUMERICHOST``.
    """
    sockets = []
    if address == "":
        address = None
    if not socket.has_ipv6 and family == socket.AF_UNSPEC:
        # Python can be compiled with --disable-ipv6, which causes
        # operations on AF_INET6 sockets to fail, but does not
        # automatically exclude those results from getaddrinfo
        # results.
        # http://bugs.python.org/issue16208
        family = socket.AF_INET
    if flags is None:
        flags = socket.AI_PASSIVE
    for res in set(socket.getaddrinfo(address, port, family, socket.SOCK_STREAM,
                                      0, flags)):
        af, socktype, proto, canonname, sockaddr = res
        try:
            sock = socket.socket(af, socktype, proto)
        except socket.error as e:
            if e.args[0] == errno.EAFNOSUPPORT:
                continue
            raise
        set_close_exec(sock.fileno())
        if os.name != 'nt':
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        if af == socket.AF_INET6:
            # On linux, ipv6 sockets accept ipv4 too by default,
            # but this makes it impossible to bind to both
            # 0.0.0.0 in ipv4 and :: in ipv6.  On other systems,
            # separate sockets *must* be used to listen for both ipv4
            # and ipv6.  For consistency, always disable ipv4 on our
            # ipv6 sockets and use a separate ipv4 socket when needed.
            #
            # Python 2.x on windows doesn't have IPPROTO_IPV6.
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
        sock.setblocking(0)
        sock.bind(sockaddr)
        sock.listen(backlog)
        sockets.append(sock)
    return sockets

 def _create_stream(self, max_buffer_size, af, addr, source_ip=None,
                    source_port=None):
     # Always connect in plaintext; we'll convert to ssl if necessary
     # after one connection has completed.
     source_port_bind = source_port if isinstance(source_port, int) else 0
     source_ip_bind = source_ip
     if source_port_bind and not source_ip:
         # User required a specific port, but did not specify
         # a certain source IP, will bind to the default loopback.
         source_ip_bind = '::1' if af == socket.AF_INET6 else '127.0.0.1'
         # Trying to use the same address family as the requested af socket:
         # - 127.0.0.1 for IPv4
         # - ::1 for IPv6
     socket_obj = socket.socket(af)
     set_close_exec(socket_obj.fileno())
     if source_port_bind or source_ip_bind:
         # If the user requires binding also to a specific IP/port.
         try:
             socket_obj.bind((source_ip_bind, source_port_bind))
         except socket.error:
             socket_obj.close()
             # Fail loudly if unable to use the IP/port.
             raise
     try:
         stream = IOStream(socket_obj,
                           max_buffer_size=max_buffer_size)
     except socket.error as e:
         fu = Future()
         fu.set_exception(e)
         return fu
     else:
         return stream, stream.connect(addr)

def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128):
    """ 创建绑定到指定端口和地址的监听sockets，返回socket对象的一个list。 """
    sockets = []
    if address == "":
        address = None
    flags = socket.AI_PASSIVE  # server socket
    if hasattr(socket, "AI_ADDRCONFIG"):
        # AI_ADDRCONFIG ensures that we only try to bind on ipv6 if the system is configured for it,
        # but the flag doesn't exist on some platforms (specifically WinXP, although newer versions of windows have it)
        flags |= socket.AI_ADDRCONFIG
    for res in set(socket.getaddrinfo(address, port, family, socket.SOCK_STREAM, 0, flags)):
        af, socktype, proto, canonname, sockaddr = res
        sock = socket.socket(af, socktype, proto)
        set_close_exec(sock.fileno())
        if os.name != "nt":
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        if af == socket.AF_INET6:
            # On linux, ipv6 sockets accept ipv4 too by default, but this makes it impossible to bind to both
            # 0.0.0.0 in ipv4 and :: in ipv6.  On other systems, separate sockets *must* be used to listen for both ipv4
            # and ipv6.  For consistency, always disable ipv4 on our ipv6 sockets and use a separate ipv4 socket when needed.
            #
            # Python 2.x on windows doesn't have IPPROTO_IPV6.
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
        sock.setblocking(0)
        sock.bind(sockaddr)
        sock.listen(backlog)  # 开始监听，非阻塞
        sockets.append(sock)
    return sockets

    def __init__(self, listener, application=None, backlog=2048,
                 socket_type=socket.SOCK_STREAM,
                 address_family=socket.AF_INET):
        self.address_family = address_family
        self.socket_type = socket_type
        host, port = listener

        if isinstance(application, Application):
            self._server = HTTPServer(application)
        elif isinstance(application, TCPServer):
            self._server = application
        elif callable(application):
            tapp = tornado.wsgi.WSGIContainer(application)
            self._server = HTTPServer(tapp)
        else:
            raise TypeError(
                "Unsupported application type: %r" % (application,))

        if host.startswith('fd://'):
            fd = int(host.split('://')[1])
            set_close_exec(fd)
            sock = socket.fromfd(fd, address_family, socket_type)
            sock.setblocking(0)
            socks = [sock]
        elif self.address_family == socket.AF_UNIX:
            filename = host[len('unix:'):]
            sock = tornado.netutil.bind_unix_socket(filename, backlog=backlog)
            socks = [sock]
        else:
            socks = tornado.netutil.bind_sockets(
                port, host, address_family, backlog)
        self._server.add_sockets(socks)
        self.application = application

 def accept_handler(fd: socket.socket, events: int) -> None:
     # More connections may come in while we're handling callbacks;
     # to prevent starvation of other tasks we must limit the number
     # of connections we accept at a time.  Ideally we would accept
     # up to the number of connections that were waiting when we
     # entered this method, but this information is not available
     # (and rearranging this method to call accept() as many times
     # as possible before running any callbacks would have adverse
     # effects on load balancing in multiprocess configurations).
     # Instead, we use the (default) listen backlog as a rough
     # heuristic for the number of connections we can reasonably
     # accept at once.
     for i in range(_DEFAULT_BACKLOG):
         if removed[0]:
             # The socket was probably closed
             return
         try:
             connection, address = sock.accept()
         except socket.error as e:
             # _ERRNO_WOULDBLOCK indicate we have accepted every
             # connection that is available.
             if errno_from_exception(e) in _ERRNO_WOULDBLOCK:
                 return
             # ECONNABORTED indicates that there was a connection
             # but it was closed while still in the accept queue.
             # (observed on FreeBSD).
             if errno_from_exception(e) == errno.ECONNABORTED:
                 continue
             raise
         set_close_exec(connection.fileno())
         callback(connection, address)

    def socket(self, family=socket.AF_UNSPEC):
        """Return a Unix domain socket instead of tcp socket"""

        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        set_close_exec(sock.fileno())
        sock.setblocking(0)
        
        return sock

 def setUp(self):
     super(ReactorReaderWriterTest, self).setUp()
     r, w = os.pipe()
     self._set_nonblocking(r)
     self._set_nonblocking(w)
     set_close_exec(r)
     set_close_exec(w)
     self._p1 = os.fdopen(r, "rb", 0)
     self._p2 = os.fdopen(w, "wb", 0)

 def setUp(self):
     self._reactor = TornadoReactor(IOLoop())
     r, w = os.pipe()
     self._set_nonblocking(r)
     self._set_nonblocking(w)
     set_close_exec(r)
     set_close_exec(w)
     self._p1 = os.fdopen(r, "rb", 0)
     self._p2 = os.fdopen(w, "wb", 0)

def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128):
    """Creates listening sockets bound to the given port and address.

    Returns a list of socket objects (multiple sockets are returned if
    the given address maps to multiple IP addresses, which is most common
    for mixed IPv4 and IPv6 use).

    Address may be either an IP address or hostname.  If it's a hostname,
    the server will listen on all IP addresses associated with the
    name.  Address may be an empty string or None to listen on all
    available interfaces.  Family may be set to either socket.AF_INET
    or socket.AF_INET6 to restrict to ipv4 or ipv6 addresses, otherwise
    both will be used if available.

    The ``backlog`` argument has the same meaning as for
    ``socket.listen()``.
    """
    sockets = []
    if address == "":
        address = None
    flags = socket.AI_PASSIVE
    if hasattr(socket, "AI_ADDRCONFIG"):
        # AI_ADDRCONFIG ensures that we only try to bind on ipv6
        # if the system is configured for it, but the flag doesn't
        # exist on some platforms (specifically WinXP, although
        # newer versions of windows have it)
        flags |= socket.AI_ADDRCONFIG
    for res in set(socket.getaddrinfo(address, port, family, socket.SOCK_STREAM,
                                  0, flags)):
        af, socktype, proto, canonname, sockaddr = res
        sock = socket.socket(af, socktype, proto)
        set_close_exec(sock.fileno())

        ## The semantics of SO_REUSEADDR is very different from what is
        ## actually specified and implemented on various Unix flavours.
        if platform.system() != 'Windows':
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        if af == socket.AF_INET6:
            # On linux, ipv6 sockets accept ipv4 too by default,
            # but this makes it impossible to bind to both
            # 0.0.0.0 in ipv4 and :: in ipv6.  On other systems,
            # separate sockets *must* be used to listen for both ipv4
            # and ipv6.  For consistency, always disable ipv4 on our
            # ipv6 sockets and use a separate ipv4 socket when needed.
            #
            # Python 2.x on windows doesn't have IPPROTO_IPV6.
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
        sock.setblocking(0)
        sock.bind(sockaddr)
        sock.listen(backlog)
        sockets.append(sock)
    return sockets

    def socket(self, family=socket.AF_UNSPEC):
        res = socket.getaddrinfo(self._host, self._port, family,
                                 socket.SOCK_STREAM, 0, socket.AI_PASSIVE)[0]
        af, socktype, proto, canonname, sockaddr = res
        sock = socket.socket(af, socktype, proto)
        set_close_exec(sock.fileno())
        sock.setblocking(0)
        if af == socket.AF_INET6:
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)

        return sock

    def test_set_close_exec(self):
        # set_close_exec works with sockets.
        s = socket.socket()
        self.addCleanup(s.close)
        set_close_exec(s.fileno())

        # But it doesn't work with pipes.
        r, w = os.pipe()
        self.addCleanup(functools.partial(os.close, r))
        self.addCleanup(functools.partial(os.close, w))
        with self.assertRaises(WindowsError):
            set_close_exec(r)

    def test_set_close_exec(self):
        # set_close_exec works with sockets.
        s = socket.socket()
        self.addCleanup(s.close)
        set_close_exec(s.fileno())

        # But it doesn't work with pipes.
        r, w = os.pipe()
        self.addCleanup(functools.partial(os.close, r))
        self.addCleanup(functools.partial(os.close, w))
        with self.assertRaises(WindowsError) as cm:
            set_close_exec(r)
        ERROR_INVALID_HANDLE = 6
        self.assertEqual(cm.exception.winerror, ERROR_INVALID_HANDLE)

    def _start_background_udev_monitoring(self):
        monitor = pyudev.Monitor.from_netlink(self.context, 'udev')
        monitor.filter_by(subsystem='net')
        monitor.start()

        monitor_fileno = monitor.fileno()

        # we do not trust the udev library....
        set_close_exec(monitor_fileno)
        fcntl.fcntl(monitor_fileno, fcntl.F_SETFL, fcntl.fcntl(monitor_fileno, fcntl.F_GETFL, 0) | os.O_NONBLOCK)

        io_loop = IOLoop.instance()

        fd_handler = partial(self._handle_udev_event, partial(monitor.poll, 0))
        io_loop.add_handler(monitor_fileno, fd_handler, IOLoop.READ | IOLoop.ERROR)

    def __init__(self, impl=None):
        self._impl = impl or _poll()
        if hasattr(self._impl, "fileno"):
            set_close_exec(self._impl.fileno())
        self._handlers = {}
        self._events = {}
        self._callbacks = []
        self._timeouts = []
        self._running = False
        self._stopped = False
        self._thread_ident = None
        self._blocking_signal_threshold = None

        # Create a pipe that we send bogus data to when we want to wake
        # the I/O loop when it is idle
        self._waker = Waker()
        self.add_handler(self._waker.fileno(), lambda fd, events: self._waker.consume(), self.READ)

def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128):
    """Creates listening sockets bound to the given port and address.

    Returns a list of socket objects (multiple sockets are returned if
    the given address maps to multiple IP addresses, which is most common
    for mixed IPv4 and IPv6 use).

    Address may be either an IP address or hostname.  If it's a hostname,
    the server will listen on all IP addresses associated with the
    name.  Address may be an empty string or None to listen on all
    available interfaces.  Family may be set to either socket.AF_INET
    or socket.AF_INET6 to restrict to ipv4 or ipv6 addresses, otherwise
    both will be used if available.

    The ``backlog`` argument has the same meaning as for
    ``socket.listen()``.
    """
    sockets = []
    if address == "":
        address = None
    flags = socket.AI_PASSIVE
    for res in set(socket.getaddrinfo(address, port, family, socket.SOCK_STREAM,
                                  0, flags)):
        af, socktype, proto, canonname, sockaddr = res
        sock = socket.socket(af, socktype, proto)
        set_close_exec(sock.fileno())
        if os.name != 'nt':
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        if af == socket.AF_INET6:
            # On linux, ipv6 sockets accept ipv4 too by default,
            # but this makes it impossible to bind to both
            # 0.0.0.0 in ipv4 and :: in ipv6.  On other systems,
            # separate sockets *must* be used to listen for both ipv4
            # and ipv6.  For consistency, always disable ipv4 on our
            # ipv6 sockets and use a separate ipv4 socket when needed.
            #
            # Python 2.x on windows doesn't have IPPROTO_IPV6.
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
        sock.setblocking(0)
        sock.bind(sockaddr)
        sock.listen(backlog)
        sockets.append(sock)
    return sockets

    def initialize(self, impl, time_func=None):
        super(PollIOLoop, self).initialize()

        # 使用的模型，使用的哪一个 select epoll kqueue
        # ubuntu系统使用的epoll
        self._impl = impl

        # 不明白
        if hasattr(self._impl, 'fileno'):
            set_close_exec(self._impl.fileno())

        # 设置获取时间的函数
        self.time_func = time_func or time.time

        # self._handlers[fd] = (obj, stack_context.wrap(handler))
        # fd这个时候是一个数字，一般的时候是一个fd的对象
        self._handlers = {}

        # 保存每一次循环所得到的fd和事件对
        self._events = {}

        # 所有的callback函数的集合
        self._callbacks = []
        self._callback_lock = threading.Lock()

        # 所有需要延时执行的函数的集合
        self._timeouts = []

        self._cancellations = 0
        self._running = False  # IOLoop是不是已经运行了
        self._stopped = False  # IOLoop是不是已经停止了，为什么有两个
        self._closing = False
        self._thread_ident = None
        self._blocking_signal_threshold = None

        # Create a pipe that we send bogus data to when we want to wake
        # the I/O loop when it is idle
        # 这个真心不懂，以后研究
        self._waker = Waker()

        # 初始化的时候添加self._waker的一个读得socket到IOLoop里面
        self.add_handler(self._waker.fileno(),
                         lambda fd, events: self._waker.consume(),
                         self.READ)