scheduler.py (6034B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\threadutils\scheduler.py 2 import heapq 3 import logging 4 import random 5 from collections import namedtuple 6 import gametime 7 import uthread2 8 from carbon.common.lib.const import SEC 9 from eveexceptions.exceptionEater import ExceptionEater 10 logger = logging.getLogger(__name__) 11 __all__ = ['Scheduler', 'SimTimeScheduler', 'WallTimeScheduler'] 12 Event = namedtuple('Event', 'time, priority, action, arguments') 13 14 class Scheduler(object): 15 16 def __init__(self, time_func = None, delay_func = None, name = 'default'): 17 self._queue = [] 18 self.time_func = time_func 19 self.delay_func = delay_func 20 self.is_running = False 21 self.is_delaying = None 22 self.is_active = False 23 self.is_executing_event = False 24 self._name = name 25 26 def enter_absolute_event(self, time, priority, action, arguments): 27 event = Event(time, priority, action, arguments) 28 self._before_enter_event(event) 29 heapq.heappush(self._queue, event) 30 logger.debug('scheduler %s adding event %s', self._name, event) 31 self._after_enter_event() 32 return event 33 34 def enter_relative_event(self, delay, priority, action, arguments): 35 time = self.time_func() + delay 36 return self.enter_absolute_event(time, priority, action, arguments) 37 38 def cancel(self, event): 39 self._queue.remove(event) 40 heapq.heapify(self._queue) 41 logger.debug('scheduler %s cancelled event %s', self._name, event) 42 43 def empty(self): 44 return not self._queue 45 46 def run(self): 47 try: 48 queue = self._queue 49 time_func = self.time_func 50 pop = heapq.heappop 51 self.is_running = True 52 while queue: 53 next_event = self.next_event() 54 now = time_func() 55 if now < next_event.time: 56 self.delay(next_event.time - now) 57 else: 58 event = pop(queue) 59 if event is next_event: 60 self.is_executing_event = True 61 self.try_execute_event(event, next_event) 62 self.is_executing_event = False 63 self.delay(0) 64 else: 65 heapq.heappush(queue, event) 66 67 finally: 68 self.is_running = False 69 70 def try_execute_event(self, event, next_event): 71 logger.debug('%s: executing event %s', self._name, event) 72 with ExceptionEater('scheduler %s' % self._name): 73 next_event.action(*next_event.arguments) 74 75 @property 76 def queue(self): 77 events = self._queue[:] 78 return map(heapq.heappop, [events] * len(events)) 79 80 def clear_all_events(self): 81 while self._queue: 82 self._queue.pop() 83 84 logger.debug('scheduler %s cleared all events', self._name) 85 86 def cancel_by_action_and_arguments(self, action, arguments): 87 action_name = action.__name__ 88 events = [ e for e in self._queue if e.action.__name__ == action_name and e.arguments == arguments ] 89 for e in events: 90 self.cancel(e) 91 92 logger.debug('scheduler %s cancelled all event named %s with arguments %s', self._name, action_name, arguments) 93 94 def enter_random_interval_event(self, min_delay, max_delay, priority, action, arguments): 95 delay = random.randint(min_delay, max_delay) 96 return self.enter_relative_event(delay, priority, action, arguments) 97 98 def enter_repeated_interval_event(self, min_delay, max_delay, priority, action, arguments): 99 repeated_arguments = (min_delay, 100 max_delay, 101 priority, 102 action, 103 arguments) 104 return self.enter_random_interval_event(min_delay, max_delay, priority, self._do_event_and_repeat, repeated_arguments) 105 106 def _do_event_and_repeat(self, min_delay, max_delay, priority, action, arguments): 107 repeated_arguments = (min_delay, 108 max_delay, 109 priority, 110 action, 111 arguments) 112 if action(*arguments): 113 self.enter_random_interval_event(min_delay, max_delay, priority, self._do_event_and_repeat, repeated_arguments) 114 115 def delay(self, time): 116 self.is_delaying = True 117 try: 118 self.delay_func(time) 119 finally: 120 self.is_delaying = False 121 122 def start(self): 123 self.is_active = True 124 if self.queue: 125 self._start_worker() 126 127 def _start_worker(self): 128 self.worker_thread = uthread2.start_tasklet(self.run, stackless_tracing_enabled=False) 129 130 def _stop_worker(self): 131 if self.worker_thread is None: 132 return 133 self.worker_thread.kill() 134 self.worker_thread = None 135 136 def stop(self): 137 self.is_active = False 138 self._stop_worker() 139 140 def next_event(self): 141 return self.queue[0] 142 143 def _after_enter_event(self): 144 if self.is_active and not self.is_running: 145 self._start_worker() 146 147 def _before_enter_event(self, new_event): 148 if self.is_executing_event: 149 return 150 if not self.is_active: 151 return 152 if not self.is_running: 153 return 154 if not self.is_delaying: 155 return 156 if not self.empty() and new_event.time < self.next_event().time: 157 self._stop_worker() 158 159 160 def sleep_sim_time(blue_time): 161 time_seconds = blue_time / float(SEC) 162 uthread2.sleep_sim(time_seconds) 163 164 165 class SimTimeScheduler(Scheduler): 166 167 def __init__(self, name = 'sim_time_scheduler'): 168 Scheduler.__init__(self, time_func=gametime.GetSimTime, delay_func=sleep_sim_time, name=name) 169 170 171 def sleep_wall_time(blue_time): 172 time_seconds = blue_time / float(SEC) 173 uthread2.sleep(time_seconds) 174 175 176 class WallTimeScheduler(Scheduler): 177 178 def __init__(self, name = 'wall_time_scheduler'): 179 Scheduler.__init__(self, time_func=gametime.GetWallclockTime, delay_func=sleep_wall_time, name=name)