| 38 | |
| 39 | |
| 40 | class PendingDownloadQueue(Queue.Queue): |
| 41 | # keep a track of objects that have been advertised to us but we haven't downloaded them yet |
| 42 | maxWait = 300 |
| 43 | |
| 44 | def __init__(self, maxsize=0): |
| 45 | Queue.Queue.__init__(self, maxsize) |
| 46 | self.stopped = False |
| 47 | self.pending = {} |
| 48 | self.lock = RLock() |
| 49 | |
| 50 | def task_done(self, hashId): |
| 51 | Queue.Queue.task_done(self) |
| 52 | try: |
| 53 | with self.lock: |
| 54 | del self.pending[hashId] |
| 55 | except KeyError: |
| 56 | pass |
| 57 | |
| 58 | def get(self, block=True, timeout=None): |
| 59 | retval = Queue.Queue.get(self, block, timeout) |
| 60 | # no exception was raised |
| 61 | if not self.stopped: |
| 62 | with self.lock: |
| 63 | self.pending[retval] = time.time() |
| 64 | return retval |
| 65 | |
| 66 | def clear(self): |
| 67 | with self.lock: |
| 68 | newPending = {} |
| 69 | for hashId in self.pending: |
| 70 | if self.pending[hashId] + PendingDownloadQueue.maxWait > time.time(): |
| 71 | newPending[hashId] = self.pending[hashId] |
| 72 | self.pending = newPending |
| 73 | |
| 74 | @staticmethod |
| 75 | def totalSize(): |
| 76 | size = 0 |
| 77 | for thread in threadingEnumerate(): |
| 78 | if thread.isAlive() and hasattr(thread, 'downloadQueue'): |
| 79 | size += thread.downloadQueue.qsize() + len(thread.downloadQueue.pending) |
| 80 | return size |
| 81 | |
| 82 | @staticmethod |
| 83 | def stop(): |
| 84 | for thread in threadingEnumerate(): |
| 85 | if thread.isAlive() and hasattr(thread, 'downloadQueue'): |
| 86 | thread.downloadQueue.stopped = True |
| 87 | with thread.downloadQueue.lock: |
| 88 | thread.downloadQueue.pending = {} |
| 89 | |
| 90 | |
| 91 | class PendingUploadDeadlineException(Exception): |