现在的位置: 首页 > 综合 > 正文

tornado之ioloop类源码分析 Tornado 之 IOLoop类分析

2018年04月12日 ⁄ 综合 ⁄ 共 4213字 ⁄ 字号小中大 ⁄ 评论关闭

Tornado 之 IOLoop类分析

分类： Tornado 学习 http://blog.csdn.net/hpghy123456/article/details/91692412013-06-25
14:04 962人阅读评论(1) 收藏举报

源代码 Tornado 1-2-1

IOLoop类是Tornado的边缘触发事件驱动模型，在Linux平台下面封装的是epoll模型，这个类的代码也很简单，比Nginx好看多了。

先看属性：

_handlers保存（fd，handler）的映射关系，_events保存就绪的fd以及对应的events事件（读／写／错误），_blocking_signal_threshold表示超时阈值，如果不为None，当select.poll（）返回后，调用fd对应的handler工作时间超过这个值时，就会触发SIGALRM信号，set_blocking_signal_threshold可以设置这个阈值，并安装SIGALRM信号处理函数。_callbacks 存储的是一些函数对象（回调函数），具体作用暂且不懂。

方法分析：

1、初始化：注意管道的作用，和Nginx的进程模型类似，管道也是用于其他进程向ioloop.start（）进程发送命令。Tornado中不解析命令，只是简单的退出监听循环。

[python]view

 plaincopy

def __init__(self, impl=None):  

        .....  

    # hp: 管道，用来发送命令给epoll_wait进程. 在这里只是为了唤醒epoll.poll()/epoll_wait,接着退出ioloop循环,停止监听事件  

        if os.name != 'nt':  

            r, w = os.pipe()  

            self._set_nonblocking(r)  

            self._set_nonblocking(w)  

            self._set_close_exec(r)  

            self._set_close_exec(w)  

            self._waker_reader = os.fdopen(r, "rb", 0)      # hp: fdopen convert fileno to FILE*, bufsize=0  

            self._waker_writer = os.fdopen(w, "wb", 0)  

        else:  

            .....         

    # hp: 只监听管道读事件，读取其他进程发送给ioloop进程的命令,使得ioloop退出监听循环  

        self.add_handler(r, self._read_waker, self.READ)

2、add_handler， update_handler、remove_handler就是简单的对epoll_ctl的封装，没有什么特殊的地方。

3、start函数，进程的监听循环，循环监听所有的fd，并调用fd对应的handler，和Nginx类似。

[python]view

 plaincopy

def start(self):  

        if self._stopped:  

            self._stopped = False  

            return  

        self._running = True  

        while True:  

            poll_timeout = 0.2  

            callbacks = self._callbacks  

            self._callbacks = []  

            for callback in callbacks:  

                self._run_callback(callback)  

            if self._callbacks:     # hp: 现在还没有看  

                poll_timeout = 0.0  

    # hp: 定时机制和Nginx一模一样   

            if self._timeouts:      # hp: 定时事件列表  

                now = time.time()  

        # hp: 处理超时的事件  

                while self._timeouts and self._timeouts[0].deadline <= now:  

                    timeout = self._timeouts.pop(0)  

                    self._run_callback(timeout.callback)  

                if self._timeouts:  

                    milliseconds = self._timeouts[0].deadline - now  

        # hp: 时间设置和Nginx一样,取下一个定时事件的时间  

                    poll_timeout = min(milliseconds, poll_timeout)  

            if not self._running:   # hp: 调用stop()设置为false,通过管道给epoll发送消息，epoll醒来便退出循环  

                break  

            if self._blocking_signal_threshold is not None:  

                # clear alarm so it doesn't fire while poll is waiting for  

                # events.  

        # hp: 清空之前的定时器  

                signal.setitimer(signal.ITIMER_REAL, 0, 0)  

            try:  

                event_pairs = self._impl.poll(poll_timeout)  

            except Exception, e:  

                ...  

        # hp: 安装定时器  

            if self._blocking_signal_threshold is not None:  

                signal.setitimer(signal.ITIMER_REAL,  

                                 self._blocking_signal_threshold, 0)   

            self._events.update(event_pairs)  

            while self._events:  

                fd, events = self._events.popitem()  

                try:  

            # hp: 处理激活的fd  

                    self._handlers[fd](fd, events)  

                except   

    # 退出监听循环  

        # reset the stopped flag so another start/stop pair can be issued  

        self._stopped = False  

        if self._blocking_signal_threshold is not None:  

            # hp: 清除定时器  

            signal.setitimer(signal.ITIMER_REAL, 0, 0)

4、stop函数：自己还不太习惯在面向对象编程中操作进程(习惯用C语言)，比如：如何调用stop()函数,当前进程已经在start()中的epoll.poll中阻塞了，当前进程已经不能调用stop()，其他子进程中的ioloop对象处于不同的进程空间，只用把IOLoop的实例放入共享内存中，其他进程才能调用stop()从而终止当前进程的epoll循环。

[python]view

 plaincopy

def stop(self):  

        self._running = False  

        self._stopped = True  

        self._wake()  

def _wake(self):  

        '''''hp: 退出当前的epoll_wait调用'''  

        try:  

            self._waker_writer.write("x")  

        except IOError:  

            pass