Tornado 高并发源码分析之五--- IOLoop 对象

IOLoop主要工作

1、将TCPServer 注册到 IOLoop 的事件记到 _handlers 字段，同时注册 READ 和 ERROR 事件到 epoll

2、IOLoop 启动一个大循环，负责轮询epoll中是否已经有就绪的事件，如果有就执行对应的回调

以下为源码分析，省略部分源码，只取主要部分

 1 class Configurable(object):
 2     """根据子类的配置，来创建一个对象，也就是说，继承自Configurable的子类，可以自己配置产生不同的类，并且每个类都会执行initialize方法
 3     __impl_class = None
 4     __impl_kwargs = None
 5 
 6     def __new__(cls, **kwargs):   
 7         base = cls.configurable_base()
 8         args = {}
 9         if cls is base:
10             impl = cls.configured_class()
11             if base.__impl_kwargs:
12                 args.update(base.__impl_kwargs)
13         else:
14             impl = cls
15         args.update(kwargs)
16         instance = super(Configurable, cls).__new__(impl)
17         instance.initialize(**args)   #执行initialize方法
18         return instance

Configurable

 1 class IOLoop(Configurable):
 2 " 一个大循环，自动根据当前的系统，是 linux 2.5以上选择epoll， mac 选择kqueue， 其他选择select".
 3 @staticmethod
 4 def instance():   #创建一个全局的 IOLoop 单例
 5     if not hasattr(IOLoop, "_instance"):
 6         with IOLoop._instance_lock:
 7             if not hasattr(IOLoop, "_instance"):
 8                 IOLoop._instance = IOLoop()
 9     return IOLoop._instance
10 
11 
12 @classmethod    配置
13 def configurable_base(cls):
14     return IOLoop
15 
16 @classmethod
17 def configurable_default(cls):   #根据系统，配置使用epoll还是 kqueue
18     if hasattr(select, "epoll"):
19         from tornado.platform.epoll import EPollIOLoop
20         return EPollIOLoop
21     if hasattr(select, "kqueue"):
22         # Python 2.6+ on BSD or Mac
23         from tornado.platform.kqueue import KQueueIOLoop
24         return KQueueIOLoop
25     from tornado.platform.select import SelectIOLoop
26     return SelectIOLoop

IOLoop

 1 class PollIOLoop(IOLoop):
 2 "继承自IOLoop， 就是IOLoop 在根据系统选择的时候，真正创建的类对象"
 3 
 4 def add_handler(self, fd, handler, events):
 5     fd, obj = self.split_fd(fd)
 6     self._handlers[fd] = (obj, stack_context.wrap(handler))    #将socket对象的句柄fd作为key， 回调函数handler作为值，添加到 _handlers 字段中， 其实这里的函数就是在TCPServer 启动时添加的 _handle_connection 方法
 7     self._impl.register(fd, events | self.ERROR)   #向_impl (在linux中是epoll，在mac中是kqueue，2.5版本一下的linux中是select)中注册对应句柄fd 的事件(其实就是READ和error事件)
 8 
 9 def start(self):
10   
11     try:
12         while True:   #启动IOLoop 大循环， 这是一个无限循环
13    
14             with self._callback_lock:   #如果有上一个循环没有执行完毕的函数，继续拿出来执行
15                 callbacks = self._callbacks
16                 self._callbacks = []
17 
18         
19             try:
20                 event_pairs = self._impl.poll(poll_timeout)   #从epoll中取出已经就绪的事件
21             except Exception as e:
22                 if errno_from_exception(e) == errno.EINTR:
23                     continue
24                 else:
25                     raise
26 
27             self._events.update(event_pairs)   #将事件更新到 _events dict中
28             while self._events:
29                 fd, events = self._events.popitem()
30                 try:
31                     fd_obj, handler_func = self._handlers[fd]   #从刚开始添加事件到_handler里面取出对应的回调函数
32                     handler_func(fd_obj, events)  #执行回调函数， 其实也就是执行TCPServer中的self._handle_connection 方法
33                 except (OSError, IOError) as e:
34                         self.handle_callback_exception(self._handlers.get(fd))
35                 except Exception:
36                     self.handle_callback_exception(self._handlers.get(fd))
37             fd_obj = handler_func = None   #清空，准备进入下一次循环
38 
39     finally:
40        ignal.set_wakeup_fd(old_wakeup_fd)

PollIOLoop