(self)
| 37 | self.lastCleaned = time.time() |
| 38 | |
| 39 | def run(self): |
| 40 | while not self._stopped: |
| 41 | requested = 0 |
| 42 | # Choose downloading peers randomly |
| 43 | connections = [x for x in BMConnectionPool().inboundConnections.values() + BMConnectionPool().outboundConnections.values() if x.fullyEstablished] |
| 44 | random.shuffle(connections) |
| 45 | try: |
| 46 | requestChunk = max(int(min(DownloadThread.maxRequestChunk, len(missingObjects)) / len(connections)), 1) |
| 47 | except ZeroDivisionError: |
| 48 | requestChunk = 1 |
| 49 | for i in connections: |
| 50 | now = time.time() |
| 51 | try: |
| 52 | request = i.objectsNewToMe.randomKeys(requestChunk) |
| 53 | except KeyError: |
| 54 | continue |
| 55 | payload = bytearray() |
| 56 | payload.extend(addresses.encodeVarint(len(request))) |
| 57 | for chunk in request: |
| 58 | if chunk in Inventory() and not Dandelion().hasHash(chunk): |
| 59 | try: |
| 60 | del i.objectsNewToMe[chunk] |
| 61 | except KeyError: |
| 62 | pass |
| 63 | continue |
| 64 | payload.extend(chunk) |
| 65 | missingObjects[chunk] = now |
| 66 | if not payload: |
| 67 | continue |
| 68 | i.append_write_buf(protocol.CreatePacket('getdata', payload)) |
| 69 | logger.debug("%s:%i Requesting %i objects", i.destination.host, i.destination.port, len(request)) |
| 70 | requested += len(request) |
| 71 | if time.time() >= self.lastCleaned + DownloadThread.cleanInterval: |
| 72 | self.cleanPending() |
| 73 | if not requested: |
| 74 | self.stop.wait(1) |
nothing calls this directly
no test coverage detected