1 # -*- test-case-name: twisted.test.test_task,twisted.test.test_cooperator -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Scheduling utility methods and classes.
15 from zope.interface import implements
17 from twisted.python import reflect
18 from twisted.python.failure import Failure
20 from twisted.internet import base, defer
21 from twisted.internet.interfaces import IReactorTime
25 """Call a function repeatedly.
27 If C{f} returns a deferred, rescheduling will not take place until the
28 deferred has fired. The result value is ignored.
30 @ivar f: The function to call.
31 @ivar a: A tuple of arguments to pass the function.
32 @ivar kw: A dictionary of keyword arguments to pass to the function.
33 @ivar clock: A provider of
34 L{twisted.internet.interfaces.IReactorTime}. The default is
35 L{twisted.internet.reactor}. Feel free to set this to
36 something else, but it probably ought to be set *before*
39 @type running: C{bool}
40 @ivar running: A flag which is C{True} while C{f} is scheduled to be called
41 (or is currently being called). It is set to C{True} when L{start} is
42 called and set to C{False} when L{stop} is called or if C{f} raises an
43 exception. In either case, it will be C{False} by the time the
44 C{Deferred} returned by L{start} fires its callback or errback.
46 @type _expectNextCallAt: C{float}
47 @ivar _expectNextCallAt: The time at which this instance most recently
48 scheduled itself to run.
50 @type _realLastTime: C{float}
51 @ivar _realLastTime: When counting skips, the time at which the skip
52 counter was last invoked.
54 @type _runAtStart: C{bool}
55 @ivar _runAtStart: A flag indicating whether the 'now' argument was passed
56 to L{LoopingCall.start}.
63 _expectNextCallAt = 0.0
67 def __init__(self, f, *a, **kw):
71 from twisted.internet import reactor
75 def withCount(cls, countCallable):
77 An alternate constructor for L{LoopingCall} that makes available the
78 number of calls which should have occurred since it was last invoked.
80 Note that this number is an C{int} value; It represents the discrete
81 number of calls that should have been made. For example, if you are
82 using a looping call to display an animation with discrete frames, this
83 number would be the number of frames to advance.
85 The count is normally 1, but can be higher. For example, if the reactor
86 is blocked and takes too long to invoke the L{LoopingCall}, a Deferred
87 returned from a previous call is not fired before an interval has
88 elapsed, or if the callable itself blocks for longer than an interval,
89 preventing I{itself} from being called.
91 @param countCallable: A callable that will be invoked each time the
92 resulting LoopingCall is run, with an integer specifying the number
93 of calls that should have been invoked.
95 @type countCallable: 1-argument callable which takes an C{int}
97 @return: An instance of L{LoopingCall} with call counting enabled,
98 which provides the count as the first positional argument.
100 @rtype: L{LoopingCall}
106 now = self.clock.seconds()
107 lastTime = self._realLastTime
109 lastTime = self.starttime
111 lastTime -= self.interval
112 self._realLastTime = now
113 lastInterval = self._intervalOf(lastTime)
114 thisInterval = self._intervalOf(now)
115 count = thisInterval - lastInterval
116 return countCallable(count)
120 self._realLastTime = None
124 withCount = classmethod(withCount)
127 def _intervalOf(self, t):
129 Determine the number of intervals passed as of the given point in
132 @param t: The specified time (from the start of the L{LoopingCall}) to
133 be measured in intervals
135 @return: The C{int} number of intervals which have passed as of the
138 elapsedTime = t - self.starttime
139 intervalNum = int(elapsedTime / self.interval)
143 def start(self, interval, now=True):
145 Start running function every interval seconds.
147 @param interval: The number of seconds between calls. May be
148 less than one. Precision will depend on the underlying
149 platform, the available hardware, and the load on the system.
151 @param now: If True, run this call right now. Otherwise, wait
152 until the interval has elapsed before beginning.
154 @return: A Deferred whose callback will be invoked with
155 C{self} when C{self.stop} is called, or whose errback will be
156 invoked when the function raises an exception or returned a
157 deferred that has its errback invoked.
159 assert not self.running, ("Tried to start an already running "
162 raise ValueError, "interval must be >= 0"
164 d = self.deferred = defer.Deferred()
165 self.starttime = self.clock.seconds()
166 self._expectNextCallAt = self.starttime
167 self.interval = interval
168 self._runAtStart = now
176 """Stop running function.
178 assert self.running, ("Tried to stop a LoopingCall that was "
181 if self.call is not None:
184 d, self.deferred = self.deferred, None
189 Skip the next iteration and reset the timer.
193 assert self.running, ("Tried to reset a LoopingCall that was "
195 if self.call is not None:
198 self._expectNextCallAt = self.clock.seconds()
206 d, self.deferred = self.deferred, None
211 d, self.deferred = self.deferred, None
215 d = defer.maybeDeferred(self.f, *self.a, **self.kw)
220 def _reschedule(self):
222 Schedule the next iteration of this looping call.
224 if self.interval == 0:
225 self.call = self.clock.callLater(0, self)
228 currentTime = self.clock.seconds()
229 # Find how long is left until the interval comes around again.
230 untilNextTime = (self._expectNextCallAt - currentTime) % self.interval
231 # Make sure it is in the future, in case more than one interval worth
232 # of time passed since the previous call was made.
234 self._expectNextCallAt + self.interval, currentTime + untilNextTime)
235 # If the interval falls on the current time exactly, skip it and
236 # schedule the call for the next interval.
237 if nextTime == currentTime:
238 nextTime += self.interval
239 self._expectNextCallAt = nextTime
240 self.call = self.clock.callLater(nextTime - currentTime, self)
244 if hasattr(self.f, 'func_name'):
245 func = self.f.func_name
246 if hasattr(self.f, 'im_class'):
247 func = self.f.im_class.__name__ + '.' + func
249 func = reflect.safe_repr(self.f)
251 return 'LoopingCall<%r>(%s, *%s, **%s)' % (
252 self.interval, func, reflect.safe_repr(self.a),
253 reflect.safe_repr(self.kw))
257 class SchedulerError(Exception):
259 The operation could not be completed because the scheduler or one of its
260 tasks was in an invalid state. This exception should not be raised
261 directly, but is a superclass of various scheduler-state-related
267 class SchedulerStopped(SchedulerError):
269 The operation could not complete because the scheduler was stopped in
270 progress or was already stopped.
275 class TaskFinished(SchedulerError):
277 The operation could not complete because the task was already completed,
278 stopped, encountered an error or otherwise permanently stopped running.
283 class TaskDone(TaskFinished):
285 The operation could not complete because the task was already completed.
290 class TaskStopped(TaskFinished):
292 The operation could not complete because the task was stopped.
297 class TaskFailed(TaskFinished):
299 The operation could not complete because the task died with an unhandled
305 class NotPaused(SchedulerError):
307 This exception is raised when a task is resumed which was not previously
313 class _Timer(object):
316 self.end = time.time() + self.MAX_SLICE
320 return time.time() >= self.end
324 _EPSILON = 0.00000001
325 def _defaultScheduler(x):
326 from twisted.internet import reactor
327 return reactor.callLater(_EPSILON, x)
330 class CooperativeTask(object):
332 A L{CooperativeTask} is a task object inside a L{Cooperator}, which can be
333 paused, resumed, and stopped. It can also have its completion (or
334 termination) monitored.
336 @see: L{CooperativeTask.cooperate}
338 @ivar _iterator: the iterator to iterate when this L{CooperativeTask} is
341 @ivar _cooperator: the L{Cooperator} that this L{CooperativeTask}
342 participates in, which is used to re-insert it upon resume.
344 @ivar _deferreds: the list of L{defer.Deferred}s to fire when this task
345 completes, fails, or finishes.
347 @type _deferreds: L{list}
349 @type _cooperator: L{Cooperator}
351 @ivar _pauseCount: the number of times that this L{CooperativeTask} has
352 been paused; if 0, it is running.
354 @type _pauseCount: L{int}
356 @ivar _completionState: The completion-state of this L{CooperativeTask}.
357 C{None} if the task is not yet completed, an instance of L{TaskStopped}
358 if C{stop} was called to stop this task early, of L{TaskFailed} if the
359 application code in the iterator raised an exception which caused it to
360 terminate, and of L{TaskDone} if it terminated normally via raising
363 @type _completionState: L{TaskFinished}
366 def __init__(self, iterator, cooperator):
368 A private constructor: to create a new L{CooperativeTask}, see
369 L{Cooperator.cooperate}.
371 self._iterator = iterator
372 self._cooperator = cooperator
375 self._completionState = None
376 self._completionResult = None
377 cooperator._addTask(self)
382 Get a L{defer.Deferred} notification of when this task is complete.
384 @return: a L{defer.Deferred} that fires with the C{iterator} that this
385 L{CooperativeTask} was created with when the iterator has been
386 exhausted (i.e. its C{next} method has raised L{StopIteration}), or
387 fails with the exception raised by C{next} if it raises some other
390 @rtype: L{defer.Deferred}
393 if self._completionState is None:
394 self._deferreds.append(d)
396 d.callback(self._completionResult)
402 Pause this L{CooperativeTask}. Stop doing work until
403 L{CooperativeTask.resume} is called. If C{pause} is called more than
404 once, C{resume} must be called an equal number of times to resume this
407 @raise TaskFinished: if this task has already finished or completed.
410 self._pauseCount += 1
411 if self._pauseCount == 1:
412 self._cooperator._removeTask(self)
417 Resume processing of a paused L{CooperativeTask}.
419 @raise NotPaused: if this L{CooperativeTask} is not paused.
421 if self._pauseCount == 0:
423 self._pauseCount -= 1
424 if self._pauseCount == 0 and self._completionState is None:
425 self._cooperator._addTask(self)
428 def _completeWith(self, completionState, deferredResult):
430 @param completionState: a L{TaskFinished} exception or a subclass
431 thereof, indicating what exception should be raised when subsequent
432 operations are performed.
434 @param deferredResult: the result to fire all the deferreds with.
436 self._completionState = completionState
437 self._completionResult = deferredResult
438 if not self._pauseCount:
439 self._cooperator._removeTask(self)
441 # The Deferreds need to be invoked after all this is completed, because
442 # a Deferred may want to manipulate other tasks in a Cooperator. For
443 # example, if you call "stop()" on a cooperator in a callback on a
444 # Deferred returned from whenDone(), this CooperativeTask must be gone
445 # from the Cooperator by that point so that _completeWith is not
446 # invoked reentrantly; that would cause these Deferreds to blow up with
447 # an AlreadyCalledError, or the _removeTask to fail with a ValueError.
448 for d in self._deferreds:
449 d.callback(deferredResult)
454 Stop further processing of this task.
456 @raise TaskFinished: if this L{CooperativeTask} has previously
457 completed, via C{stop}, completion, or failure.
460 self._completeWith(TaskStopped(), Failure(TaskStopped()))
463 def _checkFinish(self):
465 If this task has been stopped, raise the appropriate subclass of
468 if self._completionState is not None:
469 raise self._completionState
472 def _oneWorkUnit(self):
474 Perform one unit of work for this task, retrieving one item from its
475 iterator, stopping if there are no further items in the iterator, and
476 pausing if the result was a L{defer.Deferred}.
479 result = self._iterator.next()
480 except StopIteration:
481 self._completeWith(TaskDone(), self._iterator)
483 self._completeWith(TaskFailed(), Failure())
485 if isinstance(result, defer.Deferred):
488 self._completeWith(TaskFailed(), f)
489 result.addCallbacks(lambda result: self.resume(),
494 class Cooperator(object):
496 Cooperative task scheduler.
500 terminationPredicateFactory=_Timer,
501 scheduler=_defaultScheduler,
504 Create a scheduler-like object to which iterators may be added.
506 @param terminationPredicateFactory: A no-argument callable which will
507 be invoked at the beginning of each step and should return a
508 no-argument callable which will return True when the step should be
509 terminated. The default factory is time-based and allows iterators to
510 run for 1/100th of a second at a time.
512 @param scheduler: A one-argument callable which takes a no-argument
513 callable and should invoke it at some future point. This will be used
514 to schedule each step of this Cooperator.
516 @param started: A boolean which indicates whether iterators should be
517 stepped as soon as they are added, or if they will be queued up until
518 L{Cooperator.start} is called.
521 self._metarator = iter(())
522 self._terminationPredicateFactory = terminationPredicateFactory
523 self._scheduler = scheduler
524 self._delayedCall = None
525 self._stopped = False
526 self._started = started
529 def coiterate(self, iterator, doneDeferred=None):
531 Add an iterator to the list of iterators this L{Cooperator} is
534 @param doneDeferred: If specified, this will be the Deferred used as
535 the completion deferred. It is suggested that you use the default,
536 which creates a new Deferred for you.
538 @return: a Deferred that will fire when the iterator finishes.
540 if doneDeferred is None:
541 doneDeferred = defer.Deferred()
542 CooperativeTask(iterator, self).whenDone().chainDeferred(doneDeferred)
546 def cooperate(self, iterator):
548 Start running the given iterator as a long-running cooperative task, by
549 calling next() on it as a periodic timed event.
551 @param iterator: the iterator to invoke.
553 @return: a L{CooperativeTask} object representing this task.
555 return CooperativeTask(iterator, self)
558 def _addTask(self, task):
560 Add a L{CooperativeTask} object to this L{Cooperator}.
563 self._tasks.append(task) # XXX silly, I know, but _completeWith
565 task._completeWith(SchedulerStopped(), Failure(SchedulerStopped()))
567 self._tasks.append(task)
571 def _removeTask(self, task):
573 Remove a L{CooperativeTask} from this L{Cooperator}.
575 self._tasks.remove(task)
576 # If no work left to do, cancel the delayed call:
577 if not self._tasks and self._delayedCall:
578 self._delayedCall.cancel()
579 self._delayedCall = None
582 def _tasksWhileNotStopped(self):
584 Yield all L{CooperativeTask} objects in a loop as long as this
585 L{Cooperator}'s termination condition has not been met.
587 terminator = self._terminationPredicateFactory()
589 for t in self._metarator:
593 self._metarator = iter(self._tasks)
598 Run one scheduler tick.
600 self._delayedCall = None
601 for taskObj in self._tasksWhileNotStopped():
602 taskObj._oneWorkUnit()
606 _mustScheduleOnStart = False
607 def _reschedule(self):
608 if not self._started:
609 self._mustScheduleOnStart = True
611 if self._delayedCall is None and self._tasks:
612 self._delayedCall = self._scheduler(self._tick)
617 Begin scheduling steps.
619 self._stopped = False
621 if self._mustScheduleOnStart:
622 del self._mustScheduleOnStart
628 Stop scheduling steps. Errback the completion Deferreds of all
629 iterators which have been added and forget about them.
632 for taskObj in self._tasks:
633 taskObj._completeWith(SchedulerStopped(),
634 Failure(SchedulerStopped()))
636 if self._delayedCall is not None:
637 self._delayedCall.cancel()
638 self._delayedCall = None
642 _theCooperator = Cooperator()
644 def coiterate(iterator):
646 Cooperatively iterate over the given iterator, dividing runtime between it
647 and all other iterators which have been passed to this function and not yet
650 return _theCooperator.coiterate(iterator)
654 def cooperate(iterator):
656 Start running the given iterator as a long-running cooperative task, by
657 calling next() on it as a periodic timed event.
659 @param iterator: the iterator to invoke.
661 @return: a L{CooperativeTask} object representing this task.
663 return _theCooperator.cooperate(iterator)
669 Provide a deterministic, easily-controlled implementation of
670 L{IReactorTime.callLater}. This is commonly useful for writing
671 deterministic unit tests for code which schedules events using this API.
673 implements(IReactorTime)
683 Pretend to be time.time(). This is used internally when an operation
684 such as L{IDelayedCall.reset} needs to determine a a time value
685 relative to the current time.
688 @return: The time which should be considered the current time.
693 def _sortCalls(self):
695 Sort the pending calls according to the time they are scheduled.
697 self.calls.sort(lambda a, b: cmp(a.getTime(), b.getTime()))
700 def callLater(self, when, what, *a, **kw):
702 See L{twisted.internet.interfaces.IReactorTime.callLater}.
704 dc = base.DelayedCall(self.seconds() + when,
709 self.calls.append(dc)
714 def getDelayedCalls(self):
716 See L{twisted.internet.interfaces.IReactorTime.getDelayedCalls}
721 def advance(self, amount):
723 Move time on this clock forward by the given amount and run whatever
724 pending calls should be run.
726 @type amount: C{float}
727 @param amount: The number of seconds which to advance this clock's
730 self.rightNow += amount
732 while self.calls and self.calls[0].getTime() <= self.seconds():
733 call = self.calls.pop(0)
735 call.func(*call.args, **call.kw)
739 def pump(self, timings):
741 Advance incrementally by the given set of times.
743 @type timings: iterable of C{float}
745 for amount in timings:
750 def deferLater(clock, delay, callable, *args, **kw):
752 Call the given function after a certain period of time has passed.
754 @type clock: L{IReactorTime} provider
755 @param clock: The object which will be used to schedule the delayed
758 @type delay: C{float} or C{int}
759 @param delay: The number of seconds to wait before calling the function.
761 @param callable: The object to call after the delay.
763 @param *args: The positional arguments to pass to C{callable}.
765 @param **kw: The keyword arguments to pass to C{callable}.
767 @rtype: L{defer.Deferred}
769 @return: A deferred that fires with the result of the callable when the
770 specified time has elapsed.
772 def deferLaterCancel(deferred):
774 d = defer.Deferred(deferLaterCancel)
775 d.addCallback(lambda ignored: callable(*args, **kw))
776 delayedCall = clock.callLater(delay, d.callback, None)
786 'SchedulerStopped', 'Cooperator', 'coiterate',