|
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). |
|
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 Remember that in Python, multiple function arguments can be packed |
|
21 in a tuple. The action function may be an instance method so it |
|
22 has another way to reference private data (besides global variables). |
|
23 Parameterless functions or methods cannot be used, however. |
|
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 heapq |
|
32 |
|
33 __all__ = ["scheduler"] |
|
34 |
|
35 class scheduler: |
|
36 def __init__(self, timefunc, delayfunc): |
|
37 """Initialize a new instance, passing the time and delay |
|
38 functions""" |
|
39 self.queue = [] |
|
40 self.timefunc = timefunc |
|
41 self.delayfunc = delayfunc |
|
42 |
|
43 def enterabs(self, time, priority, action, argument): |
|
44 """Enter a new event in the queue at an absolute time. |
|
45 |
|
46 Returns an ID for the event which can be used to remove it, |
|
47 if necessary. |
|
48 |
|
49 """ |
|
50 event = time, priority, action, argument |
|
51 heapq.heappush(self.queue, event) |
|
52 return event # The ID |
|
53 |
|
54 def enter(self, delay, priority, action, argument): |
|
55 """A variant that specifies the time as a relative time. |
|
56 |
|
57 This is actually the more commonly used interface. |
|
58 |
|
59 """ |
|
60 time = self.timefunc() + delay |
|
61 return self.enterabs(time, priority, action, argument) |
|
62 |
|
63 def cancel(self, event): |
|
64 """Remove an event from the queue. |
|
65 |
|
66 This must be presented the ID as returned by enter(). |
|
67 If the event is not in the queue, this raises RuntimeError. |
|
68 |
|
69 """ |
|
70 self.queue.remove(event) |
|
71 heapq.heapify(self.queue) |
|
72 |
|
73 def empty(self): |
|
74 """Check whether the queue is empty.""" |
|
75 return not self.queue |
|
76 |
|
77 def run(self): |
|
78 """Execute events until the queue is empty. |
|
79 |
|
80 When there is a positive delay until the first event, the |
|
81 delay function is called and the event is left in the queue; |
|
82 otherwise, the event is removed from the queue and executed |
|
83 (its action function is called, passing it the argument). If |
|
84 the delay function returns prematurely, it is simply |
|
85 restarted. |
|
86 |
|
87 It is legal for both the delay function and the action |
|
88 function to to modify the queue or to raise an exception; |
|
89 exceptions are not caught but the scheduler's state remains |
|
90 well-defined so run() may be called again. |
|
91 |
|
92 A questionably hack is added to allow other threads to run: |
|
93 just after an event is executed, a delay of 0 is executed, to |
|
94 avoid monopolizing the CPU when other threads are also |
|
95 runnable. |
|
96 |
|
97 """ |
|
98 # localize variable access to minimize overhead |
|
99 # and to improve thread safety |
|
100 q = self.queue |
|
101 delayfunc = self.delayfunc |
|
102 timefunc = self.timefunc |
|
103 pop = heapq.heappop |
|
104 while q: |
|
105 time, priority, action, argument = checked_event = q[0] |
|
106 now = timefunc() |
|
107 if now < time: |
|
108 delayfunc(time - now) |
|
109 else: |
|
110 event = pop(q) |
|
111 # Verify that the event was not removed or altered |
|
112 # by another thread after we last looked at q[0]. |
|
113 if event is checked_event: |
|
114 void = action(*argument) |
|
115 delayfunc(0) # Let other threads run |
|
116 else: |
|
117 heapq.heappush(event) |