comparison: scheduler.py
scheduler.py
- changeset 80
- a409242121d5
- parent 79
- b075b3db3044
- child 81
- 7bcd715f19e3
equal
deleted
inserted
replaced
| 79:b075b3db3044 | 80:a409242121d5 |
|---|---|
| 1 # | |
| 2 # Scheduler for Borgend | |
| 3 # | |
| 4 # This module simply provide a way for other threads to until a given time | |
| 5 # | |
| 6 | |
| 7 import time | |
| 8 import loggers | |
| 9 import dreamtime | |
| 10 from threading import Condition, Lock, Thread | |
| 11 | |
| 12 logger=loggers.get(__name__) | |
| 13 | |
| 14 class QueuedEvent: | |
| 15 def __init__(self, cond, name=None): | |
| 16 self.next=None | |
| 17 self.prev=None | |
| 18 self.name=name | |
| 19 self.cond=cond | |
| 20 | |
| 21 def __lt__(self, other): | |
| 22 raise NotImplementedError | |
| 23 | |
| 24 def insert_after(self, ev): | |
| 25 if not self.next or ev<self.next: | |
| 26 self.insert_immediately_after(ev) | |
| 27 else: | |
| 28 self.next.insert_after(ev) | |
| 29 | |
| 30 def insert_immediately_after(self, ev): | |
| 31 assert(ev.next is None and ev.prev is None) | |
| 32 ev.prev=self | |
| 33 ev.next=self.next | |
| 34 self.next=ev | |
| 35 | |
| 36 def insert_immediately_before(self, ev): | |
| 37 assert(ev.next is None and ev.prev is None) | |
| 38 ev.next=self | |
| 39 ev.prev=self.prev | |
| 40 self.prev=ev | |
| 41 | |
| 42 def unlink(self): | |
| 43 n=self.next | |
| 44 p=self.prev | |
| 45 if n: | |
| 46 n.prev=p | |
| 47 if p: | |
| 48 p.next=n | |
| 49 self.next=None | |
| 50 self.prev=None | |
| 51 | |
| 52 class ScheduledEvent(QueuedEvent): | |
| 53 #@accepts(ScheduledEvent, dreamtime.Time, threading.Cond, str) | |
| 54 def __init__(self, when, cond, name=None): | |
| 55 super().__init__(cond, name=name) | |
| 56 self.when=when | |
| 57 | |
| 58 def __lt__(self, other): | |
| 59 return self.when < other.when | |
| 60 | |
| 61 class TerminableThread(Thread): | |
| 62 def __init__(self, *args, **kwargs): | |
| 63 super().__init__(*args, **kwargs) | |
| 64 self._terminate=False | |
| 65 self._cond=Condition() | |
| 66 | |
| 67 def terminate(self): | |
| 68 with self._cond: | |
| 69 _terminate=True | |
| 70 self._cond.notify() | |
| 71 | |
| 72 class QueueThread(TerminableThread): | |
| 73 def __init__(self, *args, **kwargs): | |
| 74 super().__init__(*args, **kwargs) | |
| 75 self.daemon = True | |
| 76 self._list = None | |
| 77 | |
| 78 def _insert(self, ev): | |
| 79 if not self._list: | |
| 80 #logger.debug("Insert first") | |
| 81 self._list=ev | |
| 82 elif ev<self._list: | |
| 83 #logger.debug("Insert beginning") | |
| 84 self._list.insert_immediately_before(ev) | |
| 85 self._list=ev | |
| 86 else: | |
| 87 #logger.debug("Insert after") | |
| 88 self._list.insert_after(ev) | |
| 89 | |
| 90 self._cond.notify() | |
| 91 | |
| 92 def _unlink(self, ev): | |
| 93 if ev==self._list: | |
| 94 self._list=ev.next | |
| 95 ev.unlink() | |
| 96 | |
| 97 def _resort(self): | |
| 98 oldlist=self._list | |
| 99 self._list=None | |
| 100 while oldlist: | |
| 101 ev=oldlist | |
| 102 oldlist=oldlist.next | |
| 103 ev.unlink() | |
| 104 self._insert(ev) | |
| 105 | |
| 106 | |
| 107 | |
| 108 class Scheduler(QueueThread): | |
| 109 # Default to precision of 60 seconds: the scheduler thread will never | |
| 110 # sleep longer than that, to get quickly back on track with the schedule | |
| 111 # when the computer wakes up from sleep | |
| 112 def __init__(self, precision=60): | |
| 113 self.precision = precision | |
| 114 self._next_event_time = None | |
| 115 super().__init__(target = self._scheduler_thread, name = 'Scheduler') | |
| 116 dreamtime.add_callback(self, self._wakeup_callback) | |
| 117 | |
| 118 def _scheduler_thread(self): | |
| 119 logger.debug("Scheduler thread started") | |
| 120 with self._cond: | |
| 121 while not self._terminate: | |
| 122 now = time.monotonic() | |
| 123 if not self._list: | |
| 124 timeout = None | |
| 125 else: | |
| 126 # Wait at most precision seconds, or until next event if it | |
| 127 # comes earlier | |
| 128 timeout=min(self.precision, self._list.when.realtime()-now) | |
| 129 | |
| 130 if not timeout or timeout>0: | |
| 131 logger.debug("Scheduler waiting %d seconds" % (timeout or (-1))) | |
| 132 self._cond.wait(timeout) | |
| 133 now = time.monotonic() | |
| 134 | |
| 135 logger.debug("Scheduler timed out") | |
| 136 | |
| 137 while self._list and self._list.when.monotonic() <= now: | |
| 138 ev=self._list | |
| 139 logger.debug("Scheduler activating %s" % (ev.name or "(unknown)")) | |
| 140 # We are only allowed to remove ev from list when ev.cond allows | |
| 141 with ev.cond: | |
| 142 self._list=ev.next | |
| 143 ev.unlink() | |
| 144 ev.cond.notifyAll() | |
| 145 | |
| 146 def _wakeup_callback(self): | |
| 147 logger.debug("Rescheduling events after wakeup") | |
| 148 with self._cond: | |
| 149 self._resort() | |
| 150 | |
| 151 def _wait(self, ev): | |
| 152 with self._cond: | |
| 153 self._insert(ev) | |
| 154 | |
| 155 # This will release the lock on cond, allowing queue manager (scheduler) | |
| 156 # thread to notify us if we are already to be released | |
| 157 ev.cond.wait() | |
| 158 | |
| 159 # If we were woken up by some other event, not the scheduler, | |
| 160 # ensure the event is removed | |
| 161 with self._cond: | |
| 162 self._unlink(ev) | |
| 163 | |
| 164 # cond has to be acquired on entry! | |
| 165 def wait_until(self, when, cond, name=None): | |
| 166 logger.debug("Scheduling '%s' in %s seconds [%s]" % | |
| 167 (name, when.seconds_to(), when.__class__.__name__)) | |
| 168 self._wait(ScheduledEvent(when, cond, name)) | |
| 169 |
