深刻tornado中的協程

時間 2019-12-09

標籤深刻 tornado 欄目 Python 简体版

原文原文鏈接

tornado使用了單進程(固然也能夠多進程) + 協程 + I/O多路複用的機制，解決了C10K中由於過多的線程(進程)的上下文切換而致使的cpu資源的浪費。html

tornado中的I/O多路複用前面已經講過了。本文不作詳細解釋。app

來看一下tornado中的協程模塊：tornado.gen：異步

tornado.gen是根據生成器(generator)實現的，用來更加簡單的實現異步。async

先來講一下tornado.gen.coroutine的實現思路：函數

　　咱們知道generator中的yield語句可使函數暫停執行，而send()方法則能夠恢復函數的執行。tornado

　　tornado將那些異步操做放置到yield語句後，當這些異步操做完成後，tornado會將結果send()至generator中恢復函數執行。oop

在tornado的官方文檔中有這麼一句話：this

Most asynchronous functions in Tornado return a Future; yielding this object returns its result..net

就是說：在tornado中大多數的異步操做返回一個Future對象，yield Future對象會返回該異步操做的結果。線程

那麼，Future對象究竟是什麼？

一 Future對象

先來講說Future對象：

Future對象能夠歸納爲：一個異步操做的佔位符，固然這個佔位符有些特殊，它特殊在：

　　1 這個佔位符是一個對象

　　2 這個對象包含了不少屬性，包括_result 以及 _callbacks，分別用來存儲異步操做的結果以及回調函數

　　3 這個對象包含了不少方法，好比添加回調函數，設置異步操做結果等。

　　4 當這個對象對應的異步操做完成後，該對象會被set_done，而後遍歷並運行_callbacks中的回調函數

來看一下Future的簡化版：

class Future(object):
    '''
        Future對象主要保存一個回調函數列表_callbacks與一個執行結果_result，當咱們set_result時，就會執行_callbacks中的函數
        若是set_result或者set_done，就會遍歷_callbacks列表並執行callback(self)函數
    '''
    def __init__(self):
        self._result = None    # 執行的結果
        self._callbacks = []    # 用來保存該future對象的回調函數

    def result(self, timeout=None):
        # 若是操做成功，返回結果。若是失敗則拋出異常
        self._clear_tb_log()
        if self._result is not None:
            return self._result
        if self._exc_info is not None:
            raise_exc_info(self._exc_info)
        self._check_done()
        return self._result

    def add_done_callback(self, fn):
        if self._done:
            fn(self)
        else:
            self._callbacks.append(fn)

    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

完整源碼：

Future源碼

二 gen.coroutine裝飾器

tornado中的協程是經過tornado.gen中的coroutine裝飾器實現的：

def coroutine(func, replace_callback=True):
    return _make_coroutine_wrapper(func, replace_callback=True)

_make_coroutine_wrapper ：

def _make_coroutine_wrapper(func, replace_callback):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        '''
            大致過程：
            future = TracebackFuture()  
            result = func(*args, **kwargs)
            if isinstance(result, GeneratorType):
                yielded = next(result)
                Runner(result, future, yielded)
            return future
        '''
        future = TracebackFuture()                   # TracebackFuture = Future

        if replace_callback and 'callback' in kwargs:
            callback = kwargs.pop('callback')
            IOLoop.current().add_future(future, lambda future: callback(future.result()))

        try:
            result = func(*args, **kwargs)           # 執行func，若func中包含yield，則返回一個generator對象
        except (Return, StopIteration) as e:
            result = _value_from_stopiteration(e)
        except Exception:
            future.set_exc_info(sys.exc_info())
            return future
        else:
            if isinstance(result, GeneratorType):      # 判斷其是否爲generator對象
                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(_value_from_stopiteration(e))
                except Exception:
                    future.set_exc_info(sys.exc_info())
                else:
                    Runner(result, future, yielded)　　# Runner(result, future, yield)
                try:
                    return future　　　　　　　　　　　　
                finally:
                    future = None
        future.set_result(result)
        return future
    return wrapper

先來看一下大致過程：

　　1 首先生成一個Future對象

　　2 運行該被裝飾函數並將結果賦值給result。在這裏由於tornado的'異步'實現是基於generator的，因此通常狀況下 result是一個generator對象

　　3 yielded = next(result) 執行到被裝飾函數的第一次yield，將結果賦值給yielded。通常狀況下，yielded很大狀況下是一個Future對象。

　　4 Runner(result, future, yielded)

　　5 return future

除了第4步之外其餘都很好理解，因此來了解一下第四步Runner()幹了些啥：

三 Runner()類

1 爲何要有Runner()？或者說Runner()的做用是什麼？

Runner()能夠自動的將異步操做的結果send()至生成器停止的地方

tornado的協程或者說異步是基於generator實現的，generator較爲經常使用的有兩個方法：send() next() ，關於這兩個方法的流程分析在這。

不少狀況下會有generator的嵌套。好比說常常會yield 一個generator。當A生成器yield B生成器時，分兩步：

　　1 咱們首先停止A的執行轉而執行B

　　2 當B執行完成後，咱們須要將B的結果send()至A停止的地方，繼續執行A

Runner()主要就是來作這些的，也就是控制生成器的執行與停止，並在合適的狀況下使用send()方法同時傳入B生成器的結果喚醒A生成器。

來看一個簡單例子：

View Code

上例中的Runner()僅僅完成了第一步，咱們還須要手動的執行第二步，而tornado的gen的Runner()則作了全套奧！

2 剖析Runner()

在Runner()中主要有三個方法__init__ handle_yield run：

Runner()

2.1 __init__方法

__init__ 裏面執行了一些初始化的操做，最主要是最後兩句：

if self.handle_yield(first_yielded): # 運行
    self.run()

2.2 handle_yield方法

handle_yield(self, yielded) 函數，這個函數顧名思義，就是用來處理yield返回的對象的。

首先咱們假設yielded是一個Future對象(由於這是最經常使用的狀況)，這樣的話代碼就縮減了不少

def handle_yield(self, yielded):
        self.future = convert_yielded(yielded)                         # 若是yielded是Future對象則原樣返回
        if not self.future.done() or self.future is moment:            # moment是tornado初始化時就創建的一個Future對象，且被set_result(None)
            self.io_loop.add_future(self.future, lambda f: self.run()) # 爲該future添加callback
            return False
        return True

也就是幹了三步：

　　首先解析出self.future

　　而後判斷self.future對象是否已經被done(完成)，若是沒有的話爲其添加回調函數，這個回調函數會執行self.run()

　　返回self.future對象是否被done

整體來講，handle_yield返回yielded對象是否被set_done，若是沒有則爲yielded對象添加回調函數，這個回調函數執行self.run()

還有一個有趣的地方，就是上面代碼的第四行： self.io_loop.add_future(self.future, lambda f: self.run())

def add_future(self, future, callback):
    # 爲future添加一個回調函數，這個回調函數的做用是：將參數callback添加至self._callbacks中
    # 你們思考一個問題： 若是某個Future對象被set_done,那麼他的回調函數應該在何時執行？ 
    # 是當即執行亦或者是將回調函數添加到IOLoop實例的_callbacks中進行統一執行？ 
    # 雖然前者更簡單，但致使回調函數的執行過於混亂，咱們應該讓全部知足執行條件的回調函數統一執行。顯而後者更合理
    # 而add_future()的做用就是這樣
    future.add_done_callback(lambda future: self.add_callback(callback, future))
        
def add_callback(self, callback, *args, **kwargs):
    # 將callback添加至_callbacks列表中
    self._callbacks.append(functools.partial(callback, *args, **kwargs))

2.3 run方法

再來看self.run()方法。這個方法實際上就是一個循環，不停的執行generator的send()方法，發送的值就是yielded的result。

咱們能夠將run()方法簡化一下：

    def run(self):
        """Starts or resumes the generator, running until it reaches a
        yield point that is not ready. 循環向generator中傳遞值，直到某個yield返回的yielded尚未被done
        """
        try:
            self.running = True 
            while True:
                future = self.future　　
                if not future.done():
                    return
                self.future = None　　　　　　# 清空self.future
                value = future.result()　　　# 獲取future對象的結果
                try:    
                    yielded = self.gen.send(value)　　# send該結果，並將self.gen返回的值賦值給yielded(通常狀況下這也是個future對象)
                except (StopIteration, Return) as e:
                    self.finished = True
                    self.future = _null_future
                    self.result_future.set_result(_value_from_stopiteration(e))
                    self.result_future = None
                    self._deactivate_stack_context()
                    return
                if not self.handle_yield(yielded):　　# 運行self.handler_yield(yielded)，若是yielded對象沒有被done，則直接返回；不然繼續循環
                    return
        finally:
            self.running = False

總結：

　　1 每個Future對應一個異步操做

　　2 該Future對象能夠添加回調函數，當該異步操做完成後，須要對該Future對象設置set_done或者set_result，而後執行其全部的回調函數

　　3 凡是使用了coroutine裝飾器的generator函數都會返回一個Future對象，同時會不斷爲該generator，該generator每一次運行send()或者next()的返回結果yielded以及future對象運行Runner()

　　4 Runner()會對generator不斷進行send()或者next()操做。具體步驟是：上一個next()或者send()操做返回的yielded(通常是一個Future對象)被set_done後，將該yielded對象的結果send()至generator中，不斷循環該操做，直到產生StopIteration或者Return異常(這表示該generator執行結束)，這時會爲該generator對應的Future對象set_result。

咱們能夠看到tornado的協程是基於generator的，generator能夠經過yield關鍵字暫停執行，也能夠經過next()或者send()恢復執行，同時send()能夠向generator中傳遞值。

而將協程鏈接起來的紐帶則是Future對象，每個Future對象都對應着一個異步操做，咱們能夠爲該對象添加許多回調函數，當異步操做完成後經過對Future對象進行set_done或者set_result就能夠執行相關的回調函數。

提供動力的則是Runner()，他不停的將generator所yield的每個future對象的結果send()至generator，當generator運行結束，他會進行最後的包裝工做，對該generator所對應的Future對象執行set_result操做。