wspbus.py (7340B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\process\wspbus.py 2 import atexit 3 import os 4 import sys 5 import threading 6 import time 7 import traceback as _traceback 8 import warnings 9 from cherrypy._cpcompat import set 10 _startup_cwd = os.getcwd() 11 12 class ChannelFailures(Exception): 13 delimiter = '\n' 14 15 def __init__(self, *args, **kwargs): 16 Exception.__init__(self, *args, **kwargs) 17 self._exceptions = list() 18 19 def handle_exception(self): 20 self._exceptions.append(sys.exc_info()) 21 22 def get_instances(self): 23 return [ instance for cls, instance, traceback in self._exceptions ] 24 25 def __str__(self): 26 exception_strings = map(repr, self.get_instances()) 27 return self.delimiter.join(exception_strings) 28 29 __repr__ = __str__ 30 31 def __nonzero__(self): 32 return bool(self._exceptions) 33 34 35 class _StateEnum(object): 36 37 class State(object): 38 name = None 39 40 def __repr__(self): 41 return 'states.%s' % self.name 42 43 def __setattr__(self, key, value): 44 if isinstance(value, self.State): 45 value.name = key 46 object.__setattr__(self, key, value) 47 48 49 states = _StateEnum() 50 states.STOPPED = states.State() 51 states.STARTING = states.State() 52 states.STARTED = states.State() 53 states.STOPPING = states.State() 54 states.EXITING = states.State() 55 56 class Bus(object): 57 states = states 58 state = states.STOPPED 59 execv = False 60 61 def __init__(self): 62 self.execv = False 63 self.state = states.STOPPED 64 self.listeners = dict([ (channel, set()) for channel in ('start', 'stop', 'exit', 'graceful', 'log', 'main') ]) 65 self._priorities = {} 66 67 def subscribe(self, channel, callback, priority = None): 68 if channel not in self.listeners: 69 self.listeners[channel] = set() 70 self.listeners[channel].add(callback) 71 if priority is None: 72 priority = getattr(callback, 'priority', 50) 73 self._priorities[channel, callback] = priority 74 75 def unsubscribe(self, channel, callback): 76 listeners = self.listeners.get(channel) 77 if listeners and callback in listeners: 78 listeners.discard(callback) 79 del self._priorities[channel, callback] 80 81 def publish(self, channel, *args, **kwargs): 82 if channel not in self.listeners: 83 return [] 84 exc = ChannelFailures() 85 output = [] 86 items = [ (self._priorities[channel, listener], listener) for listener in self.listeners[channel] ] 87 items.sort() 88 for priority, listener in items: 89 try: 90 output.append(listener(*args, **kwargs)) 91 except KeyboardInterrupt: 92 raise 93 except SystemExit as e: 94 if exc and e.code == 0: 95 e.code = 1 96 raise 97 except: 98 exc.handle_exception() 99 if channel == 'log': 100 pass 101 else: 102 self.log('Error in %r listener %r' % (channel, listener), level=40, traceback=True) 103 104 if exc: 105 raise exc 106 return output 107 108 def _clean_exit(self): 109 if self.state != states.EXITING: 110 warnings.warn('The main thread is exiting, but the Bus is in the %r state; shutting it down automatically now. You must either call bus.block() after start(), or call bus.exit() before the main thread exits.' % self.state, RuntimeWarning) 111 self.exit() 112 113 def start(self): 114 atexit.register(self._clean_exit) 115 self.state = states.STARTING 116 self.log('Bus STARTING') 117 try: 118 self.publish('start') 119 self.state = states.STARTED 120 self.log('Bus STARTED') 121 except (KeyboardInterrupt, SystemExit): 122 raise 123 except: 124 self.log('Shutting down due to error in start listener:', level=40, traceback=True) 125 e_info = sys.exc_info() 126 try: 127 self.exit() 128 except: 129 pass 130 131 raise e_info[0], e_info[1], e_info[2] 132 133 def exit(self): 134 exitstate = self.state 135 try: 136 self.stop() 137 self.state = states.EXITING 138 self.log('Bus EXITING') 139 self.publish('exit') 140 self.log('Bus EXITED') 141 except: 142 os._exit(70) 143 144 if exitstate == states.STARTING: 145 os._exit(70) 146 147 def restart(self): 148 self.execv = True 149 self.exit() 150 151 def graceful(self): 152 self.log('Bus graceful') 153 self.publish('graceful') 154 155 def block(self, interval = 0.1): 156 try: 157 self.wait(states.EXITING, interval=interval, channel='main') 158 except (KeyboardInterrupt, IOError): 159 self.log('Keyboard Interrupt: shutting down bus') 160 self.exit() 161 except SystemExit: 162 self.log('SystemExit raised: shutting down bus') 163 self.exit() 164 raise 165 166 self.log('Waiting for child threads to terminate...') 167 for t in threading.enumerate(): 168 if t != threading.currentThread() and t.isAlive(): 169 if hasattr(threading.Thread, 'daemon'): 170 d = t.daemon 171 else: 172 d = t.isDaemon() 173 if not d: 174 self.log('Waiting for thread %s.' % t.getName()) 175 t.join() 176 177 if self.execv: 178 self._do_execv() 179 180 def wait(self, state, interval = 0.1, channel = None): 181 if isinstance(state, (tuple, list)): 182 states = state 183 else: 184 states = [state] 185 186 def _wait(): 187 while self.state not in states: 188 time.sleep(interval) 189 self.publish(channel) 190 191 try: 192 sys.modules['psyco'].cannotcompile(_wait) 193 except (KeyError, AttributeError): 194 pass 195 196 _wait() 197 198 def _do_execv(self): 199 args = sys.argv[:] 200 self.log('Re-spawning %s' % ' '.join(args)) 201 if sys.platform[:4] == 'java': 202 from _systemrestart import SystemRestart 203 raise SystemRestart 204 else: 205 args.insert(0, sys.executable) 206 if sys.platform == 'win32': 207 args = [ '"%s"' % arg for arg in args ] 208 os.chdir(_startup_cwd) 209 os.execv(sys.executable, args) 210 211 def stop(self): 212 self.state = states.STOPPING 213 self.log('Bus STOPPING') 214 self.publish('stop') 215 self.state = states.STOPPED 216 self.log('Bus STOPPED') 217 218 def start_with_callback(self, func, args = None, kwargs = None): 219 if args is None: 220 args = () 221 if kwargs is None: 222 kwargs = {} 223 args = (func,) + args 224 225 def _callback(func, *a, **kw): 226 self.wait(states.STARTED) 227 func(*a, **kw) 228 229 t = threading.Thread(target=_callback, args=args, kwargs=kwargs) 230 t.setName('Bus Callback ' + t.getName()) 231 t.start() 232 self.start() 233 return t 234 235 def log(self, msg = '', level = 20, traceback = False): 236 if traceback: 237 exc = sys.exc_info() 238 msg += '\n' + ''.join(_traceback.format_exception(*exc)) 239 self.publish('log', msg, level) 240 241 242 bus = Bus()