dummy_thread.py (1898B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\dummy_thread.py 2 __all__ = ['error', 3 'start_new_thread', 4 'exit', 5 'get_ident', 6 'allocate_lock', 7 'interrupt_main', 8 'LockType'] 9 import traceback as _traceback 10 11 class error(Exception): 12 13 def __init__(self, *args): 14 self.args = args 15 16 17 def start_new_thread(function, args, kwargs = {}): 18 global _interrupt 19 global _main 20 if type(args) != type(tuple()): 21 raise TypeError('2nd arg must be a tuple') 22 if type(kwargs) != type(dict()): 23 raise TypeError('3rd arg must be a dict') 24 _main = False 25 try: 26 function(*args, **kwargs) 27 except SystemExit: 28 pass 29 except: 30 _traceback.print_exc() 31 32 _main = True 33 if _interrupt: 34 _interrupt = False 35 raise KeyboardInterrupt 36 37 38 def exit(): 39 raise SystemExit 40 41 42 def get_ident(): 43 return -1 44 45 46 def allocate_lock(): 47 return LockType() 48 49 50 def stack_size(size = None): 51 if size is not None: 52 raise error('setting thread stack size not supported') 53 return 0 54 55 56 class LockType(object): 57 58 def __init__(self): 59 self.locked_status = False 60 61 def acquire(self, waitflag = None): 62 if waitflag is None or waitflag: 63 self.locked_status = True 64 return True 65 elif not self.locked_status: 66 self.locked_status = True 67 return True 68 else: 69 return False 70 71 __enter__ = acquire 72 73 def __exit__(self, typ, val, tb): 74 self.release() 75 76 def release(self): 77 if not self.locked_status: 78 raise error 79 self.locked_status = False 80 return True 81 82 def locked(self): 83 return self.locked_status 84 85 86 _interrupt = False 87 _main = True 88 89 def interrupt_main(): 90 global _interrupt 91 if _main: 92 raise KeyboardInterrupt 93 else: 94 _interrupt = True