Python Tornado框架（ioloop对象分析）

while True:
poll_timeout = 3600.0

# Prevent IO event starvation by delaying new callbacks
# to the next iteration of the event loop.
with self._callback_lock:
	callbacks = self._callbacks
	self._callbacks = []
for callback in callbacks:
	self._run_callback(callback)

if self._timeouts:
	now = self.time()
	while self._timeouts:
		if self._timeouts[0].callback is None:
			# the timeout was cancelled
			heapq.heappop(self._timeouts)
		elif self._timeouts[0].deadline <= now:
			timeout = heapq.heappop(self._timeouts)
			self._run_callback(timeout.callback)
		else:
			seconds = self._timeouts[0].deadline - now
			poll_timeout = min(seconds, poll_timeout)
			break

if self._callbacks:
	# If any callbacks or timeouts called add_callback,
	# we don‘t want to wait in poll() before we run them.
	poll_timeout = 0.0

if not self._running:
	break

if self._blocking_signal_threshold is not None:
	# clear alarm so it doesn‘t fire while poll is waiting for
	# events.
	signal.setitimer(signal.ITIMER_REAL, 0, 0)

try:
	event_pairs = self._impl.poll(poll_timeout)
except Exception as e:
	# Depending on python version and IOLoop implementation,
	# different exception types may be thrown and there are
	# two ways EINTR might be signaled:
	# * e.errno == errno.EINTR
	# * e.args is like (errno.EINTR, ‘Interrupted system call‘)
	if (getattr(e, ‘errno‘, None) == errno.EINTR or
		(isinstance(getattr(e, ‘args‘, None), tuple) and
		 len(e.args) == 2 and e.args[0] == errno.EINTR)):
		continue
	else:
		raise

if self._blocking_signal_threshold is not None:
	signal.setitimer(signal.ITIMER_REAL,
					 self._blocking_signal_threshold, 0)

# Pop one fd at a time from the set of pending fds and run
# its handler. Since that handler may perform actions on
# other file descriptors, there may be reentrant calls to
# this IOLoop that update self._events
self._events.update(event_pairs)
while self._events:
	fd, events = self._events.popitem()
	try:
		self._handlers[fd](fd, events)
	except (OSError, IOError) as e:
		if e.args[0] == errno.EPIPE:
			# Happens when the client closes the connection
			pass
		else:
			app_log.error("Exception in I/O handler for fd %s",
						  fd, exc_info=True)
	except Exception:
		app_log.error("Exception in I/O handler for fd %s",
					  fd, exc_info=True)

def __new__(cls, **kwargs):
	base = cls.configurable_base()
	args = {}
	if cls is base:
		impl = cls.configured_class()
		if base.__impl_kwargs:
			args.update(base.__impl_kwargs)
	else:
		impl = cls
	args.update(kwargs)
	instance = super(Configurable, cls).__new__(impl)
	# initialize vs __init__ chosen for compatiblity with AsyncHTTPClient
	# singleton magic.  If we get rid of that we can switch to __init__
	# here too.
	instance.initialize(**args)
	return instance

@staticmethod
def instance():
	"""Returns a global `IOLoop` instance.

	Most applications have a single, global `IOLoop` running on the
	main thread.  Use this method to get this instance from
	another thread.  To get the current thread‘s `IOLoop`, use `current()`.
	"""
	if not hasattr(IOLoop, "_instance"):
		with IOLoop._instance_lock:
			if not hasattr(IOLoop, "_instance"):
				# New instance after double check
				IOLoop._instance = IOLoop()
	return IOLoop._instance

# Our events map exactly to the epoll events
NONE = 0
READ = _EPOLLIN
WRITE = _EPOLLOUT
ERROR = _EPOLLERR | _EPOLLHUP

http_server = tornado.httpserver.HTTPServer(application)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()

def __init__(self, impl=None):
    self._impl = impl or _poll()
    #省略部分代码
    self._waker = Waker()
    self.add_handler(self._waker.fileno(),
                     lambda fd, events: self._waker.consume(),
                     self.READ)

def add_handler(self, fd, handler, events):
    """Registers the given handler to receive the given events for fd."""
    self._handlers[fd] = stack_context.wrap(handler)
    self._impl.register(fd, events | self.ERROR)

while True:
    poll_timeout = 3600.0

    with self._callback_lock:
        callbacks = self._callbacks
        self._callbacks = []
    for callback in callbacks:
        self._run_callback(callback)

    # 超时处理
    if self._timeouts:
        now = time.time()
        while self._timeouts:
            if self._timeouts[0].callback is None:
                # the timeout was cancelled
                heapq.heappop(self._timeouts)
            elif self._timeouts[0].deadline <= now:
                timeout = heapq.heappop(self._timeouts)
                self._run_callback(timeout.callback)
            else:
                seconds = self._timeouts[0].deadline - now
                poll_timeout = min(seconds, poll_timeout)
                break

    if self._callbacks:
        # If any callbacks or timeouts called add_callback,
        # we don‘t want to wait in poll() before we run them.
        poll_timeout = 0.0

    if not self._running:
        break

    if self._blocking_signal_threshold is not None:
        # clear alarm so it doesn‘t fire while poll is waiting for events.
        signal.setitimer(signal.ITIMER_REAL, 0, 0)

    # epoll阻塞，当有事件通知或超时返回event_pairs
    try:
        event_pairs = self._impl.poll(poll_timeout)
    except Exception, e:
        # 异常处理，省略

    # 对epoll返回event_pairs事件的处理
    self._events.update(event_pairs)
    while self._events:
        fd, events = self._events.popitem()
        try:
            self._handlers[fd](fd, events)
        except Exception e:
            # 异常处理，省略

@classmethod
def configurable_default(cls):
	if hasattr(select, "epoll"):
		from tornado.platform.epoll import EPollIOLoop
		return EPollIOLoop
	if hasattr(select, "kqueue"):
		# Python 2.6+ on BSD or Mac
		from tornado.platform.kqueue import KQueueIOLoop
		return KQueueIOLoop
	from tornado.platform.select import SelectIOLoop
	return SelectIOLoop

def start(self):
	"""Starts the I/O loop.

	The loop will run until one of the callbacks calls `stop()`, which
	will make the loop stop after the current event iteration completes.
	"""
	raise NotImplementedError()

def __new__(cls, **kwargs):
	base = cls.configurable_base()
	args = {}
	if cls is base:
		impl = cls.configured_class()
		if base.__impl_kwargs:
			args.update(base.__impl_kwargs)
	else:
		impl = cls
	args.update(kwargs)
	instance = super(Configurable, cls).__new__(impl)
	# initialize vs __init__ chosen for compatiblity with AsyncHTTPClient
	# singleton magic.  If we get rid of that we can switch to __init__
	# here too.
	instance.initialize(**args)
	return instance

@classmethod
def configured_class(cls):
	"""Returns the currently configured class."""
	base = cls.configurable_base()
	if cls.__impl_class is None:
		base.__impl_class = cls.configurable_default()
	return base.__impl_class

@classmethod
def configurable_default(cls):
	if hasattr(select, "epoll"):
		from tornado.platform.epoll import EPollIOLoop
		return EPollIOLoop
	if hasattr(select, "kqueue"):
		# Python 2.6+ on BSD or Mac
		from tornado.platform.kqueue import KQueueIOLoop
		return KQueueIOLoop
	from tornado.platform.select import SelectIOLoop
	return SelectIOLoop

class IOStream(object):
	def read_until(self, delimiter, callback): 
	def read_bytes(self, num_bytes, callback, streaming_callback=None): 
	def read_until_regex(self, regex, callback): 
	def read_until_close(self, callback, streaming_callback=None): 
	def write(self, data, callback=None):

global _task_id
    assert _task_id is None
    if num_processes is None or num_processes <= 0:
        num_processes = cpu_count()
    if ioloop.IOLoop.initialized():
        raise RuntimeError("Cannot run in multiple processes: IOLoop instance "
                           "has already been initialized. You cannot call "
                           "IOLoop.instance() before calling start_processes()")
    logging.info("Starting %d processes", num_processes)
    children = {}

def start_child(i):
	pid = os.fork()
	if pid == 0:
		# child process
		_reseed_random()
		global _task_id
		_task_id = i
		return i
	else:
		children[pid] = i
		return None

for i in range(num_processes):
	id = start_child(i)
	if id is not None:
		return id

num_restarts = 0
    while children:
        try:
            pid, status = os.wait()
        except OSError, e:
            if e.errno == errno.EINTR:
                continue
            raise
        if pid not in children:
            continue
        id = children.pop(pid)
        if os.WIFSIGNALED(status):
            logging.warning("child %d (pid %d) killed by signal %d, restarting",
                            id, pid, os.WTERMSIG(status))
        elif os.WEXITSTATUS(status) != 0:
            logging.warning("child %d (pid %d) exited with status %d, restarting",
                            id, pid, os.WEXITSTATUS(status))
        else:
            logging.info("child %d (pid %d) exited normally", id, pid)
            continue
        num_restarts += 1
        if num_restarts > max_restarts:
            raise RuntimeError("Too many child restarts, giving up")
        new_id = start_child(id)
        if new_id is not None:
            return new_id

if new_id is not None:
    return new_id