1 # -*- test-case-name: twisted.test.test_internet -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Very basic functionality for a Reactor implementation.
9 import socket # needed only for sync-dns
10 from zope.interface import implements, classImplements
14 from heapq import heappush, heappop, heapify
18 from twisted.python.compat import set
19 from twisted.python.util import unsignedID
20 from twisted.internet.interfaces import IReactorCore, IReactorTime, IReactorThreads
21 from twisted.internet.interfaces import IResolverSimple, IReactorPluggableResolver
22 from twisted.internet.interfaces import IConnector, IDelayedCall
23 from twisted.internet import fdesc, main, error, abstract, defer, threads
24 from twisted.python import log, failure, reflect
25 from twisted.python.runtime import seconds as runtimeSeconds, platform
26 from twisted.internet.defer import Deferred, DeferredList
27 from twisted.persisted import styles
29 # This import is for side-effects! Even if you don't see any code using it
30 # in this module, don't delete it.
31 from twisted.python import threadable
34 class DelayedCall(styles.Ephemeral):
36 implements(IDelayedCall)
37 # enable .debug to record creator call stack, and it will be logged if
38 # an exception occurs while the function is being run
42 def __init__(self, time, func, args, kw, cancel, reset,
43 seconds=runtimeSeconds):
45 @param time: Seconds from the epoch at which to call C{func}.
46 @param func: The callable to call.
47 @param args: The positional arguments to pass to the callable.
48 @param kw: The keyword arguments to pass to the callable.
49 @param cancel: A callable which will be called with this
50 DelayedCall before cancellation.
51 @param reset: A callable which will be called with this
52 DelayedCall after changing this DelayedCall's scheduled
53 execution time. The callable should adjust any necessary
54 scheduling details to ensure this DelayedCall is invoked
55 at the new appropriate time.
56 @param seconds: If provided, a no-argument callable which will be
57 used to determine the current time any time that information is
60 self.time, self.func, self.args, self.kw = time, func, args, kw
62 self.canceller = cancel
63 self.seconds = seconds
64 self.cancelled = self.called = 0
67 self.creator = traceback.format_stack()[:-2]
70 """Return the time at which this call will fire
73 @return: The number of seconds after the epoch at which this call is
76 return self.time + self.delayed_time
79 """Unschedule this call
81 @raise AlreadyCancelled: Raised if this call has already been
84 @raise AlreadyCalled: Raised if this call has already been made.
87 raise error.AlreadyCancelled
89 raise error.AlreadyCalled
95 del self.func, self.args, self.kw
97 def reset(self, secondsFromNow):
98 """Reschedule this call for a different time
100 @type secondsFromNow: C{float}
101 @param secondsFromNow: The number of seconds from the time of the
102 C{reset} call at which this call will be scheduled.
104 @raise AlreadyCancelled: Raised if this call has been cancelled.
105 @raise AlreadyCalled: Raised if this call has already been made.
108 raise error.AlreadyCancelled
110 raise error.AlreadyCalled
112 newTime = self.seconds() + secondsFromNow
113 if newTime < self.time:
114 self.delayed_time = 0
118 self.delayed_time = newTime - self.time
120 def delay(self, secondsLater):
121 """Reschedule this call for a later time
123 @type secondsLater: C{float}
124 @param secondsLater: The number of seconds after the originally
125 scheduled time for which to reschedule this call.
127 @raise AlreadyCancelled: Raised if this call has been cancelled.
128 @raise AlreadyCalled: Raised if this call has already been made.
131 raise error.AlreadyCancelled
133 raise error.AlreadyCalled
135 self.delayed_time += secondsLater
136 if self.delayed_time < 0:
137 self.activate_delay()
140 def activate_delay(self):
141 self.time += self.delayed_time
142 self.delayed_time = 0
145 """Determine whether this call is still pending
148 @return: True if this call has not yet been made or cancelled,
151 return not (self.cancelled or self.called)
154 def __le__(self, other):
156 Implement C{<=} operator between two L{DelayedCall} instances.
158 Comparison is based on the C{time} attribute (unadjusted by the
161 return self.time <= other.time
164 def __lt__(self, other):
166 Implement C{<} operator between two L{DelayedCall} instances.
168 Comparison is based on the C{time} attribute (unadjusted by the
171 return self.time < other.time
175 if self._str is not None:
177 if hasattr(self, 'func'):
178 if hasattr(self.func, 'func_name'):
179 func = self.func.func_name
180 if hasattr(self.func, 'im_class'):
181 func = self.func.im_class.__name__ + '.' + func
183 func = reflect.safe_repr(self.func)
188 L = ["<DelayedCall 0x%x [%ss] called=%s cancelled=%s" % (
189 unsignedID(self), self.time - now, self.called,
192 L.extend((" ", func, "("))
194 L.append(", ".join([reflect.safe_repr(e) for e in self.args]))
198 L.append(", ".join(['%s=%s' % (k, reflect.safe_repr(v)) for (k, v) in self.kw.iteritems()]))
202 L.append("\n\ntraceback at creation: \n\n%s" % (' '.join(self.creator)))
209 class ThreadedResolver(object):
211 L{ThreadedResolver} uses a reactor, a threadpool, and
212 L{socket.gethostbyname} to perform name lookups without blocking the
213 reactor thread. It also supports timeouts indepedently from whatever
214 timeout logic L{socket.gethostbyname} might have.
216 @ivar reactor: The reactor the threadpool of which will be used to call
217 L{socket.gethostbyname} and the I/O thread of which the result will be
220 implements(IResolverSimple)
222 def __init__(self, reactor):
223 self.reactor = reactor
224 self._runningQueries = {}
227 def _fail(self, name, err):
228 err = error.DNSLookupError("address %r not found: %s" % (name, err))
229 return failure.Failure(err)
232 def _cleanup(self, name, lookupDeferred):
233 userDeferred, cancelCall = self._runningQueries[lookupDeferred]
234 del self._runningQueries[lookupDeferred]
235 userDeferred.errback(self._fail(name, "timeout error"))
238 def _checkTimeout(self, result, name, lookupDeferred):
240 userDeferred, cancelCall = self._runningQueries[lookupDeferred]
244 del self._runningQueries[lookupDeferred]
247 if isinstance(result, failure.Failure):
248 userDeferred.errback(self._fail(name, result.getErrorMessage()))
250 userDeferred.callback(result)
253 def getHostByName(self, name, timeout = (1, 3, 11, 45)):
255 See L{twisted.internet.interfaces.IResolverSimple.getHostByName}.
257 Note that the elements of C{timeout} are summed and the result is used
258 as a timeout for the lookup. Any intermediate timeout or retry logic
259 is left up to the platform via L{socket.gethostbyname}.
262 timeoutDelay = sum(timeout)
265 userDeferred = defer.Deferred()
266 lookupDeferred = threads.deferToThreadPool(
267 self.reactor, self.reactor.getThreadPool(),
268 socket.gethostbyname, name)
269 cancelCall = self.reactor.callLater(
270 timeoutDelay, self._cleanup, name, lookupDeferred)
271 self._runningQueries[lookupDeferred] = (userDeferred, cancelCall)
272 lookupDeferred.addBoth(self._checkTimeout, name, lookupDeferred)
277 class BlockingResolver:
278 implements(IResolverSimple)
280 def getHostByName(self, name, timeout = (1, 3, 11, 45)):
282 address = socket.gethostbyname(name)
284 msg = "address %r not found" % (name,)
285 err = error.DNSLookupError(msg)
286 return defer.fail(err)
288 return defer.succeed(address)
291 class _ThreePhaseEvent(object):
293 Collection of callables (with arguments) which can be invoked as a group in
296 This provides the underlying implementation for the reactor's system event
297 triggers. An instance of this class tracks triggers for all phases of a
298 single type of event.
300 @ivar before: A list of the before-phase triggers containing three-tuples
301 of a callable, a tuple of positional arguments, and a dict of keyword
304 @ivar finishedBefore: A list of the before-phase triggers which have
305 already been executed. This is only populated in the C{'BEFORE'} state.
307 @ivar during: A list of the during-phase triggers containing three-tuples
308 of a callable, a tuple of positional arguments, and a dict of keyword
311 @ivar after: A list of the after-phase triggers containing three-tuples
312 of a callable, a tuple of positional arguments, and a dict of keyword
315 @ivar state: A string indicating what is currently going on with this
316 object. One of C{'BASE'} (for when nothing in particular is happening;
317 this is the initial value), C{'BEFORE'} (when the before-phase triggers
318 are in the process of being executed).
327 def addTrigger(self, phase, callable, *args, **kwargs):
329 Add a trigger to the indicate phase.
331 @param phase: One of C{'before'}, C{'during'}, or C{'after'}.
333 @param callable: An object to be called when this event is triggered.
334 @param *args: Positional arguments to pass to C{callable}.
335 @param **kwargs: Keyword arguments to pass to C{callable}.
337 @return: An opaque handle which may be passed to L{removeTrigger} to
338 reverse the effects of calling this method.
340 if phase not in ('before', 'during', 'after'):
341 raise KeyError("invalid phase")
342 getattr(self, phase).append((callable, args, kwargs))
343 return phase, callable, args, kwargs
346 def removeTrigger(self, handle):
348 Remove a previously added trigger callable.
350 @param handle: An object previously returned by L{addTrigger}. The
351 trigger added by that call will be removed.
353 @raise ValueError: If the trigger associated with C{handle} has already
354 been removed or if C{handle} is not a valid handle.
356 return getattr(self, 'removeTrigger_' + self.state)(handle)
359 def removeTrigger_BASE(self, handle):
361 Just try to remove the trigger.
366 phase, callable, args, kwargs = handle
367 except (TypeError, ValueError):
368 raise ValueError("invalid trigger handle")
370 if phase not in ('before', 'during', 'after'):
371 raise KeyError("invalid phase")
372 getattr(self, phase).remove((callable, args, kwargs))
375 def removeTrigger_BEFORE(self, handle):
377 Remove the trigger if it has yet to be executed, otherwise emit a
378 warning that in the future an exception will be raised when removing an
379 already-executed trigger.
383 phase, callable, args, kwargs = handle
384 if phase != 'before':
385 return self.removeTrigger_BASE(handle)
386 if (callable, args, kwargs) in self.finishedBefore:
388 "Removing already-fired system event triggers will raise an "
389 "exception in a future version of Twisted.",
390 category=DeprecationWarning,
393 self.removeTrigger_BASE(handle)
398 Call the triggers added to this event.
400 self.state = 'BEFORE'
401 self.finishedBefore = []
404 callable, args, kwargs = self.before.pop(0)
405 self.finishedBefore.append((callable, args, kwargs))
407 result = callable(*args, **kwargs)
411 if isinstance(result, Deferred):
412 beforeResults.append(result)
413 DeferredList(beforeResults).addCallback(self._continueFiring)
416 def _continueFiring(self, ignored):
418 Call the during and after phase triggers for this event.
421 self.finishedBefore = []
422 for phase in self.during, self.after:
424 callable, args, kwargs = phase.pop(0)
426 callable(*args, **kwargs)
432 class ReactorBase(object):
434 Default base class for Reactors.
436 @type _stopped: C{bool}
437 @ivar _stopped: A flag which is true between paired calls to C{reactor.run}
438 and C{reactor.stop}. This should be replaced with an explicit state
441 @type _justStopped: C{bool}
442 @ivar _justStopped: A flag which is true between the time C{reactor.stop}
443 is called and the time the shutdown system event is fired. This is
444 used to determine whether that event should be fired after each
445 iteration through the mainloop. This should be replaced with an
446 explicit state machine.
448 @type _started: C{bool}
449 @ivar _started: A flag which is true from the time C{reactor.run} is called
450 until the time C{reactor.run} returns. This is used to prevent calls
451 to C{reactor.run} on a running reactor. This should be replaced with
452 an explicit state machine.
454 @ivar running: See L{IReactorCore.running}
456 @ivar _registerAsIOThread: A flag controlling whether the reactor will
457 register the thread it is running in as the I/O thread when it starts.
458 If C{True}, registration will be done, otherwise it will not be.
460 implements(IReactorCore, IReactorTime, IReactorPluggableResolver)
462 _registerAsIOThread = True
467 resolver = BlockingResolver()
469 __name__ = "twisted.internet.reactor"
472 self.threadCallQueue = []
473 self._eventTriggers = {}
474 self._pendingTimedCalls = []
475 self._newTimedCalls = []
476 self._cancellations = 0
478 self._started = False
479 self._justStopped = False
480 self._startedBefore = False
481 # reactor internal readers, e.g. the waker.
482 self._internalReaders = set()
485 # Arrange for the running attribute to change to True at the right time
486 # and let a subclass possibly do other things at that time (eg install
488 self.addSystemEventTrigger(
489 'during', 'startup', self._reallyStartRunning)
490 self.addSystemEventTrigger('during', 'shutdown', self.crash)
491 self.addSystemEventTrigger('during', 'shutdown', self.disconnectAll)
493 if platform.supportsThreads():
497 # override in subclasses
501 def installWaker(self):
502 raise NotImplementedError(
503 reflect.qual(self.__class__) + " did not implement installWaker")
505 def installResolver(self, resolver):
506 assert IResolverSimple.providedBy(resolver)
507 oldResolver = self.resolver
508 self.resolver = resolver
513 Wake up the event loop.
517 # if the waker isn't installed, the reactor isn't running, and
518 # therefore doesn't need to be woken up
520 def doIteration(self, delay):
522 Do one iteration over the readers and writers which have been added.
524 raise NotImplementedError(
525 reflect.qual(self.__class__) + " did not implement doIteration")
527 def addReader(self, reader):
528 raise NotImplementedError(
529 reflect.qual(self.__class__) + " did not implement addReader")
531 def addWriter(self, writer):
532 raise NotImplementedError(
533 reflect.qual(self.__class__) + " did not implement addWriter")
535 def removeReader(self, reader):
536 raise NotImplementedError(
537 reflect.qual(self.__class__) + " did not implement removeReader")
539 def removeWriter(self, writer):
540 raise NotImplementedError(
541 reflect.qual(self.__class__) + " did not implement removeWriter")
544 raise NotImplementedError(
545 reflect.qual(self.__class__) + " did not implement removeAll")
548 def getReaders(self):
549 raise NotImplementedError(
550 reflect.qual(self.__class__) + " did not implement getReaders")
553 def getWriters(self):
554 raise NotImplementedError(
555 reflect.qual(self.__class__) + " did not implement getWriters")
558 def resolve(self, name, timeout = (1, 3, 11, 45)):
559 """Return a Deferred that will resolve a hostname.
562 # XXX - This is *less than* '::', and will screw up IPv6 servers
563 return defer.succeed('0.0.0.0')
564 if abstract.isIPAddress(name):
565 return defer.succeed(name)
566 return self.resolver.getHostByName(name, timeout)
573 See twisted.internet.interfaces.IReactorCore.stop.
576 raise error.ReactorNotRunning(
577 "Can't stop reactor that isn't running.")
579 self._justStopped = True
580 self._startedBefore = True
585 See twisted.internet.interfaces.IReactorCore.crash.
587 Reset reactor state tracking attributes and re-initialize certain
588 state-transition helpers which were set up in C{__init__} but later
589 destroyed (through use).
591 self._started = False
593 self.addSystemEventTrigger(
594 'during', 'startup', self._reallyStartRunning)
596 def sigInt(self, *args):
597 """Handle a SIGINT interrupt.
599 log.msg("Received SIGINT, shutting down.")
600 self.callFromThread(self.stop)
602 def sigBreak(self, *args):
603 """Handle a SIGBREAK interrupt.
605 log.msg("Received SIGBREAK, shutting down.")
606 self.callFromThread(self.stop)
608 def sigTerm(self, *args):
609 """Handle a SIGTERM interrupt.
611 log.msg("Received SIGTERM, shutting down.")
612 self.callFromThread(self.stop)
614 def disconnectAll(self):
615 """Disconnect every reader, and writer in the system.
617 selectables = self.removeAll()
618 for reader in selectables:
619 log.callWithLogger(reader,
620 reader.connectionLost,
621 failure.Failure(main.CONNECTION_LOST))
624 def iterate(self, delay=0):
625 """See twisted.internet.interfaces.IReactorCore.iterate.
627 self.runUntilCurrent()
628 self.doIteration(delay)
631 def fireSystemEvent(self, eventType):
632 """See twisted.internet.interfaces.IReactorCore.fireSystemEvent.
634 event = self._eventTriggers.get(eventType)
635 if event is not None:
639 def addSystemEventTrigger(self, _phase, _eventType, _f, *args, **kw):
640 """See twisted.internet.interfaces.IReactorCore.addSystemEventTrigger.
642 assert callable(_f), "%s is not callable" % _f
643 if _eventType not in self._eventTriggers:
644 self._eventTriggers[_eventType] = _ThreePhaseEvent()
645 return (_eventType, self._eventTriggers[_eventType].addTrigger(
646 _phase, _f, *args, **kw))
649 def removeSystemEventTrigger(self, triggerID):
650 """See twisted.internet.interfaces.IReactorCore.removeSystemEventTrigger.
652 eventType, handle = triggerID
653 self._eventTriggers[eventType].removeTrigger(handle)
656 def callWhenRunning(self, _callable, *args, **kw):
657 """See twisted.internet.interfaces.IReactorCore.callWhenRunning.
660 _callable(*args, **kw)
662 return self.addSystemEventTrigger('after', 'startup',
663 _callable, *args, **kw)
665 def startRunning(self):
667 Method called when reactor starts: do some initialization and fire
670 Don't call this directly, call reactor.run() instead: it should take
671 care of calling this.
673 This method is somewhat misnamed. The reactor will not necessarily be
674 in the running state by the time this method returns. The only
675 guarantee is that it will be on its way to the running state.
678 raise error.ReactorAlreadyRunning()
679 if self._startedBefore:
680 raise error.ReactorNotRestartable()
682 self._stopped = False
683 if self._registerAsIOThread:
684 threadable.registerAsIOThread()
685 self.fireSystemEvent('startup')
688 def _reallyStartRunning(self):
690 Method called to transition to the running state. This should happen
691 in the I{during startup} event trigger phase.
697 seconds = staticmethod(runtimeSeconds)
699 def callLater(self, _seconds, _f, *args, **kw):
700 """See twisted.internet.interfaces.IReactorTime.callLater.
702 assert callable(_f), "%s is not callable" % _f
703 assert sys.maxint >= _seconds >= 0, \
704 "%s is not greater than or equal to 0 seconds" % (_seconds,)
705 tple = DelayedCall(self.seconds() + _seconds, _f, args, kw,
706 self._cancelCallLater,
707 self._moveCallLaterSooner,
708 seconds=self.seconds)
709 self._newTimedCalls.append(tple)
712 def _moveCallLaterSooner(self, tple):
713 # Linear time find: slow.
714 heap = self._pendingTimedCalls
716 pos = heap.index(tple)
718 # Move elt up the heap until it rests at the right place.
721 parent = (pos-1) // 2
722 if heap[parent] <= elt:
725 heap[pos] = heap[parent]
729 # element was not found in heap - oh well...
732 def _cancelCallLater(self, tple):
733 self._cancellations+=1
736 def getDelayedCalls(self):
737 """Return all the outstanding delayed calls in the system.
738 They are returned in no particular order.
739 This method is not efficient -- it is really only meant for
741 return [x for x in (self._pendingTimedCalls + self._newTimedCalls) if not x.cancelled]
743 def _insertNewDelayedCalls(self):
744 for call in self._newTimedCalls:
746 self._cancellations-=1
748 call.activate_delay()
749 heappush(self._pendingTimedCalls, call)
750 self._newTimedCalls = []
753 # insert new delayed calls to make sure to include them in timeout value
754 self._insertNewDelayedCalls()
756 if not self._pendingTimedCalls:
759 return max(0, self._pendingTimedCalls[0].time - self.seconds())
762 def runUntilCurrent(self):
763 """Run all pending timed calls.
765 if self.threadCallQueue:
766 # Keep track of how many calls we actually make, as we're
767 # making them, in case another call is added to the queue
768 # while we're in this loop.
770 total = len(self.threadCallQueue)
771 for (f, a, kw) in self.threadCallQueue:
779 del self.threadCallQueue[:count]
780 if self.threadCallQueue:
783 # insert new delayed calls now
784 self._insertNewDelayedCalls()
787 while self._pendingTimedCalls and (self._pendingTimedCalls[0].time <= now):
788 call = heappop(self._pendingTimedCalls)
790 self._cancellations-=1
793 if call.delayed_time > 0:
794 call.activate_delay()
795 heappush(self._pendingTimedCalls, call)
800 call.func(*call.args, **call.kw)
803 if hasattr(call, "creator"):
805 e += " C: previous exception occurred in " + \
806 "a DelayedCall created here:\n"
808 e += "".join(call.creator).rstrip().replace("\n","\n C:")
813 if (self._cancellations > 50 and
814 self._cancellations > len(self._pendingTimedCalls) >> 1):
815 self._cancellations = 0
816 self._pendingTimedCalls = [x for x in self._pendingTimedCalls
818 heapify(self._pendingTimedCalls)
820 if self._justStopped:
821 self._justStopped = False
822 self.fireSystemEvent("shutdown")
826 def _checkProcessArgs(self, args, env):
828 Check for valid arguments and environment to spawnProcess.
830 @return: A two element tuple giving values to use when creating the
831 process. The first element of the tuple is a C{list} of C{str}
832 giving the values for argv of the child process. The second element
833 of the tuple is either C{None} if C{env} was C{None} or a C{dict}
834 mapping C{str} environment keys to C{str} environment values.
836 # Any unicode string which Python would successfully implicitly
837 # encode to a byte string would have worked before these explicit
838 # checks were added. Anything which would have failed with a
839 # UnicodeEncodeError during that implicit encoding step would have
840 # raised an exception in the child process and that would have been
841 # a pain in the butt to debug.
843 # So, we will explicitly attempt the same encoding which Python
844 # would implicitly do later. If it fails, we will report an error
845 # without ever spawning a child process. If it succeeds, we'll save
846 # the result so that Python doesn't need to do it implicitly later.
848 # For any unicode which we can actually encode, we'll also issue a
849 # deprecation warning, because no one should be passing unicode here
853 defaultEncoding = sys.getdefaultencoding()
855 # Common check function
858 Return either a str or None. If the given value is not
859 allowable for some reason, None is returned. Otherwise, a
860 possibly different object which should be used in place of arg
861 is returned. This forces unicode encoding to happen now, rather
862 than implicitly later.
864 if isinstance(arg, unicode):
866 arg = arg.encode(defaultEncoding)
867 except UnicodeEncodeError:
870 "Argument strings and environment keys/values passed to "
871 "reactor.spawnProcess should be str, not unicode.",
872 category=DeprecationWarning,
874 if isinstance(arg, str) and '\0' not in arg:
878 # Make a few tests to check input validity
879 if not isinstance(args, (tuple, list)):
880 raise TypeError("Arguments must be a tuple or list")
884 arg = argChecker(arg)
886 raise TypeError("Arguments contain a non-string value")
888 outputArgs.append(arg)
893 for key, val in env.iteritems():
894 key = argChecker(key)
896 raise TypeError("Environment contains a non-string key")
897 val = argChecker(val)
899 raise TypeError("Environment contains a non-string value")
901 return outputArgs, outputEnv
904 if platform.supportsThreads():
906 # ID of the trigger starting the threadpool
907 _threadpoolStartupID = None
908 # ID of the trigger stopping the threadpool
909 threadpoolShutdownID = None
911 def _initThreads(self):
912 self.usingThreads = True
913 self.resolver = ThreadedResolver(self)
915 def callFromThread(self, f, *args, **kw):
917 See L{twisted.internet.interfaces.IReactorThreads.callFromThread}.
919 assert callable(f), "%s is not callable" % (f,)
920 # lists are thread-safe in CPython, but not in Jython
921 # this is probably a bug in Jython, but until fixed this code
922 # won't work in Jython.
923 self.threadCallQueue.append((f, args, kw))
926 def _initThreadPool(self):
928 Create the threadpool accessible with callFromThread.
930 from twisted.python import threadpool
931 self.threadpool = threadpool.ThreadPool(
932 0, 10, 'twisted.internet.reactor')
933 self._threadpoolStartupID = self.callWhenRunning(
934 self.threadpool.start)
935 self.threadpoolShutdownID = self.addSystemEventTrigger(
936 'during', 'shutdown', self._stopThreadPool)
938 def _uninstallHandler(self):
941 def _stopThreadPool(self):
943 Stop the reactor threadpool. This method is only valid if there
944 is currently a threadpool (created by L{_initThreadPool}). It
945 is not intended to be called directly; instead, it will be
946 called by a shutdown trigger created in L{_initThreadPool}.
948 triggers = [self._threadpoolStartupID, self.threadpoolShutdownID]
949 for trigger in filter(None, triggers):
951 self.removeSystemEventTrigger(trigger)
954 self._threadpoolStartupID = None
955 self.threadpoolShutdownID = None
956 self.threadpool.stop()
957 self.threadpool = None
960 def getThreadPool(self):
962 See L{twisted.internet.interfaces.IReactorThreads.getThreadPool}.
964 if self.threadpool is None:
965 self._initThreadPool()
966 return self.threadpool
969 def callInThread(self, _callable, *args, **kwargs):
971 See L{twisted.internet.interfaces.IReactorThreads.callInThread}.
973 self.getThreadPool().callInThread(_callable, *args, **kwargs)
975 def suggestThreadPoolSize(self, size):
977 See L{twisted.internet.interfaces.IReactorThreads.suggestThreadPoolSize}.
979 self.getThreadPool().adjustPoolsize(maxthreads=size)
981 # This is for signal handlers.
982 def callFromThread(self, f, *args, **kw):
983 assert callable(f), "%s is not callable" % (f,)
984 # See comment in the other callFromThread implementation.
985 self.threadCallQueue.append((f, args, kw))
987 if platform.supportsThreads():
988 classImplements(ReactorBase, IReactorThreads)
991 class BaseConnector(styles.Ephemeral):
992 """Basic implementation of connector.
994 State can be: "connecting", "connected", "disconnected"
997 implements(IConnector)
1002 def __init__(self, factory, timeout, reactor):
1003 self.state = "disconnected"
1004 self.reactor = reactor
1005 self.factory = factory
1006 self.timeout = timeout
1008 def disconnect(self):
1009 """Disconnect whatever our state is."""
1010 if self.state == 'connecting':
1011 self.stopConnecting()
1012 elif self.state == 'connected':
1013 self.transport.loseConnection()
1016 """Start connection to remote server."""
1017 if self.state != "disconnected":
1018 raise RuntimeError, "can't connect in this state"
1020 self.state = "connecting"
1021 if not self.factoryStarted:
1022 self.factory.doStart()
1023 self.factoryStarted = 1
1024 self.transport = transport = self._makeTransport()
1025 if self.timeout is not None:
1026 self.timeoutID = self.reactor.callLater(self.timeout, transport.failIfNotConnected, error.TimeoutError())
1027 self.factory.startedConnecting(self)
1029 def stopConnecting(self):
1030 """Stop attempting to connect."""
1031 if self.state != "connecting":
1032 raise error.NotConnectingError, "we're not trying to connect"
1034 self.state = "disconnected"
1035 self.transport.failIfNotConnected(error.UserError())
1038 def cancelTimeout(self):
1039 if self.timeoutID is not None:
1041 self.timeoutID.cancel()
1046 def buildProtocol(self, addr):
1047 self.state = "connected"
1048 self.cancelTimeout()
1049 return self.factory.buildProtocol(addr)
1051 def connectionFailed(self, reason):
1052 self.cancelTimeout()
1053 self.transport = None
1054 self.state = "disconnected"
1055 self.factory.clientConnectionFailed(self, reason)
1056 if self.state == "disconnected":
1057 # factory hasn't called our connect() method
1058 self.factory.doStop()
1059 self.factoryStarted = 0
1061 def connectionLost(self, reason):
1062 self.state = "disconnected"
1063 self.factory.clientConnectionLost(self, reason)
1064 if self.state == "disconnected":
1065 # factory hasn't called our connect() method
1066 self.factory.doStop()
1067 self.factoryStarted = 0
1069 def getDestination(self):
1070 raise NotImplementedError(
1071 reflect.qual(self.__class__) + " did not implement "
1076 class BasePort(abstract.FileDescriptor):
1077 """Basic implementation of a ListeningPort.
1079 Note: This does not actually implement IListeningPort.
1082 addressFamily = None
1085 def createInternetSocket(self):
1086 s = socket.socket(self.addressFamily, self.socketType)
1088 fdesc._setCloseOnExec(s.fileno())
1093 """Raises a RuntimeError"""
1094 raise RuntimeError, "doWrite called on a %s" % reflect.qual(self.__class__)
1098 class _SignalReactorMixin(object):
1100 Private mixin to manage signals: it installs signal handlers at start time,
1101 and define run method.
1103 It can only be used mixed in with L{ReactorBase}, and has to be defined
1104 first in the inheritance (so that method resolution order finds
1105 startRunning first).
1107 @type _installSignalHandlers: C{bool}
1108 @ivar _installSignalHandlers: A flag which indicates whether any signal
1109 handlers will be installed during startup. This includes handlers for
1110 SIGCHLD to monitor child processes, and SIGINT, SIGTERM, and SIGBREAK
1111 to stop the reactor.
1114 _installSignalHandlers = False
1116 def _handleSignals(self):
1118 Install the signal handlers for the Twisted event loop.
1123 log.msg("Warning: signal module unavailable -- "
1124 "not installing signal handlers.")
1127 if signal.getsignal(signal.SIGINT) == signal.default_int_handler:
1128 # only handle if there isn't already a handler, e.g. for Pdb.
1129 signal.signal(signal.SIGINT, self.sigInt)
1130 signal.signal(signal.SIGTERM, self.sigTerm)
1132 # Catch Ctrl-Break in windows
1133 if hasattr(signal, "SIGBREAK"):
1134 signal.signal(signal.SIGBREAK, self.sigBreak)
1137 def startRunning(self, installSignalHandlers=True):
1139 Extend the base implementation in order to remember whether signal
1140 handlers should be installed later.
1142 @type installSignalHandlers: C{bool}
1143 @param installSignalHandlers: A flag which, if set, indicates that
1144 handlers for a number of (implementation-defined) signals should be
1145 installed during startup.
1147 self._installSignalHandlers = installSignalHandlers
1148 ReactorBase.startRunning(self)
1151 def _reallyStartRunning(self):
1153 Extend the base implementation by also installing signal handlers, if
1154 C{self._installSignalHandlers} is true.
1156 ReactorBase._reallyStartRunning(self)
1157 if self._installSignalHandlers:
1158 # Make sure this happens before after-startup events, since the
1159 # expectation of after-startup is that the reactor is fully
1160 # initialized. Don't do it right away for historical reasons
1161 # (perhaps some before-startup triggers don't want there to be a
1162 # custom SIGCHLD handler so that they can run child processes with
1163 # some blocking api).
1164 self._handleSignals()
1167 def run(self, installSignalHandlers=True):
1168 self.startRunning(installSignalHandlers=installSignalHandlers)
1173 while self._started:
1175 while self._started:
1176 # Advance simulation time in delayed event
1178 self.runUntilCurrent()
1180 t = self.running and t2
1183 log.msg("Unexpected error in main loop.")
1186 log.msg('Main loop terminated.')