OLD | NEW |
(Empty) | |
| 1 """A generally useful event scheduler class. |
| 2 |
| 3 Each instance of this class manages its own queue. |
| 4 No multi-threading is implied; you are supposed to hack that |
| 5 yourself, or use a single instance per application. |
| 6 |
| 7 Each instance is parametrized with two functions, one that is |
| 8 supposed to return the current time, one that is supposed to |
| 9 implement a delay. You can implement real-time scheduling by |
| 10 substituting time and sleep from built-in module time, or you can |
| 11 implement simulated time by writing your own functions. This can |
| 12 also be used to integrate scheduling with STDWIN events; the delay |
| 13 function is allowed to modify the queue. Time can be expressed as |
| 14 integers or floating point numbers, as long as it is consistent. |
| 15 |
| 16 Events are specified by tuples (time, priority, action, argument, kwargs). |
| 17 As in UNIX, lower priority numbers mean higher priority; in this |
| 18 way the queue can be maintained as a priority queue. Execution of the |
| 19 event means calling the action function, passing it the argument |
| 20 sequence in "argument" (remember that in Python, multiple function |
| 21 arguments are be packed in a sequence) and keyword parameters in "kwargs". |
| 22 The action function may be an instance method so it |
| 23 has another way to reference private data (besides global variables). |
| 24 """ |
| 25 |
| 26 # XXX The timefunc and delayfunc should have been defined as methods |
| 27 # XXX so you can define new kinds of schedulers using subclassing |
| 28 # XXX instead of having to define a module or class just to hold |
| 29 # XXX the global state of your particular time and delay functions. |
| 30 |
| 31 import time |
| 32 import heapq |
| 33 from collections import namedtuple |
| 34 try: |
| 35 import threading |
| 36 except ImportError: |
| 37 import dummy_threading as threading |
| 38 |
| 39 # py3 only |
| 40 # from time import monotonic as _time |
| 41 from time import time as _time |
| 42 |
| 43 __all__ = ["scheduler"] |
| 44 |
| 45 class Event(namedtuple('Event', 'time, priority, action, argument, kwargs')): |
| 46 # pylint: disable=no-self-argument |
| 47 def __eq__(s, o): return (s.time, s.priority) == (o.time, o.priority) |
| 48 def __lt__(s, o): return (s.time, s.priority) < (o.time, o.priority) |
| 49 def __le__(s, o): return (s.time, s.priority) <= (o.time, o.priority) |
| 50 def __gt__(s, o): return (s.time, s.priority) > (o.time, o.priority) |
| 51 def __ge__(s, o): return (s.time, s.priority) >= (o.time, o.priority) |
| 52 |
| 53 # py3 only |
| 54 # |
| 55 # Event.time.__doc__ = ('''Numeric type compatible with the return value of the |
| 56 # timefunc function passed to the constructor.''') |
| 57 # Event.priority.__doc__ = ('''Events scheduled for the same time will be execut
ed |
| 58 # in the order of their priority.''') |
| 59 # Event.action.__doc__ = ('''Executing the event means executing |
| 60 # action(*argument, **kwargs)''') |
| 61 # Event.argument.__doc__ = ('''argument is a sequence holding the positional |
| 62 # arguments for the action.''') |
| 63 # Event.kwargs.__doc__ = ('''kwargs is a dictionary holding the keyword |
| 64 # arguments for the action.''') |
| 65 |
| 66 _sentinel = object() |
| 67 |
| 68 class scheduler: |
| 69 |
| 70 def __init__(self, timefunc=_time, delayfunc=time.sleep): |
| 71 """Initialize a new instance, passing the time and delay |
| 72 functions""" |
| 73 self._queue = [] |
| 74 self._lock = threading.RLock() |
| 75 self.timefunc = timefunc |
| 76 self.delayfunc = delayfunc |
| 77 |
| 78 def enterabs(self, time, priority, action, argument=(), kwargs=_sentinel): |
| 79 """Enter a new event in the queue at an absolute time. |
| 80 |
| 81 Returns an ID for the event which can be used to remove it, |
| 82 if necessary. |
| 83 |
| 84 """ |
| 85 if kwargs is _sentinel: |
| 86 kwargs = {} |
| 87 event = Event(time, priority, action, argument, kwargs) |
| 88 with self._lock: |
| 89 heapq.heappush(self._queue, event) |
| 90 return event # The ID |
| 91 |
| 92 def enter(self, delay, priority, action, argument=(), kwargs=_sentinel): |
| 93 """A variant that specifies the time as a relative time. |
| 94 |
| 95 This is actually the more commonly used interface. |
| 96 |
| 97 """ |
| 98 time = self.timefunc() + delay |
| 99 return self.enterabs(time, priority, action, argument, kwargs) |
| 100 |
| 101 def cancel(self, event): |
| 102 """Remove an event from the queue. |
| 103 |
| 104 This must be presented the ID as returned by enter(). |
| 105 If the event is not in the queue, this raises ValueError. |
| 106 |
| 107 """ |
| 108 with self._lock: |
| 109 self._queue.remove(event) |
| 110 heapq.heapify(self._queue) |
| 111 |
| 112 def empty(self): |
| 113 """Check whether the queue is empty.""" |
| 114 with self._lock: |
| 115 return not self._queue |
| 116 |
| 117 def run(self, blocking=True): |
| 118 """Execute events until the queue is empty. |
| 119 If blocking is False executes the scheduled events due to |
| 120 expire soonest (if any) and then return the deadline of the |
| 121 next scheduled call in the scheduler. |
| 122 |
| 123 When there is a positive delay until the first event, the |
| 124 delay function is called and the event is left in the queue; |
| 125 otherwise, the event is removed from the queue and executed |
| 126 (its action function is called, passing it the argument). If |
| 127 the delay function returns prematurely, it is simply |
| 128 restarted. |
| 129 |
| 130 It is legal for both the delay function and the action |
| 131 function to modify the queue or to raise an exception; |
| 132 exceptions are not caught but the scheduler's state remains |
| 133 well-defined so run() may be called again. |
| 134 |
| 135 A questionable hack is added to allow other threads to run: |
| 136 just after an event is executed, a delay of 0 is executed, to |
| 137 avoid monopolizing the CPU when other threads are also |
| 138 runnable. |
| 139 |
| 140 """ |
| 141 # localize variable access to minimize overhead |
| 142 # and to improve thread safety |
| 143 lock = self._lock |
| 144 q = self._queue |
| 145 delayfunc = self.delayfunc |
| 146 timefunc = self.timefunc |
| 147 pop = heapq.heappop |
| 148 while True: |
| 149 with lock: |
| 150 if not q: |
| 151 break |
| 152 time, priority, action, argument, kwargs = q[0] |
| 153 now = timefunc() |
| 154 if time > now: |
| 155 delay = True |
| 156 else: |
| 157 delay = False |
| 158 pop(q) |
| 159 if delay: |
| 160 if not blocking: |
| 161 return time - now |
| 162 delayfunc(time - now) |
| 163 else: |
| 164 action(*argument, **kwargs) |
| 165 delayfunc(0) # Let other threads run |
| 166 |
| 167 @property |
| 168 def queue(self): |
| 169 """An ordered list of upcoming events. |
| 170 |
| 171 Events are named tuples with fields for: |
| 172 time, priority, action, arguments, kwargs |
| 173 |
| 174 """ |
| 175 # Use heapq to sort the queue rather than using 'sorted(self._queue)'. |
| 176 # With heapq, two events scheduled at the same time will show in |
| 177 # the actual order they would be retrieved. |
| 178 with self._lock: |
| 179 events = self._queue[:] |
| 180 return list(map(heapq.heappop, [events]*len(events))) |
OLD | NEW |