def accept_handler(fd, events):
     # 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 xrange(_DEFAULT_BACKLOG):
         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
         callback(connection, address)

    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 accept_handler(fd, events):
     while True:
         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
         callback(connection, address)

def bind_sockets(port, address=None, family=socket.AF_UNSPEC,
                 backlog=_DEFAULT_BACKLOG, flags=None):
    sockets = []
    if address == "":
        address = None
    if flags is None:
        flags = socket.AI_PASSIVE
    bound_port = None
    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 errno_from_exception(e) == errno.EAFNOSUPPORT:
                continue
            raise
        set_close_exec(sock.fileno())
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        if af == socket.AF_INET6:
            if hasattr(socket, "IPPROTO_IPV6"):
                sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)

        host, requested_port = sockaddr[:2]
        if requested_port == 0 and bound_port is not None:
            sockaddr = tuple([host, bound_port] + list(sockaddr[2:]))

        sock.setblocking(0)
        sock.bind(sockaddr)
        bound_port = sock.getsockname()[1]
        sock.listen(backlog)
        sockets.append(sock)
    return sockets

    def connect(self):
        """Connects the object to the host:port.

        Returns:
            Future: a Future object with True as result if the connection
                process was ok.
        """
        if self.is_connected() or self.is_connecting():
            raise tornado.gen.Return(True)
        if self.unix_domain_socket is None:
            self.__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            if self.tcp_nodelay:
                self.__socket.setsockopt(socket.IPPROTO_TCP,
                                         socket.TCP_NODELAY, 1)
        else:
            if not os.path.exists(self.unix_domain_socket):
                LOG.warning("can't connect to %s, file does not exist",
                            self.unix_domain_socket)
                raise tornado.gen.Return(False)
            self.__socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.__socket.setblocking(0)
        self.__periodic_callback.start()
        try:
            LOG.debug("connecting to %s...", self._redis_server())
            self._state.set_connecting()
            if self.unix_domain_socket is None:
                self.__socket.connect((self.host, self.port))
            else:
                self.__socket.connect(self.unix_domain_socket)
        except socket.error as e:
            if (errno_from_exception(e) not in _ERRNO_INPROGRESS and
                    errno_from_exception(e) not in _ERRNO_WOULDBLOCK):
                self.disconnect()
                LOG.warning("can't connect to %s", self._redis_server())
                raise tornado.gen.Return(False)
            self.__socket_fileno = self.__socket.fileno()
            self._register_or_update_event_handler()
            yield self._state.get_changed_state_future()
            if not self.is_connected():
                LOG.warning("can't connect to %s", self._redis_server())
                raise tornado.gen.Return(False)
        else:
            LOG.debug("connected to %s", self._redis_server())
            self.__socket_fileno = self.__socket.fileno()
            self._state.set_connected()
            self._register_or_update_event_handler()
        raise tornado.gen.Return(True)

 def read_from_fd(self):
     try:
         chunk = os.read(self.fd, self.read_chunk_size)
     except (IOError, OSError) as e:
         if errno_from_exception(e) in _ERRNO_WOULDBLOCK:
             return None
         elif errno_from_exception(e) == errno.EBADF:
             # If the writing half of a pipe is closed, select will
             # report it as readable but reads will fail with EBADF.
             self.close(exc_info=True)
             return None
         else:
             raise
     if not chunk:
         self.close()
         return None
     return chunk

 def connect(self, address, callback=None, server_hostname=None):
     self._connecting = True
     if callback is not None:
         self._connect_callback = stack_context.wrap(callback)
         future = None
     else:
         future = self._connect_future = TracebackFuture()
     try:
         self.socket.connect(address)
     except socket.error as e:
         if (errno_from_exception(e) not in _ERRNO_INPROGRESS and
                 errno_from_exception(e) not in _ERRNO_WOULDBLOCK):
             if future is None:
                 print("Connect error on fd %s: %s" % (self.socket.fileno(), e))
             self.close(exc_info=True)
             return future
     self._add_io_state(self.io_loop.WRITE)
     return future

    def connect(self, address, callback=None, server_hostname=None):
        """Connects the socket to a remote address without blocking.

        May only be called if the socket passed to the constructor was
        not previously connected.  The address parameter is in the
        same format as for `socket.connect <socket.socket.connect>`,
        i.e. a ``(host, port)`` tuple.  If ``callback`` is specified,
        it will be called when the connection is completed.

        If specified, the ``server_hostname`` parameter will be used
        in SSL connections for certificate validation (if requested in
        the ``ssl_options``) and SNI (if supported; requires
        Python 3.2+).

        Note that it is safe to call `IOStream.write
        <BaseIOStream.write>` while the connection is pending, in
        which case the data will be written as soon as the connection
        is ready.  Calling `IOStream` read methods before the socket is
        connected works on some platforms but is non-portable.
        """
        self._connecting = True
        try:
            self.socket.connect(address)
        except socket.error as e:
            # In non-blocking mode we expect connect() to raise an
            # exception with EINPROGRESS or EWOULDBLOCK.
            #
            # On freebsd, other errors such as ECONNREFUSED may be
            # returned immediately when attempting to connect to
            # localhost, so handle them the same way as an error
            # reported later in _handle_connect.
            if (errno_from_exception(e) != errno.EINPROGRESS and
                    errno_from_exception(e) not in _ERRNO_WOULDBLOCK):
                gen_log.warning("Connect error on fd %s: %s",
                                self.socket.fileno(), e)
                self.close(exc_info=True)
                return
        if callback is not None:
            self._connect_callback = stack_context.wrap(callback)
            future = None
        else:
            future = self._connect_future = TracebackFuture()
        self._add_io_state(self.io_loop.WRITE)
        return future

def is_socket_abnormal(sock):
    try:
        chunk = sock.recv(1, socket.MSG_PEEK)
    except (socket.error, IOError, OSError) as e:
        if e.args[0] not in _ERRNO_WOULDBLOCK and errno_from_exception(e) != errno.EINTR:
            return True
    else:
        if not chunk:
            return True
    return False

    def __init__(self):
        from .auto import set_close_exec
        # Based on Zope select_trigger.py:
        # https://github.com/zopefoundation/Zope/blob/master/src/ZServer/medusa/thread/select_trigger.py

        self.writer = socket.socket()
        set_close_exec(self.writer.fileno())
        # Disable buffering -- pulling the trigger sends 1 byte,
        # and we want that sent immediately, to wake up ASAP.
        self.writer.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

        count = 0
        while 1:
            count += 1
            # Bind to a local port; for efficiency, let the OS pick
            # a free port for us.
            # Unfortunately, stress tests showed that we may not
            # be able to connect to that port ("Address already in
            # use") despite that the OS picked it.  This appears
            # to be a race bug in the Windows socket implementation.
            # So we loop until a connect() succeeds (almost always
            # on the first try).  See the long thread at
            # http://mail.zope.org/pipermail/zope/2005-July/160433.html
            # for hideous details.
            a = socket.socket()
            set_close_exec(a.fileno())
            a.bind(("127.0.0.1", 0))
            a.listen(1)
            connect_address = a.getsockname()  # assigned (host, port) pair
            try:
                self.writer.connect(connect_address)
                break    # success
            except socket.error as detail:
                if (not hasattr(errno, 'WSAEADDRINUSE') or
                        errno_from_exception(detail) != errno.WSAEADDRINUSE):
                    # "Address already in use" is the only error
                    # I've seen on two WinXP Pro SP2 boxes, under
                    # Pythons 2.3.5 and 2.4.1.
                    raise
                # (10048, 'Address already in use')
                # assert count <= 2 # never triggered in Tim's tests
                if count >= 10:  # I've never seen it go above 2
                    a.close()
                    self.writer.close()
                    raise socket.error("Cannot bind trigger!")
                # Close `a` and try again.  Note:  I originally put a short
                # sleep() here, but it didn't appear to help or hurt.
                a.close()

        self.reader, addr = a.accept()
        set_close_exec(self.reader.fileno())
        self.reader.setblocking(0)
        self.writer.setblocking(0)
        a.close()
        self.reader_fd = self.reader.fileno()

 def _try_cleanup_process(cls, pid: int) -> None:
     try:
         ret_pid, status = os.waitpid(pid, os.WNOHANG)
     except OSError as e:
         if errno_from_exception(e) == errno.ECHILD:
             return
     if ret_pid == 0:
         return
     assert ret_pid == pid
     subproc = cls._waiting.pop(pid)
     subproc.io_loop.add_callback_from_signal(subproc._set_returncode, status)

 def receive(self):
     while True:
         try:
             info = self.sock.recv(4)
             if not info:
                 self.sock.shutdown(socket.SHUT_WR)
                 self._connected = False
                 logging.info('Server closed!')
         except socket.error as err:
             if errno_from_exception(err) in (errno.EWOULDBLOCK, errno.EAGAIN):
                 continue
             raise

    def _handle_connection(self, connection: socket.socket, address: Any) -> None:
        if self.ssl_options is not None:
            assert ssl, "Python 2.6+ and OpenSSL required for SSL"
            try:
                connection = ssl_wrap_socket(
                    connection,
                    self.ssl_options,
                    server_side=True,
                    do_handshake_on_connect=False,
                )
            except ssl.SSLError as err:
                if err.args[0] == ssl.SSL_ERROR_EOF:
                    return connection.close()
                else:
                    raise
            except socket.error as err:
                # If the connection is closed immediately after it is created
                # (as in a port scan), we can get one of several errors.
                # wrap_socket makes an internal call to getpeername,
                # which may return either EINVAL (Mac OS X) or ENOTCONN
                # (Linux).  If it returns ENOTCONN, this error is
                # silently swallowed by the ssl module, so we need to
                # catch another error later on (AttributeError in
                # SSLIOStream._do_ssl_handshake).
                # To test this behavior, try nmap with the -sT flag.
                # https://github.com/tornadoweb/tornado/pull/750
                if errno_from_exception(err) in (errno.ECONNABORTED, errno.EINVAL):
                    return connection.close()
                else:
                    raise
        try:
            if self.ssl_options is not None:
                stream = SSLIOStream(
                    connection,
                    max_buffer_size=self.max_buffer_size,
                    read_chunk_size=self.read_chunk_size,
                )  # type: IOStream
            else:
                stream = IOStream(
                    connection,
                    max_buffer_size=self.max_buffer_size,
                    read_chunk_size=self.read_chunk_size,
                )

            future = self.handle_stream(stream, address)
            if future is not None:
                IOLoop.current().add_future(
                    gen.convert_yielded(future), lambda f: f.result()
                )
        except Exception:
            app_log.error("Error in connection callback", exc_info=True)

 def msg_loop(self):
     logging.info('Have connected to server %s:%s' % (host, self.port))
     msg = 'H' * size
     self._connected = True
     while self._connected:
         try:
             self.sock.send(msg)
         except socket.error as err:
             eno = errno_from_exception(err)
             if eno in (errno.EWOULDBLOCK, errno.EAGAIN):
                 continue
             elif eno in (errno.EPROTOTYPE, errno.EPIPE):
                 break
             raise

 def udp_handler(fd, events):
     while True:
         try:
             data, addr = sock.recvfrom(4096)
             if data:
                 # ac data arrived, deal with
                 pass
         except socket.error as e:
             if errno_from_exception(e) in _ERRNO_WOULDBLOCK:
                 # _ERRNO_WOULDBLOCK indicate we have accepted every
                 # connection that is avaiable
                 return
             import traceback
             traceback.print_exc(file=sys.stdout)
         except: 
             import traceback
             traceback.print_exc(file=sys.stdout)

 def _handle_connection(self, connection, address):
     # 如果可以的话，启用ssl
     # address = ('127.0.0.1', 63959)
     # connection : socket object
     if self.ssl_options is not None:
         assert ssl, "Python 2.6+ and OpenSSL required for SSL"
         try:
             connection = ssl_wrap_socket(connection,
                                          self.ssl_options,
                                          server_side=True,
                                          do_handshake_on_connect=False)
         except ssl.SSLError as err:
             if err.args[0] == ssl.SSL_ERROR_EOF:
                 return connection.close()
             else:
                 raise
         except socket.error as err:
             # If the connection is closed immediately after it is created
             # (as in a port scan), we can get one of several errors.
             # wrap_socket makes an internal call to getpeername,
             # which may return either EINVAL (Mac OS X) or ENOTCONN
             # (Linux).  If it returns ENOTCONN, this error is
             # silently swallowed by the ssl module, so we need to
             # catch another error later on (AttributeError in
             # SSLIOStream._do_ssl_handshake).
             # To test this behavior, try nmap with the -sT flag.
             # https://github.com/tornadoweb/tornado/pull/750
             if errno_from_exception(err) in (errno.ECONNABORTED, errno.EINVAL):
                 return connection.close()
             else:
                 raise
     try:
         if self.ssl_options is not None:
             stream = SSLIOStream(connection, io_loop=self.io_loop,
                                  max_buffer_size=self.max_buffer_size,
                                  read_chunk_size=self.read_chunk_size)
         else:
             # 根据connection获得一个IOStream的对象，便于数据的读取
             stream = IOStream(connection, io_loop=self.io_loop,
                               max_buffer_size=self.max_buffer_size,
                               read_chunk_size=self.read_chunk_size)
         # 开始处理进来的数据连接
         self.handle_stream(stream, address)
     except Exception:
         app_log.error("Error in connection callback", exc_info=True)

    def on_read(self):
        logging.debug('worker {} on read'.format(self.id))
        try:
            data = self.chan.recv(BUF_SIZE)
        except (OSError, IOError) as e:
            logging.error(e)
            if errno_from_exception(e) in _ERRNO_CONNRESET:
                self.close(reason='chan error on reading')
        else:
            logging.debug('{!r} from {}:{}'.format(data, *self.dst_addr))
            if not data:
                self.close(reason='chan closed')
                return

            logging.debug('{!r} to {}:{}'.format(data, *self.handler.src_addr))
            try:
                self.handler.write_message(data, binary=True)
            except tornado.websocket.WebSocketClosedError:
                self.close(reason='websocket closed')

    def on_read(self):
        logging.debug('worker {} on read'.format(self.id))
        try:
            data = self.chan.recv(BUF_SIZE)
        except (OSError, IOError) as e:
            logging.error(e)
            if errno_from_exception(e) in _ERRNO_CONNRESET:
                self.close()
        else:
            logging.debug('"{}" from {}'.format(data, self.dst_addr))
            if not data:
                self.close()
                return

            logging.debug('"{}" to {}'.format(data, self.handler.src_addr))
            try:
                self.handler.write_message(data)
            except tornado.websocket.WebSocketClosedError:
                self.close()

    def _read_packet(self):
        """Reads from the socket and appends the result to the read buffer.

        Returns the number of bytes read.  Returns 0 if there is nothing
        to read (i.e. the read returns EWOULDBLOCK or equivalent).  On
        error closes the socket and raises an exception.
        """
        while True:
            try:
                chunk, address = self.read_from_fd()
            except (socket.error, IOError, OSError) as e:
                if errno_from_exception(e) == errno.EINTR:
                    continue
                self.close(exc_info=True)
                raise
            break
        if chunk is None:
            return None
        self.add_unread_packet(chunk, address)
        return chunk, address

    def on_write(self):
        logging.debug('worker {} on write'.format(self.id))
        if not self.data_to_dst:
            return

        data = ''.join(self.data_to_dst)
        logging.debug('{!r} to {}:{}'.format(data, *self.dst_addr))

        try:
            sent = self.chan.send(data)
        except (OSError, IOError) as e:
            logging.error(e)
            if errno_from_exception(e) in _ERRNO_CONNRESET:
                self.close(reason='chan error on writing')
            else:
                self.update_handler(IOLoop.WRITE)
        else:
            self.data_to_dst = []
            data = data[sent:]
            if data:
                self.data_to_dst.append(data)
                self.update_handler(IOLoop.WRITE)
            else:
                self.update_handler(IOLoop.READ)