threadutils.py (8905B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\brennivin\threadutils.py 2 import inspect as _inspect 3 import sys as _sys 4 import threading as _threading 5 import time as _time 6 import traceback as _traceback 7 import random 8 from . import compat as _compat, dochelpers as _dochelpers 9 10 class ChunkIter(object): 11 12 @classmethod 13 def start_thread(cls, target, name): 14 thread = _threading.Thread(target=target, name=name) 15 thread.daemon = True 16 thread.start() 17 return thread 18 19 def __init__(self, iterable_, callback, chunksize = 50): 20 self._isFinished = False 21 self._cancelReq = False 22 self.chunksize = chunksize 23 self.fireCount = 0 24 self.iterable = iterable_ 25 self._fireCallback = Signal('list') 26 self._fireCallback.connect(callback) 27 self.threading = _threading 28 self.sleep = _time.sleep 29 self.thread = type(self).start_thread(self._run_thread, 'ChunkIterWorker') 30 31 def _run_thread(self): 32 chunk = [] 33 for item in self.iterable: 34 chunk.append(item) 35 if len(chunk) == self.chunksize: 36 self._fireCallback.emit(list(chunk)) 37 del chunk[:] 38 if self._cancelReq: 39 break 40 41 if chunk and not self._cancelReq: 42 self._fireCallback.emit(chunk) 43 self._isFinished = True 44 45 def wait_for_completion(self, timeout = None): 46 self.thread.join(timeout) 47 48 def wait_chunks(self, chunks = 1, sleep_interval = 1): 49 if self._isFinished: 50 return 51 fireCnt = [0] 52 53 def onChunk(_): 54 fireCnt[0] += 1 55 56 self._fireCallback.connect(onChunk) 57 while not self._isFinished and fireCnt[0] < chunks: 58 self.sleep(sleep_interval) 59 60 WaitChunks = wait_chunks 61 62 def is_finished(self): 63 return self._isFinished 64 65 IsFinished = is_finished 66 67 def cancel(self): 68 self._cancelReq = True 69 70 Cancel = cancel 71 72 73 class Signal(object): 74 75 def __init__(self, eventdoc = None, onerror = _dochelpers.pretty_module_func(_traceback.print_exception)): 76 self._delegates = [] 77 self.eventdoc = eventdoc 78 self.onerror = onerror 79 80 def connect(self, callback): 81 self._delegates.append(callback) 82 83 def emit(self, *args, **kwargs): 84 dels = list(self._delegates) 85 for d in dels: 86 try: 87 d(*args, **kwargs) 88 except Exception: 89 self.onerror(*_sys.exc_info()) 90 91 return len(dels) 92 93 def disconnect(self, callback): 94 self._delegates.remove(callback) 95 96 def clear(self): 97 self._delegates = [] 98 99 100 class ExceptionalThread(_threading.Thread): 101 102 def __init__(self, *args, **kwargs): 103 self.reraise = kwargs.pop('reraise', None) 104 _threading.Thread.__init__(self, *args, **kwargs) 105 self.excepted = Signal('(etype, value, tb)') 106 self.exc_info = None 107 108 def run(self): 109 try: 110 _threading.Thread.run(self) 111 except Exception: 112 self.exc_info = _sys.exc_info() 113 defaultExceptHook = _sys.excepthook == _sys.__excepthook__ 114 if not defaultExceptHook: 115 _sys.excepthook(*self.exc_info) 116 hadlisteners = self.excepted.emit(self.exc_info) 117 reraise = self.reraise 118 if reraise is None: 119 if not defaultExceptHook: 120 reraise = False 121 elif hadlisteners: 122 reraise = False 123 else: 124 reraise = True 125 if reraise is True: 126 raise 127 128 def join(self, timeout = None): 129 _threading.Thread.join(self, timeout) 130 if self.exc_info: 131 _compat.reraise(self.exc_info[0], self.exc_info[1], self.exc_info[2]) 132 133 134 class NotAThread(_threading.Thread): 135 exc_info = None 136 _started_nota = False 137 138 def start(self): 139 self._started_nota = True 140 self.run() 141 142 def join(self, timeout = None): 143 if not self._started_nota: 144 raise RuntimeError('cannot join thread before it is started') 145 146 147 class TimerExt(_compat.TimerCls): 148 149 def __init__(self, interval, function, args = (), kwargs = None): 150 if float(interval) <= 0: 151 raise ValueError('interval must be > 0, got %s' % interval) 152 if function is None: 153 raise ValueError('function cannot be None.') 154 _compat.TimerCls.__init__(self, interval, function, args, kwargs or {}) 155 self._lock = _threading.Lock() 156 self._restartRequested = False 157 self.name = 'TimerExtThread' 158 159 def restart(self): 160 with self._lock: 161 if self._restartRequested: 162 return 163 if self.finished.isSet(): 164 raise RuntimeError('Thread has already finished, cannot be restarted.') 165 self._restartRequested = True 166 self.finished.set() 167 168 def run(self): 169 while True: 170 self.finished.wait(self.interval) 171 with self._lock: 172 timedOut = not self.finished.isSet() 173 if timedOut: 174 self.finished.set() 175 else: 176 cancelled = not self._restartRequested 177 if cancelled: 178 return 179 self.finished.clear() 180 self._restartRequested = False 181 if timedOut: 182 self.function(*self.args, **self.kwargs) 183 return 184 185 186 def join_timeout(thread, timeout = 8, errtype = RuntimeError): 187 thread.join(timeout) 188 if thread.is_alive(): 189 raise errtype('%s is still alive!' % thread) 190 191 192 class Token(object): 193 194 def __init__(self): 195 self._isSet = False 196 197 def set(self): 198 self._isSet = True 199 200 def is_set(self): 201 return self._isSet 202 203 204 def memoize(func = None, uselock = False, _lockcls = _dochelpers.ignore): 205 206 def hasParameterless(func_): 207 argspec = _inspect.getargspec(func_) 208 return any(argspec) 209 210 if not func and not uselock: 211 raise AssertionError('If not using lock, must provide func (decorate without params)') 212 cache = [] 213 if callable(func): 214 if hasParameterless(func): 215 raise ValueError('Function cannot have parameters.') 216 217 def inner(*args, **kwargs): 218 if not cache: 219 cache.append(func(*args, **kwargs)) 220 return cache[0] 221 222 return inner 223 lock = (_lockcls or _threading.Lock)() 224 225 def inner(func_): 226 227 def inner2(*args, **kwargs): 228 if not cache: 229 with lock: 230 if not cache: 231 cache.append(func_(*args, **kwargs)) 232 return cache[0] 233 234 return inner2 235 236 return inner 237 238 239 class expiring_memoize(object): 240 241 def __init__(self, expiry = 0, gettime = None, jitter = None): 242 if jitter is not None: 243 if jitter >= expiry: 244 raise ValueError('Jitter must be < expiry') 245 if expiry <= 0: 246 raise ValueError('Expiry must be > 0') 247 self.jitter = jitter 248 self.expiry = expiry 249 self.gettime = gettime or _time.time 250 self.random = random.Random() 251 252 def _get_next_expiry(self): 253 from_timestamp = self.gettime() 254 if self.jitter is None: 255 return from_timestamp + self.expiry 256 jitter_time = self.random.randint(-self.jitter, self.jitter) 257 jittered_expiry = self.expiry + jitter_time 258 return from_timestamp + jittered_expiry 259 260 def __call__(self, func): 261 262 def wrapped(*args): 263 try: 264 cache = func._cache 265 except AttributeError: 266 cache = func._cache = {} 267 268 try: 269 result, expiry_timestamp = cache[args] 270 if self.gettime() < expiry_timestamp: 271 return result 272 except KeyError: 273 pass 274 275 result = func(*args) 276 cache[args] = (result, self._get_next_expiry()) 277 return result 278 279 def clear_cache(): 280 try: 281 cache = func._cache 282 cache.clear() 283 except AttributeError: 284 pass 285 286 def prime_cache_result(args, result): 287 try: 288 cache = func._cache 289 except AttributeError: 290 cache = func._cache = {} 291 292 cache[args] = (result, self._get_next_expiry()) 293 294 def remove_from_cache(args): 295 try: 296 cache = func._cache 297 except AttributeError: 298 return 299 300 if args in cache: 301 del cache[args] 302 303 wrapped.clear_cache = clear_cache 304 wrapped.prime_cache_result = prime_cache_result 305 wrapped.remove_from_cache = remove_from_cache 306 return wrapped