| 94 | |
| 95 | @Singleton |
| 96 | class PendingUpload(object): |
| 97 | # keep a track of objects that we have created but haven't distributed yet |
| 98 | def __init__(self): |
| 99 | super(self.__class__, self).__init__() |
| 100 | self.lock = RLock() |
| 101 | self.hashes = {} |
| 102 | # end by this time in any case |
| 103 | self.deadline = 0 |
| 104 | self.maxLen = 0 |
| 105 | # during shutdown, wait up to 20 seconds to finish uploading |
| 106 | self.shutdownWait = 20 |
| 107 | # forget tracking objects after 60 seconds |
| 108 | self.objectWait = 60 |
| 109 | # wait 10 seconds between clears |
| 110 | self.clearDelay = 10 |
| 111 | self.lastCleared = time.time() |
| 112 | |
| 113 | def add(self, objectHash = None): |
| 114 | with self.lock: |
| 115 | # add a new object into existing thread lists |
| 116 | if objectHash: |
| 117 | if objectHash not in self.hashes: |
| 118 | self.hashes[objectHash] = {'created': time.time(), 'sendCount': 0, 'peers': []} |
| 119 | for thread in threadingEnumerate(): |
| 120 | if thread.isAlive() and hasattr(thread, 'peer') and \ |
| 121 | thread.peer not in self.hashes[objectHash]['peers']: |
| 122 | self.hashes[objectHash]['peers'].append(thread.peer) |
| 123 | # add all objects into the current thread |
| 124 | else: |
| 125 | for objectHash in self.hashes: |
| 126 | if current_thread().peer not in self.hashes[objectHash]['peers']: |
| 127 | self.hashes[objectHash]['peers'].append(current_thread().peer) |
| 128 | |
| 129 | def len(self): |
| 130 | self.clearHashes() |
| 131 | with self.lock: |
| 132 | return sum(1 |
| 133 | for x in self.hashes if (self.hashes[x]['created'] + self.objectWait < time.time() or |
| 134 | self.hashes[x]['sendCount'] == 0)) |
| 135 | |
| 136 | def _progress(self): |
| 137 | with self.lock: |
| 138 | return float(sum(len(self.hashes[x]['peers']) |
| 139 | for x in self.hashes if (self.hashes[x]['created'] + self.objectWait < time.time()) or |
| 140 | self.hashes[x]['sendCount'] == 0)) |
| 141 | |
| 142 | def progress(self, raiseDeadline=True): |
| 143 | if self.maxLen < self._progress(): |
| 144 | self.maxLen = self._progress() |
| 145 | if self.deadline < time.time(): |
| 146 | if self.deadline > 0 and raiseDeadline: |
| 147 | raise PendingUploadDeadlineException |
| 148 | self.deadline = time.time() + 20 |
| 149 | try: |
| 150 | return 1.0 - self._progress() / self.maxLen |
| 151 | except ZeroDivisionError: |
| 152 | return 1.0 |
| 153 |
no outgoing calls
no test coverage detected