歡迎光臨
每天分享高質量文章

Tornado 原始碼閱讀:初步認識

作者:國夫君

來源:見文末

ioloop


`ioloop``tornado`的核心模塊,也是個調度模塊,各種異步事件都是由他調度的,所以必須弄清他的執行邏輯

 

原始碼分析


而`ioloop`的核心部分則是 `while True`這個迴圈內部的邏輯,貼上他的代碼下

   def start(self):

        if self._running:

            raise RuntimeError(“IOLoop is already running”)

        self._setup_logging()

        if self._stopped:

            self._stopped = False

            return

        old_current = getattr(IOLoop._current, “instance”, None)

        IOLoop._current.instance = self

        self._thread_ident = thread.get_ident()

        self._running = True

 

        old_wakeup_fd = None

        if hasattr(signal, ‘set_wakeup_fd’) and os.name == ‘posix’:

 

            try:

                old_wakeup_fd = signal.set_wakeup_fd(self._waker.write_fileno())

                if old_wakeup_fd != –1:

 

                    signal.set_wakeup_fd(old_wakeup_fd)

                    old_wakeup_fd = None

            except ValueError:

 

                old_wakeup_fd = None

 

        try:

            while True:

 

                with self._callback_lock:

                    callbacks = self._callbacks

                    self._callbacks = []

 

                due_timeouts = []

 

                if self._timeouts:

                    now = self.time()

                    while self._timeouts:

                        if self._timeouts[0].callback is None:

 

                            heapq.heappop(self._timeouts)

                            self._cancellations -= 1

                        elif self._timeouts[0].deadline <= now:

                            due_timeouts.append(heapq.heappop(self._timeouts))

                        else:

                            break

                    if (self._cancellations > 512

                            and self._cancellations > (len(self._timeouts) >> 1)):

                        self._cancellations = 0

                        self._timeouts = [x for x in self._timeouts

                                          if x.callback is not None]

                        heapq.heapify(self._timeouts)

 

                for callback in callbacks:

                    self._run_callback(callback)

                for timeout in due_timeouts:

                    if timeout.callback is not None:

                        self._run_callback(timeout.callback)

 

                callbacks = callback = due_timeouts = timeout = None

 

                if self._callbacks:

 

                    poll_timeout = 0.0

                elif self._timeouts:

 

                    poll_timeout = self._timeouts[0].deadline – self.time()

                    poll_timeout = max(0, min(poll_timeout, _POLL_TIMEOUT))

                else:

 

                    poll_timeout = _POLL_TIMEOUT

 

                if not self._running:

                    break

 

                if self._blocking_signal_threshold is not None:

 

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

 

                try:

                    event_pairs = self._impl.poll(poll_timeout)

                except Exception as e:

 

                    if errno_from_exception(e) == errno.EINTR:

                        continue

                    else:

                        raise

 

                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:

                        fd_obj, handler_func = self._handlers[fd]

                        handler_func(fd_obj, events)

                    except (OSError, IOError) as e:

                        if errno_from_exception(e) == errno.EPIPE:

 

                            pass

                        else:

                            self.handle_callback_exception(self._handlers.get(fd))

                    except Exception:

                        self.handle_callback_exception(self._handlers.get(fd))

                fd_obj = handler_func = None

 

        finally:

 

            self._stopped = False

            if self._blocking_signal_threshold is not None:

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

            IOLoop._current.instance = old_current

            if old_wakeup_fd is not None:

                signal.set_wakeup_fd(old_wakeup_fd)


除去註釋,代碼其實沒多少行. 由while 內部代碼可以看出ioloop主要由三部分組成:


1.回呼 callbacks


他是ioloop回呼的基礎部分,通過IOLoop.instance().add_callback()添加到self._callbacks,他們將在每一次loop中被運行.


主要用途是將邏輯分塊,在適合時機將包裝好的callback添加到self._callbacks讓其執行.


例如ioloop中的add_future


def add_future(self, future, callback):

        “””Schedules a callback on the “IOLoop“ when the given

        `.Future` is finished.

 

        The callback is invoked with one argument, the

        `.Future`.

        “””

        assert is_future(future)

        callback = stack_context.wrap(callback)

        future.add_done_callback(

            lambda futureself.add_callback(callback, future))


future物件得到result的時候會呼叫future.add_done_callback添加的callback,再將其轉至ioloop執行


2.定時器 due_timeouts


這是定時器,在指定的事件執行callback.


跟1中的callback類似,通過IOLoop.instance().add_callback在每一次迴圈,會計算timeouts回呼串列里的事件,運行已到期的callback.


當然不是無節操的迴圈.


因為poll操作會阻塞到有io操作發生,所以只要計算最近的timeout, 然後用這個時間作為self._impl.poll(poll_timeout) 的 poll_timeout ,就可以達到按時運行了


但是,假設poll_timeout的時間很大時,self._impl.poll一直在堵塞中(沒有io事件,但在處理某一個io事件), 那添加剛纔1中的callback不是要等很久才會被運行嗎? 答案當然是不會.


ioloop中有個waker物件,他是由兩個fd組成,一個讀一個寫.


ioloop在初始化的時候把waker系結到epoll里了,add_callback時會觸發waker的讀寫.


這樣ioloop就會在poll中被喚醒了,接著就可以及時處理timeout callback了


用這樣的方式也可以自己封裝一個小的定時器功能玩玩


3.io事件的event loop


處理epoll事件的功能

通過IOLoop.instance().add_handler(fd, handler, events)系結fd event的處理事件

在httpserver.listen的代碼內,

netutil.py中的netutil.py的add_accept_handler系結accept handler處理客戶端接入的邏輯


如法炮製,其他的io事件也這樣系結,業務邏輯的分塊交由ioloop的callback和future處理


關於epoll的用法的內容.詳情見我第一篇文章吧,哈哈


總結


ioloop由callback(業務分塊), timeout callback(定時任務) io event(io傳輸和解析) 三塊組成,互相配合完成異步的功能,構建gen,httpclient,iostream等功能。


串聯大致的流程是,tornado 系結io event,處理io傳輸解析,傳輸完成後(結合Future)回呼(callback)業務處理的邏輯和一些固定操作 . 定時器則是較為獨立的模塊


Futrue


個人認為Future是tornado僅此ioloop重要的模塊,他貫穿全文,所有異步操作都有他的身影。顧名思義,他主要是關註日後要做的事,類似jquery的Deferred吧


一般的用法是通過ioloop的add_future定義future的done callback, 當future被set_result的時候,future的done callback就會被呼叫. 從而完成Future的功能.


具體可以參考gen.coroutine的實現,本文後面也會講到


他的組成不複雜,只有幾個重要的方法,最重要的是 add_done_callback , set_result


tornado用Future和ioloop,yield實現了gen.coroutine


1. add_done_callback


跟ioloop的callback類似 , 儲存事件完成後的callback在self._callbacks里


def add_done_callback(self, fn):

        if self._done:

            fn(self)

        else:

            self._callbacks.append(fn)


2.set_result


設置事件的結果,並運行之前儲存好的callback


def set_result(self, result):

        self._result = result

        self._set_done()

 

def _set_done(self):

        self._done = True

        for cb in self._callbacks:

            try:

                cb(self)

            except Exception:

                app_log.exception(‘Exception in callback %r for %r’,

                                  cb, self)

        self._callbacks = None


為了驗證之前所說的,上一段測試代碼


#! /usr/bin/env python

#coding=utf-8

 

import tornado.web

import tornado.ioloop

 

from tornado.gen import coroutine

from tornado.concurrent import Future

 

 

def test():

    def pp(s):

        print s

 

    future = Future()

    iol = tornado.ioloop.IOLoop.instance()

 

    print ‘init future %s’%future

 

    iol.add_future(future, lambda fpp(‘ioloop callback after future done,future is %s’%f))

 

    #模擬io延遲操作

    iol.add_timeout(iol.time()+5,lambda:future.set_result(‘set future is done’))

 

    print ‘init complete’

    tornado.ioloop.IOLoop.instance().start()

 

if __name__ == “__main__”:

    test() 


運行結果:



gen.coroutine


接著繼續延伸,看看coroutine的實現。


gen.coroutine實現的功能其實是將原來的callback的寫法,用yield的寫法代替. 即以yield為分界,將代碼分成兩部分.


如:


#! /usr/bin/env python

#coding=utf-8

 

import tornado.ioloop

from tornado.gen import coroutine

from tornado.httpclient import AsyncHTTPClient

 

@coroutine

def cotest():

    client = AsyncHTTPClient()

    res = yield client.fetch(“http://www.segmentfault.com/”)

    print res

 

if __name__ == “__main__”:

    f = cotest()    

    print f #這裡傳回了一個future哦

    tornado.ioloop.IOLoop.instance().start()


運行結果:



原始碼分析


接下來分析下coroutine的實現


def _make_coroutine_wrapper(func, replace_callback):

 

    @functools.wraps(func)

    def wrapper(*args, **kwargs):

        future = TracebackFuture()

 

        if replace_callback and ‘callback’ in kwargs:

            callback = kwargs.pop(‘callback’)

            IOLoop.current().add_future(

                future, lambda futurecallback(future.result()))

 

        try:

            result = func(*args, **kwargs)

        except (Return, StopIteration) as e:

            result = getattr(e, ‘value’, None)

        except Exception:

            future.set_exc_info(sys.exc_info())

            return future

        else:

            if isinstance(result, types.GeneratorType):

                try:

                    orig_stack_contexts = stack_context._state.contexts

                    yielded = next(result)

                    if stack_context._state.contexts is not orig_stack_contexts:

                        yielded = TracebackFuture()

                        yielded.set_exception(

                            stack_context.StackContextInconsistentError(

                                ‘stack_context inconsistency (probably caused ‘

                                ‘by yield within a “with StackContext” block)’))

                except (StopIteration, Return) as e:

                    future.set_result(getattr(e, ‘value’, None))

                except Exception:

                    future.set_exc_info(sys.exc_info())

                else:

                    Runner(result, future, yielded)

                try:

                    return future

                finally:

                    future = None

        future.set_result(result)

        return future

    return wrapper


如原始碼所示,func運行的結果是GeneratorType ,yielded = next(result), 運行至原函式的yield位置,傳回的是原函式func內部 yield 右邊傳回的物件(必須是Future或Future的list)給yielded.經過Runner(result, future, yielded) 對yielded進行處理.在此就 貼出Runner的代碼了.


Runner初始化過程,呼叫handle_yield, 查看yielded是否已done了,否則add_future運行Runner的run方法, run方法中如果yielded物件已完成,用對它的gen呼叫send,發送完成的結果.


所以yielded在什麼地方被set_result非常重要, 當被set_result的時候,才會send結果給原func,完成整個異步操作


詳情可以查看tornado 中重要的物件 iostream,原始碼中iostream的 _handle_connect,如此設置了連接的result.


def _handle_connect(self):

        err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)

        if err != 0:

            self.error = socket.error(err, os.strerror(err))

            if self._connect_future is None:

                gen_log.warning(“Connect error on fd %s: %s”,

                                self.socket.fileno(), errno.errorcode[err])

            self.close()

            return

        if self._connect_callback is not None:

            callback = self._connect_callback

            self._connect_callback = None

            self._run_callback(callback)

        if self._connect_future is not None:

            future = self._connect_future

            self._connect_future = None

            future.set_result(self)

        self._connecting = False


最後貼上一個簡單的測試代碼,演示coroutine,future的用法


import tornado.ioloop

from tornado.gen import coroutine

from tornado.concurrent import Future

 

@coroutine

def asyn_sum(a, b):

    print(“begin calculate:sum %d+%d”%(a,b))

    future = Future()

    future2 = Future()

    iol = tornado.ioloop.IOLoop.instance()

 

    print future

 

    def callback(a, b):

        print(“calculating the sum of %d+%d:”%(a,b))

        future.set_result(a+b)

 

        iol.add_timeout(iol.time()+3,lambda f:f.set_result(None),future2)

    iol.add_timeout(iol.time()+3,callback, a, b)

 

    result = yield future

 

    print(“after yielded”)

    print(“the %d+%d=%d”%(a, b, result))

 

    yield future2

 

    print ‘after future2’

 

def main():

    f =  asyn_sum(2,3)

 

    print 

    print f

    tornado.ioloop.IOLoop.instance().start()

 

if __name__ == “__main__”:

    main()


運行結果:



為什麼代碼中個yield都起作用了? 因為Runner.run里,最後繼續用handle_yield處理了send後傳回的yielded物件,意思是func里可以有n幹個yield操作


if not self.handle_yield(yielded):

        return



至此,已完成tornado中重要的幾個模塊的流程,其他模塊也是由此而來.寫了這麼多,越寫越卡,就到此為止先吧。

來源:國夫君

segmentfault.com/a/1190000002971992

《Linux雲計算及運維架構師高薪實戰班》2018年09月16日即將開課中,120天衝擊Linux運維年薪30萬,改變速約~~~~

    *宣告:推送內容及圖片來源於網絡,部分內容會有所改動,版權歸原作者所有,如來源信息有誤或侵犯權益,請聯繫我們刪除或授權事宜。

    – END –




    更多Linux好文請點擊【閱讀原文】

    ↓↓↓

    赞(0)

    分享創造快樂