selectreactor.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. # -*- test-case-name: twisted.test.test_internet -*-
  2. # Copyright (c) Twisted Matrix Laboratories.
  3. # See LICENSE for details.
  4. """
  5. Select reactor
  6. """
  7. import select
  8. import sys
  9. from errno import EBADF, EINTR
  10. from time import sleep
  11. from typing import Type
  12. from zope.interface import implementer
  13. from twisted.internet import posixbase
  14. from twisted.internet.interfaces import IReactorFDSet
  15. from twisted.python import log
  16. from twisted.python.runtime import platformType
  17. def win32select(r, w, e, timeout=None):
  18. """Win32 select wrapper."""
  19. if not (r or w):
  20. # windows select() exits immediately when no sockets
  21. if timeout is None:
  22. timeout = 0.01
  23. else:
  24. timeout = min(timeout, 0.001)
  25. sleep(timeout)
  26. return [], [], []
  27. # windows doesn't process 'signals' inside select(), so we set a max
  28. # time or ctrl-c will never be recognized
  29. if timeout is None or timeout > 0.5:
  30. timeout = 0.5
  31. r, w, e = select.select(r, w, w, timeout)
  32. return r, w + e, []
  33. if platformType == "win32":
  34. _select = win32select
  35. else:
  36. _select = select.select
  37. try:
  38. from twisted.internet.win32eventreactor import _ThreadedWin32EventsMixin
  39. except ImportError:
  40. _extraBase: Type[object] = object
  41. else:
  42. _extraBase = _ThreadedWin32EventsMixin
  43. @implementer(IReactorFDSet)
  44. class SelectReactor(posixbase.PosixReactorBase, _extraBase): # type: ignore[misc,valid-type]
  45. """
  46. A select() based reactor - runs on all POSIX platforms and on Win32.
  47. @ivar _reads: A set containing L{FileDescriptor} instances which will be
  48. checked for read events.
  49. @ivar _writes: A set containing L{FileDescriptor} instances which will be
  50. checked for writability.
  51. """
  52. def __init__(self):
  53. """
  54. Initialize file descriptor tracking dictionaries and the base class.
  55. """
  56. self._reads = set()
  57. self._writes = set()
  58. posixbase.PosixReactorBase.__init__(self)
  59. def _preenDescriptors(self):
  60. log.msg("Malformed file descriptor found. Preening lists.")
  61. readers = list(self._reads)
  62. writers = list(self._writes)
  63. self._reads.clear()
  64. self._writes.clear()
  65. for selSet, selList in ((self._reads, readers), (self._writes, writers)):
  66. for selectable in selList:
  67. try:
  68. select.select([selectable], [selectable], [selectable], 0)
  69. except Exception as e:
  70. log.msg("bad descriptor %s" % selectable)
  71. self._disconnectSelectable(selectable, e, False)
  72. else:
  73. selSet.add(selectable)
  74. def doSelect(self, timeout):
  75. """
  76. Run one iteration of the I/O monitor loop.
  77. This will run all selectables who had input or output readiness
  78. waiting for them.
  79. """
  80. try:
  81. r, w, ignored = _select(self._reads, self._writes, [], timeout)
  82. except ValueError:
  83. # Possibly a file descriptor has gone negative?
  84. self._preenDescriptors()
  85. return
  86. except TypeError:
  87. # Something *totally* invalid (object w/o fileno, non-integral
  88. # result) was passed
  89. log.err()
  90. self._preenDescriptors()
  91. return
  92. except OSError as se:
  93. # select(2) encountered an error, perhaps while calling the fileno()
  94. # method of a socket. (Python 2.6 socket.error is an IOError
  95. # subclass, but on Python 2.5 and earlier it is not.)
  96. if se.args[0] in (0, 2):
  97. # windows does this if it got an empty list
  98. if (not self._reads) and (not self._writes):
  99. return
  100. else:
  101. raise
  102. elif se.args[0] == EINTR:
  103. return
  104. elif se.args[0] == EBADF:
  105. self._preenDescriptors()
  106. return
  107. else:
  108. # OK, I really don't know what's going on. Blow up.
  109. raise
  110. _drdw = self._doReadOrWrite
  111. _logrun = log.callWithLogger
  112. for selectables, method, fdset in (
  113. (r, "doRead", self._reads),
  114. (w, "doWrite", self._writes),
  115. ):
  116. for selectable in selectables:
  117. # if this was disconnected in another thread, kill it.
  118. # ^^^^ --- what the !@#*? serious! -exarkun
  119. if selectable not in fdset:
  120. continue
  121. # This for pausing input when we're not ready for more.
  122. _logrun(selectable, _drdw, selectable, method)
  123. doIteration = doSelect
  124. def _doReadOrWrite(self, selectable, method):
  125. try:
  126. why = getattr(selectable, method)()
  127. except BaseException:
  128. why = sys.exc_info()[1]
  129. log.err()
  130. if why:
  131. self._disconnectSelectable(selectable, why, method == "doRead")
  132. def addReader(self, reader):
  133. """
  134. Add a FileDescriptor for notification of data available to read.
  135. """
  136. self._reads.add(reader)
  137. def addWriter(self, writer):
  138. """
  139. Add a FileDescriptor for notification of data available to write.
  140. """
  141. self._writes.add(writer)
  142. def removeReader(self, reader):
  143. """
  144. Remove a Selectable for notification of data available to read.
  145. """
  146. self._reads.discard(reader)
  147. def removeWriter(self, writer):
  148. """
  149. Remove a Selectable for notification of data available to write.
  150. """
  151. self._writes.discard(writer)
  152. def removeAll(self):
  153. return self._removeAll(self._reads, self._writes)
  154. def getReaders(self):
  155. return list(self._reads)
  156. def getWriters(self):
  157. return list(self._writes)
  158. def install():
  159. """Configure the twisted mainloop to be run using the select() reactor."""
  160. reactor = SelectReactor()
  161. from twisted.internet.main import installReactor
  162. installReactor(reactor)
  163. __all__ = ["install"]