Tornado源码剖析 （转）

       原文地址：    http://blog.csdn.net/goldlevi/article/details/7047726

       代码详解可参考：http://www.nowamagic.net/academy/detail/13321019

       本文大部分参考原文，因为代码版本不同做了少量修改。

1 Tornado来历

           Tornado是一个开源的网络服务器框架，该平台基于社交聚合网站FriendFeed的实时信息服务开发而来。2007年，4名谷歌前软件工程师一起创办了FriendFeed，旨在使用户能方便地跟踪好友在Facebook和Twitter等多个社交网站上的活动。结果两年后，Facebook宣布收购FriendFeed，这一交易的价格约为5000万美元。而此时，FriendFeed只有12名员工。据说这帮人后来又到了Google，搞出了现在的Google App Engine …… 

            Tornado由Python编写，跟其他主流的Web服务器框架不同是采用epoll非阻塞IO，响应快速，可处理数千并发连接，特别适用用于实时的Web服务。Tornado当前版本为3.2，官方网站为http://www.tornadoweb.org/，有兴趣的同学可以去尝试一下。

        本文代码使用版本为2.4。

2 Tornado简介

         Tornado主要包含了如下四部分内容。官方的帮助文档，实际上只是源码注释的集合。大家直接看源码就可以了。

• Core web framework
      	tornado.web —RequestHandler and Application classes
  	tornado.httpserver — Non-blocking HTTP server
  	tornado.template — Flexible output generation
  	tornado.escape — Escaping and string manipulation
  	tornado.locale — Internationalization support
  Asynchronous networking
  	tornado.ioloop — Main event loop
  	tornado.iostream — Convenient wrappers for non-blocking sockets
  	tornado.httpclient — Non-blocking HTTP client
  	tornado.netutil — Miscellaneous network utilities
  Integration with other services
  	tornado.auth — Third-party login with OpenID and OAuth
  	tornado.database — Simple MySQL client wrapper
  	tornado.platform.twisted — Run code written for Twisted on Tornado
  	tornado.websocket — Bidirectional communication to the browser
  	tornado.wsgi — Interoperability with other Python frameworks and servers
  Utilities
  	tornado.autoreload — Automatically detect code changes in development
  	tornado.gen — Simplify asynchronous code
  	tornado.httputil — Manipulate HTTP headers and URLs
  	tornado.options — Command-line parsing
  	tornado.process — Utilities for multiple processes
  	tornado.stack_context — Exception handling across asynchronous callbacks
  	tornado.testing — Unit testing support for asynchronous code

   

     今天主要和大家分享一下HTTP SERVER的相关内容。

 

2.1Tornado HTTP SERVER

     使用Tornado可以很方便地架构出各种类型的web服务器。我们现在从HTTP服务器入手，来看一下它的实现。下面这张图大家应该见得很多了，是所有web server的一般工作方式。

    

 

 

l  服务器端bind到一个端口，然后开始listen。

l  客户端connect上来以后，将请求发送给服务端。

l  服务端处理完成后返回给客户端。

这样，一个请求就处理结束了。不过，当需要处理成千上万的连接的时候，我们就会在这个基础上考虑更多的情况。这也就是大家熟悉的。一般大家会有如下一些选择：

l  一个线程服务多个客户端，使用非阻塞I/O和水平触发的就绪通知

l  一个线程服务多个客户端，使用非阻塞I/O和就绪改变时通知

l  一个服务线程服务多个客户端，使用异步I/O

l  一个服务线程服务一个客户端，使用阻塞I/O

l  把服务代码编译进内核

Tornado采用的就是：多进程 + 非阻塞 + epoll模型

下面这张图基本上就显示了Tornado与网络相关的所有内容了：

       

 

2.2 第一个HTTP server例子

   下面是一个hello world的代码示范。

 

import tornado.ioloop

import tornado.web

 

class MainHandler(tornado.web.RequestHandler):

    def get(self):

        self.write("Hello, world")


if __name__ == "__main__":


    //注册我们自己的回调函数，在读取完数据后，系统会根据url的匹配调用合适的RequestHandler

    app = tornado.web.Application(handlers=[

        (r"/", MainHandler), ])  

    //创建HTTPServer，将Application好好的保护起来。

    http_server = HTTPServer(app)

    //此时默认会将读取网卡的所有ip进行绑定，当然你可以指定要绑定的ip。

     http_server.bind(11111)

    //此时创建子进程，每个子进程会创建epoll，并且将绑定ip放入到epoll中，监听连接事件。

    http_server.start(3) 
    //开始死循环处理事件，先处理系统生成超时事件，而后处理网络事件。 
    IOLoop.instance().start()
 

 

 

 

2.3 模块分析

    我们接下来将逐个分析这部分代码。首先对Tornado有个全面的了解。Tornado服务器有4大核心模块：

(1) IOLoop

     从上面的代码可能看出，Tornado为了实现高并发和高性能, 使用了一个IOLoop来处理socket的读写事件, IOLoop基于epoll, 可以高效的响应网络事件. 这是Tornado高效的保证. 

(2) IOStream

     为了在处理请求的时候, 实现对socket的异步读写, Tornado实现了IOStream类, 用来处理socket的异步读写。

(3) HTTPConnection

    这个类用来处理http的请求，包括读取http请求头，读取post过来的数据，调用用户自定义的处理方法，以及把响应数据写给客户端socket。

    下面这幅图描述了tornado服务器的大体处理流程, 接下来我们将会详细分析每一步流程的实现。

    

 
 

(4) Application

这个类负责管理用户注册的RequestHandler类，在HTTPConnection解析完http后，Application的__call__放回会被回调，此时Application会根据http请求的url选择合适的handler来处理这个request，并发送响应数据。

 

 3 源码分析

 

3.1 bind和listen

     服务器的第一步就是bind。Httpserver.py的bind函数可以看到一个标准的服务器启动过程：

 

def bind(self, port, address=None, family=socket.AF_UNSPEC):
        if address == "":
            address = None
             // 查找网卡信息
        for res in socket.getaddrinfo(address, port, family, socket.SOCK_STREAM,
                                      0, socket.AI_PASSIVE | socket.AI_ADDRCONFIG):
            af, socktype, proto, canonname, sockaddr = res
            sock = socket.socket(af, socktype, proto)
            flags = fcntl.fcntl(sock.fileno(), fcntl.F_GETFD)
            //设置进程结束后，文件也关闭
flags |= fcntl.FD_CLOEXEC
            fcntl.fcntl(sock.fileno(), fcntl.F_SETFD, flags)
           //设置端口可重用
            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)
            sock.setblocking(0)
           // bind和listen
            sock.bind(sockaddr)
            sock.listen(128)
            self._sockets[sock.fileno()] = sock
           
            if self._started:
                self.io_loop.add_handler(sock.fileno(), self._handle_events,
                                         ioloop.IOLoop.READ)

  

for循环保证对每张网卡上的请求都得到监听。对于每个网卡，先建立socket，然后bind listen，最后将socket加入到io_loop，注册的事件是ioloop.IOLoop.READ，也就是读事件。程序中还添加了对ipv6的处理。一旦listen socket可读, 说明客户端请求到来, 然后调用_handle_events接受客户端的请求。接下来，看一下_handle_events是怎么处理的。

3.2 accept

 

接上一节，Httpserver.py的_handle_events函数实现了accept的过程。代码如下：

    def _handle_events(self, fd, events):
        while True:
            try:
                   connection, address = self._sockets[fd].accept()
            except socket.error, e:
                if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
                    return
                raise
            if self.ssl_options is not None:
                                   //这里有一段处理ssl的代码，比较长，省略
            try:
                stream = iostream.IOStream(connection, io_loop=self.io_loop)
                HTTPConnection(stream, address, self.request_callback,
                               self.no_keep_alive, self.xheaders)
            except:
                logging.error("Error in connection callback", exc_info=True)
 

 

accept方法返回客户端的socket, 以及客户端的地址。然后创建IOStream对象, 用来处理socket的异步读写. 这一步会调用ioloop.add_handler把client socket加入ioloop，再然后创建HTTPConnection, 处理用户的请求。接下来，我们看下iostream和httpconnection。

 

3.3 iostream

     为了实现对client socket的异步读写, 需要为client socket创建两个缓冲区: _read_buffer和_write_buffer，这样我们就不用直接读写socket，进而实现异步读写。这些操作都封装在IOStream类中。概括来说，IOStream对socket的读写做了一层封装，通过使用两个缓冲区，实现对socket的异步读写。

 

 def __init__(self, socket, io_loop=None, max_buffer_size=104857600,
                 read_chunk_size=4096):
        self.socket = socket
        self.socket.setblocking(False)
        self.io_loop = io_loop or ioloop.IOLoop.instance()
        self._read_buffer = collections.deque()
        self._write_buffer = collections.deque()
        self._state = self.io_loop.ERROR
        with stack_context.NullContext():
            self.io_loop.add_handler(
                self.socket.fileno(), self._handle_events, self._state)

 

      可以看到，初始化的时候建立了两个buffer，然后把自己的socket放到了io_loop。这样，当这个socket有读写的时候，就会回调到注册的事件self._handle_events里面了。_handle_events就很容易理解了，代码如下：

 def _handle_events(self, fd, events):
        if not self.socket:
            logging.warning("Got events for closed stream %d", fd)
            return
        try:
            if events & self.io_loop.READ:
                self._handle_read()
            if events & self.io_loop.WRITE:
                self._handle_write()
            if events & self.io_loop.ERROR:
                self.io_loop.add_callback(self.close)
                return
            state = self.io_loop.ERROR
            if self.reading():
                state |= self.io_loop.READ
            if self.writing():
                state |= self.io_loop.WRITE
            if state != self._state:
                self._state = state
                self.io_loop.update_handler(self.socket.fileno(), self._state)
        except:
            logging.error("Uncaught exception, closing connection.",
                          exc_info=True)
            self.close()
            raise

 

     其中 self._handle_read()读取完数据后，会调用Application类 (web.py) ， Application类存储着我们注册的处理函数，Application会根据request中的url找到合适的处理函数，Application处理函数如下：

     

def __call__(self, request):
        """Called by HTTPServer to execute the request."""
        … …
        //查找合适的handler来处理请求
        if not handlers:
            handler = RedirectHandler(
                self, request, url="http://" + self.default_host + "/")
        else:
            for spec in handlers:
                match = spec.regex.match(request.path)
                if match:
                    handler = spec.handler_class(self, request, **spec.kwargs)
            …
        handler._execute(transforms, *args, **kwargs)
       return handler

 

3.4 ioloop

     在Tornado服务器中，IOLoop是调度的核心模块，Tornado服务器回把所有的socket描述符都注册到IOLoop，注册的时候指明回调处理函数，IOLoop内部不断的监听IO事件，一旦发现某个socket可读写，就调用其注册时指定的回调函数。 IOLoop使用了单例模式。

     在Tornado运行的整个过程中，只有一个IOLoop实例，仅需一个 IOLoop实例, 就可以处理全部的IO事件。上文中多次用到了ioloop.IOLoop.instance()这个方法。它会返回ioloop的一个单例,通过查看是否存在属性来实现。

    

    @staticmethod
    def instance():
        """Returns a global IOLoop instance.

        Most single-threaded applications have a single, global IOLoop.
        Use this method instead of passing around IOLoop instances
        throughout your code.

        A common pattern for classes that depend on IOLoops is to use
        a default argument to enable programs with multiple IOLoops
        but not require the argument for simpler applications::

            class MyClass(object):
                def __init__(self, io_loop=None):
                    self.io_loop = io_loop or IOLoop.instance()
        """
        if not hasattr(IOLoop, "_instance"):
            with IOLoop._instance_lock:
                if not hasattr(IOLoop, "_instance"):
                    # New instance after double check
                    IOLoop._instance = IOLoop()
        return IOLoop._instance

       start函数开始了死循环，代码如下：

 

    def start(self):
        //开始死循环处理
        while True:
            poll_timeout = 3600.0

            # Prevent IO event starvation by delaying new callbacks
            # to the next iteration of the event loop.
            //最先处理系统产生的事件，比如定时查看文件是否被修改
              with self._callback_lock:
                callbacks = self._callbacks
                self._callbacks = []
            for callback in callbacks:
                self._run_callback(callback)
            //计算poll的超时时间，根据超时时间用堆来存放事件。
            if self._timeouts:
                now = time.time()
                while self._timeouts:
                    if self._timeouts[0].callback is None:
                        # the timeout was cancelled
                        heapq.heappop(self._timeouts)
                    elif self._timeouts[0].deadline <= now:
                        timeout = heapq.heappop(self._timeouts)
                        self._run_callback(timeout.callback)
                    else:
                        seconds = self._timeouts[0].deadline - now
                        poll_timeout = min(seconds, poll_timeout)
                        break

            if self._callbacks:
                # If any callbacks or timeouts called add_callback,
                # we don't want to wait in poll() before we run them.
                poll_timeout = 0.0

            if not self._running:
                break

            if self._blocking_signal_threshold is not None:
                # clear alarm so it doesn't fire while poll is waiting for
                # events.
                signal.setitimer(signal.ITIMER_REAL, 0, 0)
            //等待网络事件
            try:
                event_pairs = self._impl.poll(poll_timeout)
            except Exception, e:
                # Depending on python version and IOLoop implementation,
                # different exception types may be thrown and there are
                # two ways EINTR might be signaled:
                # * e.errno == errno.EINTR
                # * e.args is like (errno.EINTR, 'Interrupted system call')
                if (getattr(e, 'errno', None) == errno.EINTR or
                    (isinstance(getattr(e, 'args', None), tuple) and
                     len(e.args) == 2 and e.args[0] == errno.EINTR)):
                    continue
                else:
                    raise

            if self._blocking_signal_threshold is not None:
                signal.setitimer(signal.ITIMER_REAL,
                                 self._blocking_signal_threshold, 0)

            # Pop one fd at a time from the set of pending fds and run
            # its handler. Since that handler may perform actions on
            # other file descriptors, there may be reentrant calls to
            # this IOLoop that update self._events
            //根据文件描述符找到回调函数处理网络事件
            self._events.update(event_pairs)
            while self._events:
                fd, events = self._events.popitem()
                try:
                    self._handlers[fd](fd, events)
                except (OSError, IOError), e:
                    if e.args[0] == errno.EPIPE:
                        # Happens when the client closes the connection
                        pass
                    else:
                        logging.error("Exception in I/O handler for fd %s",
                                      fd, exc_info=True)
                except Exception:
                    logging.error("Exception in I/O handler for fd %s",
                                  fd, exc_info=True)
        # reset the stopped flag so another start/stop pair can be issued
        self._stopped = False
        if self._blocking_signal_threshold is not None:
            signal.setitimer(signal.ITIMER_REAL, 0, 0)

 

4 性能比较

   这是一段官网上的描述：

      “一个 Web 应用的性能表现，主要看它的整体架构，而不仅仅是前端的表现。和其它的 Python Web 框架相比，Tornado 的速度要快很多。我们在一些流行的 Python Web 框架上（Django、web.py、CherryPy），针对最简单的 Hello, world 例子作了一个测试。对于 Django 和 web.py，我们使用 Apache/mod_wsgi 的方式来带，CherryPy 就让它自己裸跑。这也是在生产环境中各框架常用的部署方案。对于我们的 Tornado，使用的部署方案为前端使用nginx 做反向代理，带动 4 个线程模式的 Tornado，这种方案也是我们推荐的在生产环境下的 Tornado 部署方案（根据具体的硬件情况，我们推荐一个 CPU 核对应一个 Tornado 伺服实例，我们的负载测试使用的是四核处理器）。我们使用 Apache Benchmark (ab)，在另外一台机器上使用了如下指令进行负载测试：

ab -n 100000 -c 25 http://10.0.1.x/

在 AMD Opteron 2.4GHz 的四核机器上，结果如下图所示：

 

       

    在我们的测试当中，相较于第二快的服务器，Tornado 在数据上的表现也是它的 4 倍之多。即使只用了一个 CPU 核的裸跑模式，Tornado 也有 33% 的优势。”

 

使用同样的参数，对旺旺灰度发布服务器测试结果如下：

ab -n 20000 -c 50 'http://10.20.147.160:8080/redirect?uid=cnalichntest&ver=6.05.10&ctx=alitalk&site=cnalichn'

配置nginx + 1个tornado服务器的时候：Requests per second:    672.55 [#/sec] (mean)

配置nginx + 4个tornado服务器的时候：Requests per second:    2187.45 [#/sec] (mean)