def _resolve(self, ioloop):
        """ Resolve host addr (domain)

        Args:
            ioloop (IOLoop): io_loop to use
        Returns:
            Tuple of address family and ip address
        """
        resolver = Resolver(io_loop=ioloop)
        addrinfo = yield resolver.resolve(self.addr, int(self.port), socket.AF_UNSPEC)
        raise gen.Return(addrinfo[0])

class TCPClient(object):
    """A non-blocking TCP connection factory.

    .. versionchanged:: 4.1
       The ``io_loop`` argument is deprecated.
    """
    def __init__(self, resolver=None, io_loop=None):
        self.io_loop = io_loop or IOLoop.current()
        if resolver is not None:
            self.resolver = resolver
            self._own_resolver = False
        else:
            self.resolver = Resolver(io_loop=io_loop)
            self._own_resolver = True

    def close(self):
        if self._own_resolver:
            self.resolver.close()

    @gen.coroutine
    def connect(self, host, port, af=socket.AF_UNSPEC, ssl_options=None,
                max_buffer_size=None):
        """Connect to the given host and port.

        Asynchronously returns an `.IOStream` (or `.SSLIOStream` if
        ``ssl_options`` is not None).
        """
        addrinfo = yield self.resolver.resolve(host, port, af)
        connector = _Connector(
            addrinfo, self.io_loop,
            functools.partial(self._create_stream, max_buffer_size))
        af, addr, stream = yield connector.start()
        # TODO: For better performance we could cache the (af, addr)
        # information here and re-use it on subsequent connections to
        # the same host. (http://tools.ietf.org/html/rfc6555#section-4.2)
        if ssl_options is not None:
            stream = yield stream.start_tls(False, ssl_options=ssl_options,
                                            server_hostname=host)
        raise gen.Return(stream)

    def _create_stream(self, max_buffer_size, af, addr):
        # Always connect in plaintext; we'll convert to ssl if necessary
        # after one connection has completed.
        try:
            stream = IOStream(socket.socket(af),
                            io_loop=self.io_loop,
                            max_buffer_size=max_buffer_size)
        except socket.error as e:
            fu = Future()
            fu.set_exception(e)
            return fu
        else:
            return stream.connect(addr)

class TCPClient(object):
    """A non-blocking TCP connection factory.
    """
    def __init__(self, resolver=None, io_loop=None):
        self.io_loop = io_loop or IOLoop.current()
        if resolver is not None:
            self.resolver = resolver
            self._own_resolver = False
        else:
            self.resolver = Resolver(io_loop=io_loop)
            self._own_resolver = True

    def close(self):
        if self._own_resolver:
            self.resolver.close()

    @gen.coroutine
    def connect(self, host, port, af=socket.AF_UNSPEC, ssl_options=None,
                max_buffer_size=None):
        """Connect to the given host and port.

        Asynchronously returns an `.IOStream` (or `.SSLIOStream` if
        ``ssl_options`` is not None).
        """
        addrinfo = yield self.resolver.resolve(host, port, af)
        connector = _Connector(
            addrinfo, self.io_loop,
            # 这个涉及到三个函数的调用，不说了，多是泪
            functools.partial(self._create_stream,
                              host, ssl_options, max_buffer_size))
        af, addr, stream = yield connector.start()
        # TODO: For better performance we could cache the (af, addr)
        # information here and re-use it on sbusequent connections to
        # the same host. (http://tools.ietf.org/html/rfc6555#section-4.2)
        raise gen.Return(stream)  # 这里gen会捕获这个错误

    def _create_stream(self, host, ssl_options, max_buffer_size, af, addr):
        # TODO: we should connect in plaintext mode and start the
        # ssl handshake only after stopping the _Connector.
        if ssl_options is None:
            stream = IOStream(socket.socket(af),
                              io_loop=self.io_loop,
                              max_buffer_size=max_buffer_size)
        else:
            stream = SSLIOStream(socket.socket(af),
                                 io_loop=self.io_loop,
                                 ssl_options=ssl_options,
                                 max_buffer_size=max_buffer_size)
        return stream.connect(addr, server_hostname=host)

 def __init__(self, sock):
     self._iostream = IOStream(sock)
     self._resolver = Resolver()
     self._readtimeout = 0
     self._connecttimeout = 0
     self._rbuffer = StringIO(b'')
     self._rbuffer_size = 0

 def __init__(self, resolver=None):
     if resolver is not None:
         self.resolver = resolver
         self._own_resolver = False
     else:
         self.resolver = Resolver()
         self._own_resolver = True

 def __init__(self, resolver=None, io_loop=None):
     self.io_loop = io_loop or IOLoop.current()
     if resolver is not None:
         self.resolver = resolver
         self._own_resolver = False
     else:
         self.resolver = Resolver(io_loop=io_loop)
         self._own_resolver = True

    def __init__(self, io_loop, request):
        self.connect_future = TracebackFuture()
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))

        scheme, sep, rest = request.url.partition(':')
        scheme = {'ws': 'http', 'wss': 'https'}[scheme]
        request.url = scheme + sep + rest
        request.headers.update({
            'Upgrade': 'websocket',
            'Connection': 'Upgrade',
            'Sec-WebSocket-Key': self.key,
            'Sec-WebSocket-Version': '13',
        })

        self.resolver = Resolver(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response,
            104857600, self.resolver)

class URLString(str):
    ''' A str wrapper, has more supports of URL '''
    __slots__ = ('parsed', 'solver')

    def __new__(cls, s):
        ''' New hook '''
        return str.__new__(cls, s)

    def __init__(self, s):
        super().__init__()
        self.parsed = urlparse(self)
        self.solver = Resolver()

    @classmethod
    def config_solver(cls, solver_type='tornado.netutil.BlockingResolver'):
        Resolver.configure(solver_type)

    @property
    def resolve(self) -> list:
        ''' DNS resolve '''
        return self.solver.resolve(self.parsed.netloc, port=80).result()

    def HEAD(self, **kwargs) -> bytes:
        ''' http HEAD method '''
        assert self.parsed.scheme
        return requests.head(self, **kwargs).content

    def GET(self, **kwargs) -> bytes:
        ''' HTTP GET method '''
        assert self.parsed.scheme
        return requests.get(self, **kwargs).content

    def POST(self, data=dict(), **kwargs) -> bytes:
        ''' HTTP POST method '''
        assert self.parsed.scheme
        return requests.post(self, data=data, **kwargs).content

    def __getattr__(self, attr):
        ''' Get attributes support '''
        return self.parsed.__getattribute__(attr)

    def __init__(self, io_loop, request):
        self.connect_future = TracebackFuture()
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))

        scheme, sep, rest = request.url.partition(":")
        scheme = {"ws": "http", "wss": "https"}[scheme]
        request.url = scheme + sep + rest
        request.headers.update(
            {
                "Upgrade": "websocket",
                "Connection": "Upgrade",
                "Sec-WebSocket-Key": self.key,
                "Sec-WebSocket-Version": "13",
            }
        )

        self.resolver = Resolver(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response, 104857600, self.resolver
        )

        #回调函数
        self.finish(result)

    @async_execute
    def dosomething(self,a,b,callback=None):
        #耗时操作
        result='return'
        return result
"""


import functools
from tornado.netutil import Resolver, ThreadedResolver
from tornado.ioloop import IOLoop

Resolver.configure('tornado.netutil.ThreadedResolver', num_threads=10)

def async_execute(fn):
    """
    新版tornado已有这个函数的实现，但是需要在class中绑定self.ioloop和self.executor,不太方便，这个版本消除了这个问题
    """
    thread_resolver = ThreadedResolver()

    @functools.wraps(fn)
    def wrapper(self, *args, **kwargs):
        callback = kwargs.pop("callback", None)
        future = thread_resolver.executor.submit(fn, self, *args, **kwargs)
        if callback:
            IOLoop.current().add_future(future,
                                    lambda future: callback(future.result()))
        return future

class TCPClient(object):
    """A non-blocking TCP connection factory.

    .. versionchanged:: 4.1
       The ``io_loop`` argument is deprecated.
    """
    def __init__(self, resolver=None, io_loop=None):
        self.io_loop = io_loop or IOLoop.current()
        if resolver is not None:
            self.resolver = resolver
            self._own_resolver = False
        else:
            self.resolver = Resolver(io_loop=io_loop)
            self._own_resolver = True

    def close(self):
        if self._own_resolver:
            self.resolver.close()

    @gen.coroutine
    def connect(self, host, port, af=socket.AF_UNSPEC, ssl_options=None,
                max_buffer_size=None, source_ip=None, source_port=None):
        """Connect to the given host and port.

        Asynchronously returns an `.IOStream` (or `.SSLIOStream` if
        ``ssl_options`` is not None).

        Using the ``source_ip`` kwarg, one can specify the source
        IP address to use when establishing the connection.
        In case the user needs to resolve and
        use a specific interface, it has to be handled outside
        of Tornado as this depends very much on the platform.

        Similarly, when the user requires a certain source port, it can
        be specified using the ``source_port`` arg.
        """
        addrinfo = yield self.resolver.resolve(host, port, af)
        connector = _Connector(
            addrinfo, self.io_loop,
            functools.partial(self._create_stream, max_buffer_size,
                              source_ip=source_ip, source_port=source_port)
        )
        af, addr, stream = yield connector.start()
        # TODO: For better performance we could cache the (af, addr)
        # information here and re-use it on subsequent connections to
        # the same host. (http://tools.ietf.org/html/rfc6555#section-4.2)
        if ssl_options is not None:
            stream = yield stream.start_tls(False, ssl_options=ssl_options,
                                            server_hostname=host)
        raise gen.Return(stream)

    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)
        if source_port_bind or source_ip_bind:
            # If the user requires binding also to a specific IP/port.
            socket_obj.bind((source_ip_bind, source_port_bind))
            # Fail loudly if unable to use the IP/port.
        try:
            stream = IOStream(socket_obj,
                              io_loop=self.io_loop,
                              max_buffer_size=max_buffer_size)
        except socket.error as e:
            fu = Future()
            fu.set_exception(e)
            return fu
        else:
            return stream.connect(addr)

class TCPClient(object):
    """A non-blocking TCP connection factory.

    .. versionchanged:: 5.0
       The ``io_loop`` argument (deprecated since version 4.1) has been removed.
    """
    def __init__(self, resolver=None):
        if resolver is not None:
            self.resolver = resolver
            self._own_resolver = False
        else:
            self.resolver = Resolver()
            self._own_resolver = True

    def close(self):
        if self._own_resolver:
            self.resolver.close()

    @gen.coroutine
    def connect(self, host, port, af=socket.AF_UNSPEC, ssl_options=None,
                max_buffer_size=None, source_ip=None, source_port=None,
                timeout=None):
        """Connect to the given host and port.

        Asynchronously returns an `.IOStream` (or `.SSLIOStream` if
        ``ssl_options`` is not None).

        Using the ``source_ip`` kwarg, one can specify the source
        IP address to use when establishing the connection.
        In case the user needs to resolve and
        use a specific interface, it has to be handled outside
        of Tornado as this depends very much on the platform.

        Raises `TimeoutError` if the input future does not complete before
        ``timeout``, which may be specified in any form allowed by
        `.IOLoop.add_timeout` (i.e. a `datetime.timedelta` or an absolute time
        relative to `.IOLoop.time`)

        Similarly, when the user requires a certain source port, it can
        be specified using the ``source_port`` arg.

        .. versionchanged:: 4.5
           Added the ``source_ip`` and ``source_port`` arguments.
        """
        if timeout is not None:
            if isinstance(timeout, numbers.Real):
                timeout = IOLoop.current().time() + timeout
            elif isinstance(timeout, datetime.timedelta):
                timeout = IOLoop.current().time() + timedelta_to_seconds(timeout)
            else:
                raise TypeError("Unsupported timeout %r" % timeout)
        if timeout is not None:
            addrinfo = yield gen.with_timeout(
                timeout, self.resolver.resolve(host, port, af))
        else:
            addrinfo = yield self.resolver.resolve(host, port, af)
        connector = _Connector(
            addrinfo,
            functools.partial(self._create_stream, max_buffer_size,
                              source_ip=source_ip, source_port=source_port)
        )
        af, addr, stream = yield connector.start(connect_timeout=timeout)
        # TODO: For better performance we could cache the (af, addr)
        # information here and re-use it on subsequent connections to
        # the same host. (http://tools.ietf.org/html/rfc6555#section-4.2)
        if ssl_options is not None:
            if timeout is not None:
                stream = yield gen.with_timeout(timeout, stream.start_tls(
                    False, ssl_options=ssl_options, server_hostname=host))
            else:
                stream = yield stream.start_tls(False, ssl_options=ssl_options,
                                                server_hostname=host)
        raise gen.Return(stream)

    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)
#.........这里部分代码省略.........

    @async_execute
    def dosomething(self,a,b,callback=None):
        #耗时操作
        result='return'
        return result
"""

import functools
from tornado.netutil import Resolver, ThreadedResolver
from tornado.ioloop import IOLoop
from ..settings_manager import settings
from multiprocessing import cpu_count


#设置接口的实现类的类型，同时设置设置实现类的参数
Resolver.configure('tornado.netutil.ThreadedResolver',
                   num_threads=settings.THREADS_NUM if 'THREADS_NUM' in settings else cpu_count())


def async_execute(fn):
    """
    新版tornado已有这个函数的实现，但是需要在class中绑定self.ioloop和self.executor,不太方便，这个版本消除了这个问题
    
    其实在 concurrent.py 中 run_on_executor 装饰器方法实现了类似的功能。

    """
    thread_resolver = ThreadedResolver()

    @functools.wraps(fn)
    def wrapper(self, *args, **kwargs):
        callback = kwargs.pop("callback", None)
        future = thread_resolver.executor.submit(fn, self, *args, **kwargs)

class WebSocketClientConnection(simple_httpclient._HTTPConnection):
    """WebSocket client connection."""

    def __init__(self, io_loop, request):
        self.connect_future = TracebackFuture()
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))

        scheme, sep, rest = request.url.partition(":")
        scheme = {"ws": "http", "wss": "https"}[scheme]
        request.url = scheme + sep + rest
        request.headers.update(
            {
                "Upgrade": "websocket",
                "Connection": "Upgrade",
                "Sec-WebSocket-Key": self.key,
                "Sec-WebSocket-Version": "13",
            }
        )

        self.resolver = Resolver(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response, 104857600, self.resolver
        )

    def _on_close(self):
        self.on_message(None)
        self.resolver.close()

    def _on_http_response(self, response):
        if not self.connect_future.done():
            if response.error:
                self.connect_future.set_exception(response.error)
            else:
                self.connect_future.set_exception(WebSocketError("Non-websocket response"))

    def _handle_1xx(self, code):
        assert code == 101
        assert self.headers["Upgrade"].lower() == "websocket"
        assert self.headers["Connection"].lower() == "upgrade"
        accept = WebSocketProtocol13.compute_accept_value(self.key)
        assert self.headers["Sec-Websocket-Accept"] == accept

        self.protocol = WebSocketProtocol13(self, mask_outgoing=True)
        self.protocol._receive_frame()

        if self._timeout is not None:
            self.io_loop.remove_timeout(self._timeout)
            self._timeout = None

        self.connect_future.set_result(self)

    def write_message(self, message, binary=False):
        """Sends a message to the WebSocket server."""
        self.protocol.write_message(message, binary)

    def read_message(self, callback=None):
        """Reads a message from the WebSocket server.

        Returns a future whose result is the message, or None
        if the connection is closed.  If a callback argument
        is given it will be called with the future when it is
        ready.
        """
        assert self.read_future is None
        future = TracebackFuture()
        if self.read_queue:
            future.set_result(self.read_queue.popleft())
        else:
            self.read_future = future
        if callback is not None:
            self.io_loop.add_future(future, callback)
        return future

    def on_message(self, message):
        if self.read_future is not None:
            self.read_future.set_result(message)
            self.read_future = None
        else:
            self.read_queue.append(message)

    def on_pong(self, data):
        pass

#!/usr/bin/env python
# encoding: utf-8
from copy import copy
from tornado.web import Cookie
from tornado.gen import coroutine, Return
from tornado.httpclient import AsyncHTTPClient, HTTPRequest, HTTPError, HTTPResponse
from tornado.httputil import HTTPHeaders
from tornado.ioloop import IOLoop
from . import PY2

try:
    import pycares
    from tornado.netutil import Resolver
    from tornado.platform.caresresolver import CaresResolver

    Resolver.configure(CaresResolver)
except ImportError:
    pass


if PY2:
    b = unicode
    iteritems = lambda x: x.iteritems()
else:
    b = str
    iteritems = lambda x: x.items()


try:
    import ujson as json
except ImportError:

class AsyncSocket(object):
    def __init__(self, sock):
        self._iostream = IOStream(sock)
        self._resolver = Resolver()
        self._readtimeout = 0
        self._connecttimeout = 0
   
    def set_readtimeout(self, timeout):
        self._readtimeout = timeout

    def set_connecttimeout(self, timeout):
        self._connecttimeout = timeout

    @synclize
    def connect(self, address):
        host, port = address
        timer = None
        try:
            if self._connecttimeout:
                timer = Timeout(self._connecttimeout)
                timer.start()
            resolved_addrs = yield self._resolver.resolve(host, port, family=socket.AF_INET)
            for addr in resolved_addrs:
                family, host_port = addr
                yield self._iostream.connect(host_port)
                break
        except TimeoutException:
            self.close()
            raise
        finally:
            if timer:
                timer.cancel()
    #@synclize
    def sendall(self, buff):
        self._iostream.write(buff)

    @synclize
    def read(self, nbytes, partial=False):
        timer = None
        try:
            if self._readtimeout:
                timer = Timeout(self._readtimeout)
                timer.start()
            buff = yield self._iostream.read_bytes(nbytes, partial=partial)
            raise Return(buff)
        except TimeoutException:
            self.close()
            raise
        finally:
            if timer:
                timer.cancel()

    def recv(self, nbytes):
        return self.read(nbytes, partial=True)

    @synclize
    def readline(self, max_bytes=-1):
        timer = None
        if self._readtimeout:
            timer = Timeout(self._readtimeout)
            timer.start()
        try:
            if max_bytes > 0:
                buff = yield self._iostream.read_until('\n', max_bytes=max_bytes)
            else:
                buff = yield self._iostream.read_until('\n')
            raise Return(buff)
        except TimeoutException:
            self.close()
            raise
        finally:
            if timer:
                timer.cancel()

    def close(self):
        self._iostream.close()

    def set_nodelay(self, flag):
        self._iostream.set_nodelay(flag)

    def settimeout(self, timeout):
        pass

    def shutdown(self, direction):
        if self._iostream.fileno():
            self._iostream.fileno().shutdown(direction)

    def recv_into(self, buff):
        expected_rbytes = len(buff)
        data = self.read(expected_rbytes, True)
        srcarray = bytearray(data)
        nbytes = len(srcarray)
        buff[0:nbytes] = srcarray
        return nbytes

    def makefile(self, mode, other):
        return self

def set_resolver(resolver):
    Resolver.configure(resolver)

 def __init__(self, io_loop):
     self.io_loop = io_loop
     # Default blocking resolver calling socket.getaddrinfo
     self.resolver = Resolver(io_loop=io_loop)
     self._own_resolver = True

class WebSocketClientConnection(simple_httpclient._HTTPConnection):
    """WebSocket client connection.

    This class should not be instantiated directly; use the
    `websocket_connect` function instead.
    """
    def __init__(self, io_loop, request):
        self.connect_future = TracebackFuture()
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))

        scheme, sep, rest = request.url.partition(':')
        scheme = {'ws': 'http', 'wss': 'https'}[scheme]
        request.url = scheme + sep + rest
        request.headers.update({
            'Upgrade': 'websocket',
            'Connection': 'Upgrade',
            'Sec-WebSocket-Key': self.key,
            'Sec-WebSocket-Version': '13',
        })

        self.resolver = Resolver(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response,
            104857600, self.resolver)

    def close(self):
        """Closes the websocket connection.

        .. versionadded:: 3.2
        """
        if self.protocol is not None:
            self.protocol.close()
            self.protocol = None

    def _on_close(self):
        self.on_message(None)
        self.resolver.close()
        super(WebSocketClientConnection, self)._on_close()

    def _on_http_response(self, response):
        if not self.connect_future.done():
            if response.error:
                self.connect_future.set_exception(response.error)
            else:
                self.connect_future.set_exception(WebSocketError(
                    "Non-websocket response"))

    def _handle_1xx(self, code):
        assert code == 101
        assert self.headers['Upgrade'].lower() == 'websocket'
        assert self.headers['Connection'].lower() == 'upgrade'
        accept = WebSocketProtocol13.compute_accept_value(self.key)
        assert self.headers['Sec-Websocket-Accept'] == accept

        self.protocol = WebSocketProtocol13(self, mask_outgoing=True)
        self.protocol._receive_frame()

        if self._timeout is not None:
            self.io_loop.remove_timeout(self._timeout)
            self._timeout = None

        self.connect_future.set_result(self)

    def write_message(self, message, binary=False):
        """Sends a message to the WebSocket server."""
        self.protocol.write_message(message, binary)

    def read_message(self, callback=None):
        """Reads a message from the WebSocket server.

        Returns a future whose result is the message, or None
        if the connection is closed.  If a callback argument
        is given it will be called with the future when it is
        ready.
        """
        assert self.read_future is None
        future = TracebackFuture()
        if self.read_queue:
            future.set_result(self.read_queue.popleft())
        else:
            self.read_future = future
        if callback is not None:
            self.io_loop.add_future(future, callback)
        return future

    def on_message(self, message):
        if self.read_future is not None:
            self.read_future.set_result(message)
            self.read_future = None
        else:
            self.read_queue.append(message)

    def on_pong(self, data):
        pass

 def _config_resolver(cls, num_threads=10):
     from tornado.netutil import Resolver
     Resolver.configure(
             'tornado.netutil.ThreadedResolver',
             num_threads=num_threads)
     cls._resolver_configured = True