twisted/internet/selectreactor.py

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