__init__.py (22963B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\uthread\__init__.py 2 from __future__ import print_function 3 from future.utils import raise_ 4 __version__ = '0.1' 5 __license__ = 'Python Microthread Library version 0.1\nCopyright (C)2000 Will Ware, Christian Tismer\n\nPermission to use, copy, modify, and distribute this software and its\ndocumentation for any purpose and without fee is hereby granted,\nprovided that the above copyright notice appear in all copies and that\nboth that copyright notice and this permission notice appear in\nsupporting documentation, and that the names of the authors not be\nused in advertising or publicity pertaining to distribution of the\nsoftware without specific, written prior permission.\n\nWILL WARE AND CHRISTIAN TISMER DISCLAIM ALL WARRANTIES WITH REGARD TO\nTHIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND\nFITNESS. IN NO EVENT SHALL WILL WARE OR CHRISTIAN TISMER BE LIABLE FOR\nANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES\nWHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN\nACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT\nOF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.' 6 import stackless 7 import sys 8 import blue 9 import bluepy 10 import weakref 11 import collections 12 import contextlib 13 import logmodule as log 14 import blue.heapq as heapq 15 import locks 16 import threading 17 from stacklesslib.util import atomic 18 SEC = 10000000 19 MIN = 60 * SEC 20 tasks = [] 21 schedule = stackless.schedule 22 stackless.globaltrace = True 23 ctxtfilter = bluepy.ctxtfilter 24 25 def new(func, *args, **kw): 26 return bluepy.CreateTaskletExt(func, *args, **kw) 27 28 29 class TaskletExtTimeoutable(bluepy.TaskletExt): 30 __slots__ = bluepy.TaskletExt.__slots__ + ['doneChannel', 'isTimedOut'] 31 32 33 def newJoinable(func, *args, **kw): 34 35 def wrapper(selfTasklet, *args, **kw): 36 exce = None 37 result = None 38 try: 39 result = func(*args, **kw) 40 except Exception as e: 41 exce = sys.exc_info() 42 43 try: 44 if not selfTasklet.isTimedOut: 45 stackless.channel.send(selfTasklet.doneChannel, (result, exce)) 46 else: 47 log.general.Log("newJoinable timed out on function '" + str(func) + "' with args '" + strx(args)[:2048] + "'", log.LGWARN) 48 finally: 49 exce = None 50 51 return result 52 53 ctx = ctxtfilter.sub('at (snip)', repr(func)) 54 ctx = blue.pyos.taskletTimer.GetCurrent().split('^')[-1] + '^' + ctx 55 t = TaskletExtTimeoutable(ctx, wrapper) 56 t.isTimedOut = False 57 t.doneChannel = stackless.channel() 58 t(t, *args, **kw) 59 return t 60 61 62 def waitForJoinable(t, timeout): 63 resultOk, data = ChannelWait(t.doneChannel, timeout=timeout) 64 if resultOk: 65 result, exce = data 66 if exce is not None: 67 try: 68 raise_(exce[0], exce[1], exce[2]) 69 finally: 70 exce = None 71 72 return result 73 t.isTimedOut = True 74 raise TaskletWaitTimeout() 75 76 77 class TaskletWaitTimeout(Exception): 78 pass 79 80 81 idIndex = 0 82 83 def uniqueId(): 84 global idIndex 85 z = idIndex 86 idIndex += 1 87 return z 88 89 90 def irandom(n): 91 import random 92 n = random.randrange(0, n) 93 return n 94 95 96 semaphores = weakref.WeakKeyDictionary() 97 98 def GetSemaphores(): 99 return semaphores 100 101 102 class Semaphore: 103 __guid__ = 'uthread.Semaphore' 104 105 def __init__(self, semaphoreName = None, maxcount = 1, strict = True): 106 self.semaphoreName = semaphoreName 107 self.maxcount = maxcount 108 self.count = maxcount 109 self.waiting = stackless.channel() 110 self.n_waiting = 0 111 self.waiting.preference = 0 112 self.threads = [] 113 self.lockedWhen = None 114 self.strict = strict 115 locks.Register(self) 116 117 def IsCool(self): 118 return self.count == self.maxcount and not self.n_waiting 119 120 def __repr__(self): 121 return '<Semaphore %r, t:%f at %#x>' % (self.semaphoreName, self.LockedFor(), id(self)) 122 123 def __del__(self): 124 if not self.IsCool(): 125 log.general.Log("Semaphore '" + str(self.semaphoreName) + "' is being destroyed in a locked or waiting state", 4, 0) 126 127 def acquire(self): 128 with atomic(): 129 if self.strict: 130 if self.count <= 0 and stackless.getcurrent() in self.threads: 131 raise RuntimeError('tasklet deadlock, acquiring tasklet holds strict semaphore') 132 while self.count == 0: 133 self.n_waiting += 1 134 try: 135 self.waiting.receive() 136 except: 137 self._safe_pump() 138 raise 139 finally: 140 self.n_waiting -= 1 141 142 self.count -= 1 143 self.lockedWhen = blue.os.GetWallclockTime() 144 self.threads.append(stackless.getcurrent()) 145 146 claim = acquire 147 148 def try_acquire(self): 149 with atomic(): 150 if self.strict: 151 if self.count <= 0 and stackless.getcurrent() in self.threads: 152 raise RuntimeError('tasklet deadlock, acquiring tasklet holds strict semaphore') 153 if self.count > 0: 154 self.count -= 1 155 self.lockedWhen = blue.os.GetWallclockTime() 156 self.threads.append(stackless.getcurrent()) 157 return True 158 return False 159 160 def release(self, override = False): 161 with atomic(): 162 if self.strict and not override: 163 if stackless.getcurrent() not in self.threads: 164 raise RuntimeError('wrong tasklet releasing strict semaphore') 165 self.count += 1 166 self.threads.remove(stackless.getcurrent()) 167 self.lockedWhen = None 168 self._pump() 169 170 def _safe_pump(self): 171 try: 172 self._pump() 173 except Exception: 174 pass 175 176 def _pump(self): 177 for i in range(min(self.count, -self.waiting.balance)): 178 self.waiting.send(None) 179 180 def __enter__(self): 181 self.acquire() 182 183 def __exit__(self, exc, val, tb): 184 self.release() 185 186 def WaitingTasklets(self): 187 r = [] 188 first = next = self.waiting.queue 189 while next: 190 r.append(next) 191 next = next.next 192 if next == first: 193 break 194 195 return r 196 197 def HoldingTasklets(self): 198 return self.threads[:] 199 200 def LockedAt(self): 201 return self.lockedWhen 202 203 def LockedFor(self): 204 if self.lockedWhen: 205 return (blue.os.GetWallclockTime() - self.lockedWhen) * 1e-07 206 else: 207 return -1.0 208 209 210 class CriticalSection(Semaphore): 211 __guid__ = 'uthread.CriticalSection' 212 213 def __init__(self, semaphoreName = None, strict = True): 214 Semaphore.__init__(self, semaphoreName) 215 self.__reentrantRefs = 0 216 217 def __repr__(self): 218 return '<CriticalSection %r, t:%f at %#x>' % (self.semaphoreName, self.LockedFor(), id(self)) 219 220 def _owns(self): 221 if stackless.getcurrent() in self.threads: 222 return True 223 for t in self.threads: 224 if locks.Inherits(t): 225 return True 226 227 return False 228 229 def acquire(self): 230 if self.count <= 0 and self._owns(): 231 self.__reentrantRefs += 1 232 else: 233 Semaphore.acquire(self) 234 235 def try_acquire(self): 236 if self.count <= 0 and self._owns(): 237 self.__reentrantRefs += 1 238 return True 239 else: 240 return Semaphore.try_acquire(self) 241 242 def release(self): 243 if self.__reentrantRefs: 244 if not self._owns(): 245 raise RuntimeError('wrong tasklet releasing reentrant CriticalSection') 246 self.__reentrantRefs -= 1 247 else: 248 Semaphore.release(self) 249 250 251 def FNext(f): 252 first = stackless.getcurrent() 253 try: 254 cursor = first.next 255 while cursor != first: 256 if cursor.frame.f_back == f: 257 return FNext(cursor.frame) 258 cursor = cursor.next 259 260 return f 261 finally: 262 first = None 263 cursor = None 264 265 266 class RWLock(object): 267 __guid__ = 'uthread.RWLock' 268 269 def __init__(self, lockName = ''): 270 self.name = lockName 271 self.rchan = stackless.channel() 272 self.wchan = stackless.channel() 273 self.rchan.preference = self.wchan.preference = 0 274 self.state = 0 275 self.tasklets = [] 276 self.lockedWhen = None 277 locks.Register(self) 278 279 def __repr__(self): 280 return '<RWLock %r, state:%d, rdwait:%d, wrwait:%d, t:%f at %#x>' % (self.name, 281 self.state, 282 -self.rchan.balance, 283 -self.wchan.balance, 284 self.LockedFor(), 285 id(self)) 286 287 def RDLock(self): 288 if not self.TryRDLock(): 289 self.rchan.receive() 290 291 def TryRDLock(self): 292 if self.state >= 0: 293 if self.wchan.balance == 0 or stackless.getcurrent() in self.tasklets: 294 self.state += 1 295 self._AddTasklet() 296 return True 297 return False 298 else: 299 return self.TryWRLock() 300 301 def WRLock(self): 302 if not self.TryWRLock(): 303 if stackless.getcurrent() in self.tasklets: 304 raise RuntimeError('Deadlock. Trying to WRLock while holding a RDLock') 305 self.wchan.receive() 306 307 def TryWRLock(self): 308 if self.state == 0 or self.state < 0 and stackless.getcurrent() == self.tasklets[0]: 309 self.state -= 1 310 self._AddTasklet() 311 return True 312 return False 313 314 def Unlock(self, tasklet = None): 315 if tasklet is None: 316 tasklet = stackless.getcurrent() 317 try: 318 self.tasklets.remove(tasklet) 319 except ValueError: 320 raise RuntimeError('Trying to release a rwlock without a matching lock!') 321 322 if self.state > 0: 323 self.state -= 1 324 else: 325 self.state += 1 326 if self.state == 0: 327 self.lockedWhen = None 328 self._Pump() 329 330 def _AddTasklet(self, tasklet = None): 331 if not tasklet: 332 tasklet = stackless.getcurrent() 333 self.tasklets.append(tasklet) 334 if not self.lockedWhen: 335 self.lockedWhen = blue.os.GetWallclockTime() 336 337 def _Pump(self): 338 while True: 339 chan = self.wchan 340 if chan.balance: 341 if self.state == 0: 342 self.state = -1 343 self._AddTasklet(chan.queue) 344 chan.send(None) 345 return 346 else: 347 chan = self.rchan 348 if chan.balance and self.state >= 0: 349 self.state += 1 350 self._AddTasklet(chan.queue) 351 chan.send(None) 352 continue 353 return 354 355 def __enter__(self): 356 self.RDLock() 357 358 def __exit__(self, e, v, tb): 359 self.Unlock() 360 361 class WRCtxt(object): 362 363 def __init__(self, lock): 364 self.lock = lock 365 366 def __enter__(self): 367 self.lock.WRLock() 368 369 def __exit__(self, e, v, tb): 370 self.lock.Unlock() 371 372 def WRLocked(self): 373 return self.WRCtxt(self) 374 375 def RDLocked(self): 376 return self 377 378 def IsCool(self): 379 return self.state == 0 380 381 @property 382 def thread(self): 383 if self.tasklets: 384 return self.tasklets[0] 385 386 def WaitingTasklets(self): 387 r = [] 388 for chan in (self.rchan, self.wchan): 389 first = t = chan.queue 390 while t: 391 r.append(t) 392 t = t.next 393 if t is first: 394 break 395 396 return r 397 398 def HoldingTasklets(self): 399 return list(self.tasklets) 400 401 def LockedAt(self): 402 return self.lockedWhen 403 404 def IsWRLocked(self): 405 return self.state < 0 406 407 def IsRDLocked(self): 408 return self.state > 0 409 410 def LockedFor(self): 411 if self.lockedWhen: 412 return (blue.os.GetWallclockTime() - self.lockedWhen) * 1e-07 413 else: 414 return -1.0 415 416 417 channels = weakref.WeakKeyDictionary() 418 419 def GetChannels(): 420 global channels 421 return channels 422 423 424 class Channel(stackless.channel): 425 __guid__ = 'uthread.Channel' 426 427 def __new__(cls, channelName = None): 428 return stackless.channel.__new__(cls) 429 430 def __init__(self, channelName = None): 431 stackless.channel.__init__(self) 432 self.channelName = channelName 433 channels[self] = 1 434 435 436 class FIFO(object): 437 __slots__ = ('data',) 438 439 def __init__(self): 440 self.data = [[], []] 441 442 def push(self, v): 443 self.data[1].append(v) 444 445 def pop(self): 446 d = self.data 447 if not d[0]: 448 d.reverse() 449 d[0].reverse() 450 return d[0].pop() 451 452 def front(self): 453 d = self.data 454 if d[0]: 455 return d[0][-1] 456 return d[1][0] 457 458 def __contains__(self, o): 459 d = self.data 460 return o in d[0] or o in d[1] 461 462 def __len__(self): 463 d = self.data 464 return len(d[0]) + len(d[1]) 465 466 def clear(self): 467 self.data = [[], []] 468 469 def remove(self, o): 470 d = self.data 471 if d[0]: 472 d[0].reverse() 473 d[1] = d[0] + d[1] 474 d[0] = [] 475 d[1].remove(o) 476 477 478 class Queue(FIFO): 479 __guid__ = 'uthread.Queue' 480 __slots__ = 'chan' 481 482 def __init__(self, preference = 0): 483 FIFO.__init__(self) 484 self.chan = stackless.channel() 485 self.chan.preference = preference 486 487 def put(self, x): 488 if self.chan.balance < 0: 489 self.chan.send(x) 490 else: 491 self.push(x) 492 493 non_blocking_put = put 494 495 def get(self): 496 if len(self): 497 return self.pop() 498 else: 499 return self.chan.receive() 500 501 502 def LockCheck(): 503 while 1: 504 blue.pyos.synchro.SleepSim(60660) 505 now = blue.os.GetWallclockTime() 506 try: 507 for each in locks.GetLocks(): 508 if each.LockedAt() and each.WaitingTasklets(): 509 problem = now - each.LockedAt() 510 if problem >= 1 * MIN: 511 break 512 else: 513 problem = 0 514 515 if not problem: 516 continue 517 with log.general.open(log.LGERR) as s: 518 print('Locks have been held for a long time (%ss). Locking conflict log' % (problem / SEC), file=s) 519 foundCycles = locks.LockCycleReport(out=s, timeLimit=60) 520 if not foundCycles: 521 print('logical analysis found no cycles.', file=s) 522 if not foundCycles: 523 print('Full dump of locks with waiting tasklets:', file=s) 524 locks.OldLockReport(None, out=s) 525 print('End of locking conflict log', file=s) 526 except Exception: 527 log.LogException() 528 sys.exc_clear() 529 530 531 def PoolWorker(ctx, func, *args, **keywords): 532 return PoolWithoutTheStars(ctx, func, args, keywords, True) 533 534 535 def PoolWorkerWithoutTheStars(ctx, func, args, keywords): 536 return PoolWithoutTheStars(ctx, func, args, keywords, True) 537 538 539 def PoolWithoutTheStars(ctx, func, args, kw, notls = False): 540 if not ctx: 541 ctx = ctxtfilter.sub('at (snip)', repr(func)) 542 if ctx[0] != '^': 543 prefix = blue.pyos.taskletTimer.GetCurrent().split('^')[-1] 544 ctx = prefix + '^' + ctx 545 stackless_tracing_enabled, kw = bluepy.tracing_enabled(**kw) 546 tasklet = bluepy.TaskletExt(ctx, func, stackless_tracing_enabled=stackless_tracing_enabled) 547 if notls: 548 tasklet.localStorage.clear() 549 tasklet(*args, **kw) 550 return tasklet 551 552 553 def Pool(ctx, func, *args, **keywords): 554 return PoolWithoutTheStars(ctx, func, args, keywords) 555 556 557 def ParallelHelper(ch, idx, what): 558 queue, parentTasklet = ch 559 try: 560 with locks.Inheritance(parentTasklet): 561 if len(what) > 2: 562 result = what[0](*what[1], **what[2]) 563 else: 564 result = what[0](*what[1]) 565 ret = (1, (idx, result)) 566 except: 567 ret = (0, sys.exc_info()) 568 569 queue.put(ret) 570 571 572 def Parallel(funcs, exceptionHandler = None, maxcount = 30, contextSuffix = None, funcContextSuffixes = None): 573 if not funcs: 574 return 575 context = blue.pyos.taskletTimer.GetCurrent() 576 if contextSuffix: 577 context += '::' + contextSuffix 578 ch = (Queue(preference=-1), stackless.getcurrent()) 579 n = len(funcs) 580 ret = [None] * n 581 if n > maxcount: 582 n = maxcount 583 for i in xrange(n): 584 funcContext = context 585 if funcContextSuffixes is not None: 586 funcContext = funcContext + '::' + funcContextSuffixes[i] 587 Pool(funcContext, ParallelHelper, ch, i, funcs[i]) 588 589 error = None 590 try: 591 for i in xrange(len(funcs)): 592 ok, bunch = ch[0].get() 593 if ok: 594 idx, val = bunch 595 ret[idx] = val 596 else: 597 error = bunch 598 if n < len(funcs): 599 funcContext = context 600 if funcContextSuffixes is not None: 601 funcContext = funcContext + '::' + funcContextSuffixes[i] 602 Pool(funcContext, ParallelHelper, ch, n, funcs[n]) 603 n += 1 604 605 if error: 606 if exceptionHandler: 607 exceptionHandler(error[1]) 608 else: 609 raise_(error[0], error[1], error[2]) 610 return ret 611 finally: 612 del error 613 614 615 class TaskletSequencer(object): 616 617 def __init__(self, expected = None): 618 self.queue = [] 619 self.expected = expected 620 self.lastThrough = None 621 self.closed = False 622 623 def State(self): 624 return [self.expected, self.lastThrough, self.closed] 625 626 def IsClosed(self): 627 return self.closed 628 629 def close(self): 630 self.closed = True 631 while self.queue: 632 heapq.heappop(self.queue)[1].insert() 633 634 def Assert(self, expression): 635 if not expression: 636 raise AssertionError(expression) 637 638 def Pass(self, sequenceNo): 639 if self.closed: 640 return 641 if self.expected is None: 642 self.expected = sequenceNo 643 if sequenceNo < self.expected: 644 return 645 while sequenceNo > self.expected: 646 me = (sequenceNo, stackless.getcurrent()) 647 heapq.heappush(self.queue, me) 648 self.OnSleep(sequenceNo) 649 stackless.schedule_remove() 650 self.OnWakeUp(sequenceNo) 651 if self.closed: 652 return 653 654 self.Assert(self.expected == sequenceNo) 655 self.expected += 1 656 expected = self.expected 657 while self.queue and self.queue[0][0] == expected: 658 self.OnWakingUp(sequenceNo, expected) 659 expected += 1 660 other = heapq.heappop(self.queue) 661 other[1].insert() 662 663 if self.lastThrough is not None: 664 self.Assert(self.lastThrough + 1 == sequenceNo) 665 self.lastThrough = sequenceNo 666 667 def OnSleep(self, sequenceNo): 668 pass 669 670 def OnWakeUp(self, sequenceNo): 671 pass 672 673 def OnWakingUp(self, sequenceNo, target): 674 pass 675 676 677 def CallOnThread(cmd, args = (), kwds = {}): 678 chan = stackless.channel() 679 680 def Helper(): 681 try: 682 r = cmd(*args, **kwds) 683 chan.send(r) 684 except: 685 e, v = sys.exc_info()[:2] 686 chan.send_exception(e, v) 687 finally: 688 blue.pyos.NextScheduledEvent(0) 689 690 thread = threading.Thread(target=Helper) 691 thread.start() 692 return chan.receive() 693 694 695 namedlocks = collections.defaultdict(lambda : [None, 0]) 696 697 def Lock(object, *args): 698 t = (id(object), args) 699 l = namedlocks[t] 700 if not l[0]: 701 l[0] = Semaphore(t, strict=False) 702 l[1] += 1 703 l[0].acquire() 704 705 706 def TryLock(object, *args): 707 t = (id(object), args) 708 l = namedlocks[t] 709 if not l[0]: 710 l[0] = Semaphore(t, strict=False) 711 l[1] += 1 712 if not l[0].try_acquire(): 713 l[1] -= 1 714 return False 715 return True 716 717 718 def ReentrantLock(object, *args): 719 t = (id(object), args) 720 l = namedlocks[t] 721 if not l[0]: 722 l[0] = CriticalSection(t) 723 l[1] += 1 724 l[0].acquire() 725 726 727 def UnLock(object, *args): 728 t = (id(object), args) 729 l = namedlocks[t] 730 l[0].release() 731 l[1] -= 1 732 if not l[1]: 733 del namedlocks[t] 734 735 736 def CheckLock(object, *args): 737 t = (id(object), args) 738 l = namedlocks[t] 739 if not l[1]: 740 del namedlocks[t] 741 return False 742 return True 743 744 745 @contextlib.contextmanager 746 def Locked(object, *args): 747 Lock(object, *args) 748 try: 749 yield 750 finally: 751 UnLock(object, *args) 752 753 754 class BlockTrapSection(object): 755 __guid__ = 'uthread.BlockTrapSection' 756 757 def __init__(self): 758 self.oldBlocktrap = False 759 760 def __enter__(self): 761 self.oldBlockTrap = stackless.getcurrent().block_trap 762 stackless.getcurrent().block_trap = True 763 764 def __exit__(self, exc, val, tb): 765 stackless.getcurrent().block_trap = self.oldBlockTrap 766 767 768 def ChannelWait(chan, timeout = None): 769 if timeout is None: 770 return (True, chan.receive()) 771 waiting_tasklet = stackless.getcurrent() 772 773 def break_wait(): 774 try: 775 blue.pyos.synchro.SleepWallclock(int(timeout * 1000)) 776 except _TimeoutError: 777 return 778 779 with atomic(): 780 if waiting_tasklet and waiting_tasklet.blocked: 781 waiting_tasklet.raise_exception(_TimeoutError) 782 783 with atomic(): 784 breaker = new(break_wait) 785 try: 786 result = chan.receive() 787 if breaker.blocked: 788 breaker.raise_exception(_TimeoutError) 789 return (True, result) 790 except _TimeoutError: 791 return (False, None) 792 finally: 793 waiting_tasklet = None 794 795 796 class _TimeoutError(Exception): 797 pass 798 799 800 parallel = Parallel 801 worker = PoolWorker 802 workerWithoutTheStars = PoolWorkerWithoutTheStars 803 pool = Pool 804 exports = {'uthread.parallel': parallel, 805 'uthread.worker': worker, 806 'uthread.workerWithoutTheStars': workerWithoutTheStars, 807 'uthread.new': new, 808 'uthread.pool': Pool, 809 'uthread.irandom': irandom, 810 'uthread.uniqueId': uniqueId, 811 'uthread.schedule': schedule, 812 'uthread.GetChannels': GetChannels, 813 'uthread.GetSemaphores': GetSemaphores, 814 'uthread.FNext': FNext, 815 'uthread.Lock': Lock, 816 'uthread.TryLock': TryLock, 817 'uthread.ReentrantLock': ReentrantLock, 818 'uthread.UnLock': UnLock, 819 'uthread.TaskletSequencer': TaskletSequencer, 820 'uthread.CallOnThread': CallOnThread, 821 'uthread.CheckLock': CheckLock, 822 'uthread.BlockTrapSection': BlockTrapSection, 823 'uthread.ChannelWait': ChannelWait, 824 'uthread.newJoinable': newJoinable, 825 'uthread.waitForJoinable': waitForJoinable, 826 'uthread.TaskletWaitTimeout': TaskletWaitTimeout} 827 new(LockCheck).context = '^uthread::LockCheck' 828 locks.Startup(new)