详解Python的Twisted框架中reactor事件管理器的用法

from twisted.internet import reactor
reactor.run()

from twisted.internet import pollreactor
pollreactor.install()

from twited.internet import pollreactor
pollreactor.install()
from twisted.internet import reactor
reactor.run()

# twisted/internet/reactor.py
import sys
del sys.modules['twisted.internet.reactor']
from twisted.internet import default
default.install()

# twisted/internet/default.py
def _getInstallFunction(platform):
 """
 Return a function to install the reactor most suited for the given platform.
 
 @param platform: The platform for which to select a reactor.
 @type platform: L{twisted.python.runtime.Platform}
 
 @return: A zero-argument callable which will install the selected
 reactor.
 """
 try:
 if platform.isLinux():
 try:
 from twisted.internet.epollreactor import install
 except ImportError:
 from twisted.internet.pollreactor import install
 elif platform.getType() == 'posix' and not platform.isMacOSX():
 from twisted.internet.pollreactor import install
 else:
 from twisted.internet.selectreactor import install
 except ImportError:
 from twisted.internet.selectreactor import install
 return install
 
 
install = _getInstallFunction(platform)

# twisted/internet/selectreactor.py：
def install():
 """Configure the twisted mainloop to be run using the select() reactor.
 """
 # 单例
 reactor = SelectReactor()
 from twisted.internet.main import installReactor
 installReactor(reactor)
 
# twisted/internet/main.py：
def installReactor(reactor):
 """
 Install reactor C{reactor}.
 
 @param reactor: An object that provides one or more IReactor* interfaces.
 """
 # this stuff should be common to all reactors.
 import twisted.internet
 import sys
 if 'twisted.internet.reactor' in sys.modules:
 raise error.ReactorAlreadyInstalledError("reactor already installed")
 twisted.internet.reactor = reactor
 sys.modules['twisted.internet.reactor'] = reactor

SelectReactor
# twisted/internet/selectreactor.py
@implementer(IReactorFDSet)
class SelectReactor(posixbase.PosixReactorBase, _extraBase)

# twisted/internet/interfaces.py
class IReactorFDSet(Interface):
 
 def addReader(reader):
 
 def addWriter(writer):
 
 def removeReader(reader):
 
 def removeWriter(writer):
 
 def removeAll():
 
 def getReaders():
 
 def getWriters():
reactor.listenTCP()

# twisted/internet/posixbase.py
@implementer(IReactorTCP, IReactorUDP, IReactorMulticast)
class PosixReactorBase(_SignalReactorMixin, _DisconnectSelectableMixin,
 ReactorBase):
 
 def listenTCP(self, port, factory, backlog=50, interface=''):
 p = tcp.Port(port, factory, backlog, interface, self)
 p.startListening()
 return p
 
# twisted/internet/tcp.py
@implementer(interfaces.IListeningPort)
class Port(base.BasePort, _SocketCloser):
 def __init__(self, port, factory, backlog=50, interface='', reactor=None):
 """Initialize with a numeric port to listen on.
 """
 base.BasePort.__init__(self, reactor=reactor)
 self.port = port
 self.factory = factory
 self.backlog = backlog
 if abstract.isIPv6Address(interface):
 self.addressFamily = socket.AF_INET6
 self._addressType = address.IPv6Address
 self.interface = interface
 ...
 
 def startListening(self):
 """Create and bind my socket, and begin listening on it.
 创建并绑定套接字，开始监听。
 
 This is called on unserialization, and must be called after creating a
 server to begin listening on the specified port.
 """
 if self._preexistingSocket is None:
 # Create a new socket and make it listen
 try:
 # 创建套接字
 skt = self.createInternetSocket()
 if self.addressFamily == socket.AF_INET6:
 addr = _resolveIPv6(self.interface, self.port)
 else:
 addr = (self.interface, self.port)
 # 绑定
 skt.bind(addr)
 except socket.error as le:
 raise CannotListenError(self.interface, self.port, le)
 # 监听
 skt.listen(self.backlog)
 else:
 # Re-use the externally specified socket
 skt = self._preexistingSocket
 self._preexistingSocket = None
 # Avoid shutting it down at the end.
 self._shouldShutdown = False
 
 # Make sure that if we listened on port 0, we update that to
 # reflect what the OS actually assigned us.
 self._realPortNumber = skt.getsockname()[1]
 
 log.msg("%s starting on %s" % (
 self._getLogPrefix(self.factory), self._realPortNumber))
 
 # The order of the next 5 lines is kind of bizarre. If no one
 # can explain it, perhaps we should re-arrange them.
 self.factory.doStart()
 self.connected = True
 self.socket = skt
 self.fileno = self.socket.fileno
 self.numberAccepts = 100
 
 # startReading调用reactor的addReader方法将Port加入读集合
 self.startReading()

# twisted/internet/posixbase.py
@implementer(IReactorTCP, IReactorUDP, IReactorMulticast)
class PosixReactorBase(_SignalReactorMixin, _DisconnectSelectableMixin,
 ReactorBase)
 
# twisted/internet/base.py
class _SignalReactorMixin(object):
 
 def startRunning(self, installSignalHandlers=True):
 """
 PosixReactorBase的父类_SignalReactorMixin和ReactorBase都有该函数，但是
 _SignalReactorMixin在前，安装mro顺序的话，会先调用_SignalReactorMixin中的。
 """
 self._installSignalHandlers = installSignalHandlers
 ReactorBase.startRunning(self)
 
 def run(self, installSignalHandlers=True):
 self.startRunning(installSignalHandlers=installSignalHandlers)
 self.mainLoop()
 
 def mainLoop(self):
 while self._started:
 try:
 while self._started:
 # Advance simulation time in delayed event
 # processors.
 self.runUntilCurrent()
 t2 = self.timeout()
 t = self.running and t2
 # doIteration是关键，select,poll,epool实现各有不同
 self.doIteration(t)
 except:
 log.msg("Unexpected error in main loop.")
 log.err()
 else:
 log.msg('Main loop terminated.')

# twisted/internet/selectreactor.py
@implementer(IReactorFDSet)
class SelectReactor(posixbase.PosixReactorBase, _extraBase):
 
 def __init__(self):
 """
 Initialize file descriptor tracking dictionaries and the base class.
 """
 self._reads = set()
 self._writes = set()
 posixbase.PosixReactorBase.__init__(self)
 
 def doSelect(self, timeout):
 """
 Run one iteration of the I/O monitor loop.
 
 This will run all selectables who had input or output readiness
 waiting for them.
 """
 try:
 # 调用select方法监控读写集合，返回准备好读写的描述符
 r, w, ignored = _select(self._reads,
 self._writes,
 [], timeout)
 except ValueError:
 # Possibly a file descriptor has gone negative?
 self._preenDescriptors()
 return
 except TypeError:
 # Something *totally* invalid (object w/o fileno, non-integral
 # result) was passed
 log.err()
 self._preenDescriptors()
 return
 except (select.error, socket.error, IOError) as se:
 # select(2) encountered an error, perhaps while calling the fileno()
 # method of a socket. (Python 2.6 socket.error is an IOError
 # subclass, but on Python 2.5 and earlier it is not.)
 if se.args[0] in (0, 2):
 # windows does this if it got an empty list
 if (not self._reads) and (not self._writes):
 return
 else:
 raise
 elif se.args[0] == EINTR:
 return
 elif se.args[0] == EBADF:
 self._preenDescriptors()
 return
 else:
 # OK, I really don't know what's going on. Blow up.
 raise
 
 _drdw = self._doReadOrWrite
 _logrun = log.callWithLogger
 for selectables, method, fdset in ((r, "doRead", self._reads),
 (w,"doWrite", self._writes)):
 for selectable in selectables:
 # if this was disconnected in another thread, kill it.
 # ^^^^ --- what the !@#*? serious! -exarkun
 if selectable not in fdset:
 continue
 # This for pausing input when we're not ready for more.
 
 # 调用_doReadOrWrite方法
 _logrun(selectable, _drdw, selectable, method)
 
 doIteration = doSelect
 
 def _doReadOrWrite(self, selectable, method):
 try:
 # 调用method，doRead或者是doWrite，
 # 这里的selectable可能是我们监听的tcp.Port
 why = getattr(selectable, method)()
 except:
 why = sys.exc_info()[1]
 log.err()
 if why:
 self._disconnectSelectable(selectable, why, method=="doRead")

# twisted/internet/tcp.py
 
@implementer(interfaces.IListeningPort)
class Port(base.BasePort, _SocketCloser):
 
 def doRead(self):
 """Called when my socket is ready for reading.
 当套接字准备好读的时候调用
 
 This accepts a connection and calls self.protocol() to handle the
 wire-level protocol.
 """
 try:
 if platformType == "posix":
 numAccepts = self.numberAccepts
 else:
 numAccepts = 1
 for i in range(numAccepts):
 if self.disconnecting:
 return
 try:
 # 调用accept
 skt, addr = self.socket.accept()
 except socket.error as e:
 if e.args[0] in (EWOULDBLOCK, EAGAIN):
 self.numberAccepts = i
 break
 elif e.args[0] == EPERM:
 continue
 elif e.args[0] in (EMFILE, ENOBUFS, ENFILE, ENOMEM, ECONNABORTED):
 log.msg("Could not accept new connection (%s)" % (
 errorcode[e.args[0]],))
 break
 raise
 
 fdesc._setCloseOnExec(skt.fileno())
 protocol = self.factory.buildProtocol(self._buildAddr(addr))
 if protocol is None:
 skt.close()
 continue
 s = self.sessionno
 self.sessionno = s+1
 # transport初始化的过程中，会将自身假如到reactor的读集合中，那么当它准备
 # 好读的时候，就可以调用它的doRead方法读取客户端发过来的数据了
 transport = self.transport(skt, protocol, addr, self, s, self.reactor)
 protocol.makeConnection(transport)
 else:
 self.numberAccepts = self.numberAccepts+20
 except:
 log.deferr()

# twisted/internet/tcp.py
@implementer(interfaces.ITCPTransport, interfaces.ISystemHandle)
class Connection(_TLSConnectionMixin, abstract.FileDescriptor, _SocketCloser,
 _AbortingMixin):
 
 def doRead(self):
 try:
 # 接收数据
 data = self.socket.recv(self.bufferSize)
 except socket.error as se:
 if se.args[0] == EWOULDBLOCK:
 return
 else:
 return main.CONNECTION_LOST
 
 return self._dataReceived(data)
 
 def _dataReceived(self, data):
 if not data:
 return main.CONNECTION_DONE
 # 调用我们自定义protocol的dataReceived方法处理数据
 rval = self.protocol.dataReceived(data)
 if rval is not None:
 offender = self.protocol.dataReceived
 warningFormat = (
 'Returning a value other than None from %(fqpn)s is '
 'deprecated since %(version)s.')
 warningString = deprecate.getDeprecationWarningString(
 offender, versions.Version('Twisted', 11, 0, 0),
 format=warningFormat)
 deprecate.warnAboutFunction(offender, warningString)
 return rval