1 # -*- test-case-name: twisted.test.test_internet,twisted.internet.test.test_posixbase -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Posix reactor base class
15 from zope.interface import implements, classImplements
17 from twisted.python.compat import set
18 from twisted.internet.interfaces import IReactorUNIX, IReactorUNIXDatagram
19 from twisted.internet.interfaces import (
20 IReactorTCP, IReactorUDP, IReactorSSL, _IReactorArbitrary, IReactorSocket)
21 from twisted.internet.interfaces import IReactorProcess, IReactorMulticast
22 from twisted.internet.interfaces import IHalfCloseableDescriptor
23 from twisted.internet import error
24 from twisted.internet import tcp, udp
26 from twisted.python import log, failure, util
27 from twisted.persisted import styles
28 from twisted.python.runtime import platformType, platform
30 from twisted.internet.base import ReactorBase, _SignalReactorMixin
31 from twisted.internet.main import CONNECTION_DONE, CONNECTION_LOST
33 # Exceptions that doSelect might return frequently
34 _NO_FILENO = error.ConnectionFdescWentAway('Handler has no fileno method')
35 _NO_FILEDESC = error.ConnectionFdescWentAway('File descriptor lost')
39 from twisted.protocols import tls
43 from twisted.internet import ssl
48 from twisted.internet import unix
53 processEnabled = False
54 if platformType == 'posix':
55 from twisted.internet import fdesc, process, _signals
58 if platform.isWindows():
66 class _SocketWaker(log.Logger, styles.Ephemeral):
68 The I{self-pipe trick<http://cr.yp.to/docs/selfpipe.html>}, implemented
69 using a pair of sockets rather than pipes (due to the lack of support in
70 select() on Windows for pipes), used to wake up the main loop from
75 def __init__(self, reactor):
78 self.reactor = reactor
79 # Following select_trigger (from asyncore)'s example;
80 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
81 client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
82 client.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
83 server.bind(('127.0.0.1', 0))
85 client.connect(server.getsockname())
86 reader, clientaddr = server.accept()
91 self.fileno = self.r.fileno
94 """Send a byte to my connection.
97 util.untilConcludes(self.w.send, 'x')
98 except socket.error, (err, msg):
99 if err != errno.WSAEWOULDBLOCK:
103 """Read some data from my connection.
110 def connectionLost(self, reason):
116 class _FDWaker(object, log.Logger, styles.Ephemeral):
118 The I{self-pipe trick<http://cr.yp.to/docs/selfpipe.html>}, used to wake
119 up the main loop from another thread or a signal handler.
121 L{_FDWaker} is a base class for waker implementations based on
122 writing to a pipe being monitored by the reactor.
124 @ivar o: The file descriptor for the end of the pipe which can be
125 written to to wake up a reactor monitoring this waker.
127 @ivar i: The file descriptor which should be monitored in order to
128 be awoken by this waker.
135 def __init__(self, reactor):
138 self.reactor = reactor
139 self.i, self.o = os.pipe()
140 fdesc.setNonBlocking(self.i)
141 fdesc._setCloseOnExec(self.i)
142 fdesc.setNonBlocking(self.o)
143 fdesc._setCloseOnExec(self.o)
144 self.fileno = lambda: self.i
149 Read some bytes from the pipe and discard them.
151 fdesc.readFromFD(self.fileno(), lambda data: None)
154 def connectionLost(self, reason):
155 """Close both ends of my pipe.
157 if not hasattr(self, "o"):
159 for fd in self.i, self.o:
168 class _UnixWaker(_FDWaker):
170 This class provides a simple interface to wake up the event loop.
172 This is used by threads or signals to wake up the event loop.
176 """Write one byte to the pipe, and flush it.
178 # We don't use fdesc.writeToFD since we need to distinguish
179 # between EINTR (try again) and EAGAIN (do nothing).
180 if self.o is not None:
182 util.untilConcludes(os.write, self.o, 'x')
184 # XXX There is no unit test for raising the exception
185 # for other errnos. See #4285.
186 if e.errno != errno.EAGAIN:
191 if platformType == 'posix':
194 # Primarily Windows and Jython.
195 _Waker = _SocketWaker
198 class _SIGCHLDWaker(_FDWaker):
200 L{_SIGCHLDWaker} can wake up a reactor whenever C{SIGCHLD} is
203 @see: L{twisted.internet._signals}
205 def __init__(self, reactor):
206 _FDWaker.__init__(self, reactor)
211 Install the handler necessary to make this waker active.
213 _signals.installHandler(self.o)
218 Remove the handler which makes this waker active.
220 _signals.installHandler(-1)
225 Having woken up the reactor in response to receipt of
226 C{SIGCHLD}, reap the process which exited.
228 This is called whenever the reactor notices the waker pipe is
229 writeable, which happens soon after any call to the C{wakeUp}
232 _FDWaker.doRead(self)
233 process.reapAllProcesses()
238 class _DisconnectSelectableMixin(object):
240 Mixin providing the C{_disconnectSelectable} method.
243 def _disconnectSelectable(self, selectable, why, isRead, faildict={
244 error.ConnectionDone: failure.Failure(error.ConnectionDone()),
245 error.ConnectionLost: failure.Failure(error.ConnectionLost())
248 Utility function for disconnecting a selectable.
250 Supports half-close notification, isRead should be boolean indicating
251 whether error resulted from doRead().
253 self.removeReader(selectable)
254 f = faildict.get(why.__class__)
256 if (isRead and why.__class__ == error.ConnectionDone
257 and IHalfCloseableDescriptor.providedBy(selectable)):
258 selectable.readConnectionLost(f)
260 self.removeWriter(selectable)
261 selectable.connectionLost(f)
263 self.removeWriter(selectable)
264 selectable.connectionLost(failure.Failure(why))
268 class PosixReactorBase(_SignalReactorMixin, _DisconnectSelectableMixin,
271 A basis for reactors that use file descriptors.
273 @ivar _childWaker: C{None} or a reference to the L{_SIGCHLDWaker}
274 which is used to properly notice child process termination.
276 implements(_IReactorArbitrary, IReactorTCP, IReactorUDP, IReactorMulticast)
278 # Callable that creates a waker, overrideable so that subclasses can
279 # substitute their own implementation:
280 _wakerFactory = _Waker
282 def installWaker(self):
284 Install a `waker' to allow threads and signals to wake up the IO thread.
286 We use the self-pipe trick (http://cr.yp.to/docs/selfpipe.html) to wake
287 the reactor. On Windows we use a pair of sockets.
290 self.waker = self._wakerFactory(self)
291 self._internalReaders.add(self.waker)
292 self.addReader(self.waker)
296 def _handleSignals(self):
298 Extend the basic signal handling logic to also support
299 handling SIGCHLD to know when to try to reap child processes.
301 _SignalReactorMixin._handleSignals(self)
302 if platformType == 'posix':
303 if not self._childWaker:
304 self._childWaker = _SIGCHLDWaker(self)
305 self._internalReaders.add(self._childWaker)
306 self.addReader(self._childWaker)
307 self._childWaker.install()
308 # Also reap all processes right now, in case we missed any
309 # signals before we installed the SIGCHLD waker/handler.
310 # This should only happen if someone used spawnProcess
311 # before calling reactor.run (and the process also exited
313 process.reapAllProcesses()
315 def _uninstallHandler(self):
317 If a child waker was created and installed, uninstall it now.
319 Since this disables reactor functionality and is only called
320 when the reactor is stopping, it doesn't provide any directly
321 useful functionality, but the cleanup of reactor-related
322 process-global state that it does helps in unit tests
323 involving multiple reactors and is generally just a nice
326 # XXX This would probably be an alright place to put all of
327 # the cleanup code for all internal readers (here and in the
328 # base class, anyway). See #3063 for that cleanup task.
330 self._childWaker.uninstall()
334 def spawnProcess(self, processProtocol, executable, args=(),
336 uid=None, gid=None, usePTY=0, childFDs=None):
337 args, env = self._checkProcessArgs(args, env)
338 if platformType == 'posix':
340 if childFDs is not None:
341 raise ValueError("Using childFDs is not supported with usePTY=True.")
342 return process.PTYProcess(self, executable, args, env, path,
343 processProtocol, uid, gid, usePTY)
345 return process.Process(self, executable, args, env, path,
346 processProtocol, uid, gid, childFDs)
347 elif platformType == "win32":
349 raise ValueError("Setting UID is unsupported on this platform.")
351 raise ValueError("Setting GID is unsupported on this platform.")
353 raise ValueError("The usePTY parameter is not supported on Windows.")
355 raise ValueError("Customizing childFDs is not supported on Windows.")
358 from twisted.internet._dumbwin32proc import Process
359 return Process(self, processProtocol, executable, args, env, path)
361 raise NotImplementedError, "spawnProcess not available since pywin32 is not installed."
363 raise NotImplementedError, "spawnProcess only available on Windows or POSIX."
367 def listenUDP(self, port, protocol, interface='', maxPacketSize=8192):
368 """Connects a given L{DatagramProtocol} to the given numeric UDP port.
370 @returns: object conforming to L{IListeningPort}.
372 p = udp.Port(port, protocol, interface, maxPacketSize, self)
378 def listenMulticast(self, port, protocol, interface='', maxPacketSize=8192, listenMultiple=False):
379 """Connects a given DatagramProtocol to the given numeric UDP port.
383 @returns: object conforming to IListeningPort.
385 p = udp.MulticastPort(port, protocol, interface, maxPacketSize, self, listenMultiple)
392 def connectUNIX(self, address, factory, timeout=30, checkPID=0):
393 """@see: twisted.internet.interfaces.IReactorUNIX.connectUNIX
395 assert unixEnabled, "UNIX support is not present"
396 c = unix.Connector(address, factory, timeout, self, checkPID)
400 def listenUNIX(self, address, factory, backlog=50, mode=0666, wantPID=0):
402 @see: twisted.internet.interfaces.IReactorUNIX.listenUNIX
404 assert unixEnabled, "UNIX support is not present"
405 p = unix.Port(address, factory, backlog, mode, self, wantPID)
410 # IReactorUNIXDatagram
412 def listenUNIXDatagram(self, address, protocol, maxPacketSize=8192,
415 Connects a given L{DatagramProtocol} to the given path.
419 @returns: object conforming to L{IListeningPort}.
421 assert unixEnabled, "UNIX support is not present"
422 p = unix.DatagramPort(address, protocol, maxPacketSize, mode, self)
426 def connectUNIXDatagram(self, address, protocol, maxPacketSize=8192,
427 mode=0666, bindAddress=None):
429 Connects a L{ConnectedDatagramProtocol} instance to a path.
433 assert unixEnabled, "UNIX support is not present"
434 p = unix.ConnectedDatagramPort(address, protocol, maxPacketSize, mode, bindAddress, self)
439 # IReactorSocket (but not on Windows)
440 def adoptStreamPort(self, fileDescriptor, addressFamily, factory):
442 Create a new L{IListeningPort} from an already-initialized socket.
444 This just dispatches to a suitable port implementation (eg from
445 L{IReactorTCP}, etc) based on the specified C{addressFamily}.
447 @see: L{twisted.internet.interfaces.IReactorSocket.adoptStreamPort}
449 if addressFamily not in (socket.AF_INET, socket.AF_INET6):
450 raise error.UnsupportedAddressFamily(addressFamily)
452 p = tcp.Port._fromListeningDescriptor(
453 self, fileDescriptor, addressFamily, factory)
460 def listenTCP(self, port, factory, backlog=50, interface=''):
461 """@see: twisted.internet.interfaces.IReactorTCP.listenTCP
463 p = tcp.Port(port, factory, backlog, interface, self)
467 def connectTCP(self, host, port, factory, timeout=30, bindAddress=None):
468 """@see: twisted.internet.interfaces.IReactorTCP.connectTCP
470 c = tcp.Connector(host, port, factory, timeout, bindAddress, self)
474 # IReactorSSL (sometimes, not implemented)
476 def connectSSL(self, host, port, factory, contextFactory, timeout=30, bindAddress=None):
477 """@see: twisted.internet.interfaces.IReactorSSL.connectSSL
480 tlsFactory = tls.TLSMemoryBIOFactory(contextFactory, True, factory)
481 return self.connectTCP(host, port, tlsFactory, timeout, bindAddress)
482 elif ssl is not None:
484 host, port, factory, contextFactory, timeout, bindAddress, self)
488 assert False, "SSL support is not present"
492 def listenSSL(self, port, factory, contextFactory, backlog=50, interface=''):
493 """@see: twisted.internet.interfaces.IReactorSSL.listenSSL
496 tlsFactory = tls.TLSMemoryBIOFactory(contextFactory, False, factory)
497 port = self.listenTCP(port, tlsFactory, backlog, interface)
500 elif ssl is not None:
502 port, factory, contextFactory, backlog, interface, self)
506 assert False, "SSL support is not present"
510 def listenWith(self, portType, *args, **kw):
512 "listenWith is deprecated since Twisted 10.1. "
513 "See IReactorFDSet.",
514 category=DeprecationWarning,
517 p = portType(*args, **kw)
522 def connectWith(self, connectorType, *args, **kw):
524 "connectWith is deprecated since Twisted 10.1. "
525 "See IReactorFDSet.",
526 category=DeprecationWarning,
529 c = connectorType(*args, **kw)
534 def _removeAll(self, readers, writers):
536 Remove all readers and writers, and list of removed L{IReadDescriptor}s
537 and L{IWriteDescriptor}s.
539 Meant for calling from subclasses, to implement removeAll, like::
542 return self._removeAll(self._reads, self._writes)
544 where C{self._reads} and C{self._writes} are iterables.
546 removedReaders = set(readers) - self._internalReaders
547 for reader in removedReaders:
548 self.removeReader(reader)
550 removedWriters = set(writers)
551 for writer in removedWriters:
552 self.removeWriter(writer)
554 return list(removedReaders | removedWriters)
557 class _PollLikeMixin(object):
559 Mixin for poll-like reactors.
561 Subclasses must define the following attributes::
563 - _POLL_DISCONNECTED - Bitmask for events indicating a connection was
565 - _POLL_IN - Bitmask for events indicating there is input to read.
566 - _POLL_OUT - Bitmask for events indicating output can be written.
568 Must be mixed in to a subclass of PosixReactorBase (for
569 _disconnectSelectable).
572 def _doReadOrWrite(self, selectable, fd, event):
574 fd is available for read or write, do the work and raise errors if
579 if event & self._POLL_DISCONNECTED and not (event & self._POLL_IN):
580 # Handle disconnection. But only if we finished processing all
582 if fd in self._reads:
583 # If we were reading from the descriptor then this is a
584 # clean shutdown. We know there are no read events pending
585 # because we just checked above. It also might be a
586 # half-close (which is why we have to keep track of inRead).
588 why = CONNECTION_DONE
590 # If we weren't reading, this is an error shutdown of some
592 why = CONNECTION_LOST
594 # Any non-disconnect event turns into a doRead or a doWrite.
596 # First check to see if the descriptor is still valid. This
597 # gives fileno() a chance to raise an exception, too.
598 # Ideally, disconnection would always be indicated by the
599 # return value of doRead or doWrite (or an exception from
600 # one of those methods), but calling fileno here helps make
601 # buggy applications more transparent.
602 if selectable.fileno() == -1:
603 # -1 is sort of a historical Python artifact. Python
604 # files and sockets used to change their file descriptor
605 # to -1 when they closed. For the time being, we'll
606 # continue to support this anyway in case applications
607 # replicated it, plus abstract.FileDescriptor.fileno
608 # returns -1. Eventually it'd be good to deprecate this
612 if event & self._POLL_IN:
613 # Handle a read event.
614 why = selectable.doRead()
616 if not why and event & self._POLL_OUT:
617 # Handle a write event, as long as doRead didn't
619 why = selectable.doWrite()
622 # Any exception from application code gets logged and will
623 # cause us to disconnect the selectable.
624 why = sys.exc_info()[1]
627 self._disconnectSelectable(selectable, why, inRead)
631 if tls is not None or ssl is not None:
632 classImplements(PosixReactorBase, IReactorSSL)
634 classImplements(PosixReactorBase, IReactorUNIX, IReactorUNIXDatagram)
636 classImplements(PosixReactorBase, IReactorProcess)
637 if getattr(socket, 'fromfd', None) is not None:
638 classImplements(PosixReactorBase, IReactorSocket)
640 __all__ = ["PosixReactorBase"]