Tornado源码分析系列之一: 化异步为'同步'的Future和gen.coroutine

1
2
3
4
5
6
7
8
9

class AsyncHandler(RequestHandler):
@asynchronous
def get(self):
http_client = AsyncHTTPClient()
http_client.fetch('http://example.com', callback=self.on_fetch)

def on_fetch(self, response):
do_something_with_response(response)
self.render('template.html')

1
2
3
4
5
6
7

class GenAsyncHandler(RequestHandler):
@gen.coroutine
def get(self):
http_client = AsyncHTTPClient()
response = yield http_client.fetch('http://example.com')
do_something_with_response(response)
self.render('template.html')

1
2
3
4
5
6
7
8
9

def test():
print ('step 1.......')
res = yield 10
print ('step 2.......', res) (3)

gen = test()
gen.send(None) #next(gen)  (1)
data = gen.send(20) (2)
print ('yield out .....', data)

1
2
3

step 1.......
step 2....... 20
yield out ..... 10

1
2
3
4
5
6
7
8
910
11
12
13
14
15
16
17
18
19
20

class Future(object):
def __init__(self):
self._callback = []
self._result = None
self._done = False

def set_callback(self, cb):
self._callback.append(cb)

def _run_callback(self):
for cb in self._callback:
cb()

def set_result(self, result)
self._done = True
self._result = result
self._run_callback()

def is_ready(self):
return self._done is True

1
2
3
4
5
6
7
8
9

def Task(func, *args, **kwargs):
future = Future
def set_result(result):
if future.done():
return
future.set_result(result)

func(*args, callback=_argument_adapter(set_result), **kwargs)
return future

1
2
3
4
5
6
7
8
910
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

def fetch(self, request, callback=None, raise_error=True, **kwargs):
.....
future = TracebackFuture()
if callback is not None:
callback = stack_context.wrap(callback)

def handle_future(future):
exc = future.exception()
if isinstance(exc, HTTPError) and exc.response is not None:
response = exc.response
elif exc is not None:
response = HTTPResponse(
request, 599, error=exc,
request_time=time.time()-request.start_time
)
else:
response = future.result()
self.io_loop.add_callback(callback, response)
future.add_done_callback(handle_future)

def handle_response(response):
if raise_error and response.error:
future.set_exception(response.error)
else:
future.set_result(response)
self.fetch_impl(request, handle_response)
return future

1
2
3
4
5
6
7
8
9

try:
yielded = next(result)
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)

1
2
3
4
5
6
7
8
910
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

def handle_yield(self, yielded):
#处理YieldPoint忽略掉，但是原理跟Future是一样的
try:
self.future = convert_yielded(yielded)
except BadYieldError:
self.future = TracebackFuture()
self.future.set_exc_info(sys.exc_info())

if not self.future.done() or self.future is moment:
self.io_loop.add_future(
self.future, lambda f: self.run())
return False
return True

def run(self):
if self.running or self.finished:
return
try:
self.running = True
while True:
future = self.future
if not future.done(): #执行run时generator返回的那个future必须已经有结果，否则就没必要传回到generator中了
return
self.future = None
try:
orig_stack_contexts = stack_context._state.contexts
exc_info = None

try:
value = future.result()
except Exception:
self.had_exception = True
exc_info = sys.exc_info()

if exc_info is not None:
yielded = self.gen.throw(*exc_info)
exc_info = None
else:
yielded = self.gen.send(value)

if stack_context._state.contexts is not orig_stack_contexts:
self.gen.throw(
stack_context.StackContextInconsistentError(
'stack_context inconsistency (probably caused '
'by yield within a "with StackContext" block)'))
except (StopIteration, Return) as e:
#generator执行完毕并成功的处理
except Exception:
#generator执行过程中异常的处理
if not self.handle_yield(yielded):
#这里generator还没有执行完毕，yielded是generator迭代过一次之后返回的新yielded。如果yieled还没有被解析出结果就通过handle_yield给yieled设置完成时的重启run的回调,
#否则yielded已经有结果，就再次运行run，所以run中才会有一个循环
return
finally:
self.running = False