tornado的gen.engine浅析

1. 首先看看示例代码，也是官方给出的：

 1 class Main(RequestHandler):
 2     @tornado.web.asynchronous
 3     @gen.engine
 4     def get(self):
 5         if _DEBUG:
 6             pdb.set_trace()
 7         http_client = AsyncHTTPClient()
 8         response = yield gen.Task(http_client.fetch, "http://code.rootk.com")
 9         self.write(str(response))
10         self.finish()

2. 初始化异步的HTTP客户端，tornado自带的。

http_client = AsyncHTTPClient()

3. @gen.engine

这是一个装饰器，它装饰生成器函数，看看实现的代码：

 1 def engine(func):
 2     @functools.wraps(func)
 3     def wrapper(*args, **kwargs):
 4         runner = None
 5 
 6         def handle_exception(typ, value, tb):
 7             # if the function throws an exception before its first "yield"
 8             # (or is not a generator at all), the Runner won't exist yet.
 9             # However, in that case we haven't reached anything asynchronous
10             # yet, so we can just let the exception propagate.
11             if runner is not None:
12                 return runner.handle_exception(typ, value, tb)
13             return False
14         with ExceptionStackContext(handle_exception) as deactivate:
15             # gen是一个RequestHandler的子类的方法
16             gen = func(*args, **kwargs)
17             # 如果方法是一个生成器
18             if isinstance(gen, types.GeneratorType):
19                 # 实例化Runner类，传入生成器作为参数
20                 runner = Runner(gen, deactivate)
21                 runner.run()
22                 return
23             assert gen is None, gen
24             deactivate()
25             # no yield, so we're done
26     return wrapper

(1). 在代码runner = Runner(gen, deactivate)之中，gen指的是示例代码中get这个生成器函数。

(2). 然后调用实例的run方法。

4. 分析一下类的Runner实例的run方法。

run函数循环会执行三遍：
第一遍，当gen.engine执行时，遇到runner.run()会执行一遍，最后会调用Task的start方法
这次执行的结果是让示例代码中的yield语句运行完毕。

第二遍，start方法执行完毕后，http_client.fetch方法会回调Runner的result_callback方法，会再次执行。
这次执行的结果是向生成器函数发送执行结果：
yielded = self.gen.send(next)
next的内容就是在上面的代码中：
next = self.yield_point.get_result()
而且self.finished = True，被设置为True。

第三遍，最后start方法执行完毕(它是由第一步调用的)，run方法里的循环会最后一次运行。
当它发现self.finished被设置为True时，直接return。

实现代码：

 1 if self.running or self.finished:
 2             return
 3         try:
 4             self.running = True
 5             while True:
 6                 # 如果上一次执行没有抛出异常
 7                 if self.exc_info is None:
 8                     try:
 9                         if not self.yield_point.is_ready():
10                             return
11                         # self.yield_point是一个_NullYieldPoint类的实例
12                         # 实例的get_result方法总是返回None
13                         next = self.yield_point.get_result()
14                     except Exception:
15                         self.exc_info = sys.exc_info()
16                 try:
17                     if self.exc_info is not None:
18                         self.had_exception = True
19                         exc_info = self.exc_info
20                         self.exc_info = None
21                         yielded = self.gen.throw(*exc_info)
22                     else:
23                         # 这里调用生成器函数的send方法，参数为None
24                         # 对应到示例中的代码，就完整的执行了下面这一句
25                         # yield gen.Task(http_client.fetch, "http://code.rootk.com")
26                         # 返回了一个Task类的实例
27                         # 因为可以同时执行多个任务，所以返回的也有可能是含有一个Task类的列表
28                         # 下面会做判断
29                         yielded = self.gen.send(next)
30                 except StopIteration:
31                     self.finished = True
32                     if self.pending_callbacks and not self.had_exception:
33                         # If we ran cleanly without waiting on all callbacks
34                         # raise an error (really more of a warning).  If we
35                         # had an exception then some callbacks may have been
36                         # orphaned, so skip the check in that case.
37                         raise LeakedCallbackError(
38                             "finished without waiting for callbacks %r" %
39                             self.pending_callbacks)
40                     self.deactivate_stack_context()
41                     return
42                 except Exception:
43                     self.finished = True
44                     raise
45                 # 判断yield返回的内容
46                 # 如果是列表类型(说明返回了多个任务，即多个Task类的实例)
47                 if isinstance(yielded, list):
48                     yielded = Multi(yielded)
49                 # 如果是YieldPoint类的实例，就是Task类本身(Task类是YieldPoint类的子类)
50                 # 说明只有一个任务(一个Task类的实例)
51                 if isinstance(yielded, YieldPoint):
52                     self.yield_point = yielded
53                     try:
54                         # 调用Task类实例的start方法
55                         # 参数为Runner类实例自身
56                         self.yield_point.start(self)
57                     except Exception:
58                         self.exc_info = sys.exc_info()
59                 else:
60                     self.exc_info = (BadYieldError("yielded unknown object %r" % yielded),)
61         finally:
62             self.running = False

到这里Runner类实例的run方法执行完毕，接下来执行到了Task类实例的start方法。

5. 看看Task类实例的start方法的实现代码：

def start(self, runner):
    # runner是一个Runner类实例
    self.runner = runner
        
    # object每次返回了不同的python的对象实例
    # 每次生成的对象的id都不一样，所以把它当作一个唯一的key
    self.key = object()
        
    # 调用runner的register_callback方法，以key作为参数
    # 该方法只有一句代码：self.pending_callbacks.add(key)
    # 在self.pending_callbacks这个set类型数据中，加入key
    runner.register_callback(self.key)
        
    # 加入callback，以key作为参数
    # result_callback是一个装饰器
    # 返回的是内层函数
    # !!!下面的self.func执行成功后，会调用这个返回的内层函数!!!
    self.kwargs["callback"] = runner.result_callback(self.key)
    
    # 执行真正的函数，示例代码中执行的就是http_client.fetch函数
    # self.kwargs中已经加入了一个关键字参数：'callback'
    self.func(*self.args, **self.kwargs)

6. 负责返回内容的回调函数，就是Runner类实例的result_callback方法，它是一个装饰器。

# 外层函数接受key作为参数
def result_callback(self, key):
    # 内层函数负责返回内容
    # 由http_client.fetch函数调用
    # result就是fetch函数返回的内容
    def inner(*args, **kwargs):
        if _DEBUG:
            pdb.set_trace()
        if kwargs or len(args) > 1:
            result = Arguments(args, kwargs)
        elif args:
            result = args[0]
        else:
            result = None
        #最后调用Runner类实例的set_result方法
        self.set_result(key, result)
    return inner

7. 将返回的内容添加到Runner类实例的result字典中

def set_result(self, key, result):
    """Sets the result for ``key`` and attempts to resume the generator."""
    # 将内容加入到self.results字典中
    self.results[key] = result
    #最后调用一遍run
    self.run()