Better tracking in downloading objects

- remember what was requested from which node
- remember if it was received
- re-request object if we haven't received any new object for more than
  a minute
This commit is contained in:
Peter Šurda 2017-01-16 15:17:23 +01:00
parent 8dfa0faca9
commit ca6bc9981c
Signed by untrusted user: PeterSurda
GPG Key ID: 0C5F50C0B5F37D87
2 changed files with 51 additions and 21 deletions

View File

@ -217,7 +217,6 @@ class receiveDataThread(threading.Thread):
self.data = self.data[payloadLength + protocol.Header.size:] # take this message out and then process the next message self.data = self.data[payloadLength + protocol.Header.size:] # take this message out and then process the next message
if self.data == '': # if there are no more messages if self.data == '': # if there are no more messages
if Missing().len() > 0 and self.sendDataThreadQueue.qsize() < 10:
for objectHash in Missing().pull(100): for objectHash in Missing().pull(100):
if self.sendDataThreadQueue.full(): if self.sendDataThreadQueue.full():
break break

View File

@ -37,7 +37,7 @@ class Inventory(collections.MutableMapping):
value = self.InventoryItem(*value) value = self.InventoryItem(*value)
self._inventory[hash] = value self._inventory[hash] = value
self._streams[value.stream].add(hash) self._streams[value.stream].add(hash)
Missing().delete(hash) Missing().delete(hash, True)
def __delitem__(self, hash): def __delitem__(self, hash):
raise NotImplementedError raise NotImplementedError
@ -92,15 +92,25 @@ class Missing(object):
self.hashes = {} self.hashes = {}
self.stopped = False self.stopped = False
self.frequency = 60 self.frequency = 60
self.pending = {}
def add(self, objectHash): def add(self, objectHash):
if self.stopped: if self.stopped:
return return
with self.lock: with self.lock:
if not objectHash in self.hashes: if objectHash not in self.hashes:
self.hashes[objectHash] = {'peers':[], 'requested':0} self.hashes[objectHash] = {'peers':[], 'requested':0}
self.hashes[objectHash]['peers'].append(current_thread().peer) self.hashes[objectHash]['peers'].append(current_thread().peer)
def addPending(self, objectHash=None):
if self.stopped:
return
if current_thread().peer not in self.pending:
self.pending[current_thread().peer] = {'objects':[], 'requested':0, 'received':0}
if objectHash not in self.pending[current_thread().peer]['objects'] and not objectHash is None:
self.pending[current_thread().peer]['objects'].append(objectHash)
self.pending[current_thread().peer]['requested'] = time.time()
def len(self): def len(self):
with self.lock: with self.lock:
return len(self.hashes) return len(self.hashes)
@ -122,33 +132,54 @@ class Missing(object):
if count < 1: if count < 1:
raise ValueError("Must be at least one") raise ValueError("Must be at least one")
objectHashes = [] objectHashes = []
if self.stopped:
return objectHashes
try: try:
for objectHash in self.hashes.keys(): for objectHash in self.hashes.keys():
if self.hashes[objectHash]['requested'] < time.time() - self.frequency:
if len(self.hashes[objectHash]['peers']) == 0:
self.removeObjectFromCurrentThread(objectHash)
continue
if current_thread().peer in self.hashes[objectHash]['peers']:
objectHashes.append(objectHash)
self.hashes[objectHash]['requested'] = time.time()
self.removeObjectFromCurrentThread(objectHash)
if len(objectHashes) >= count: if len(objectHashes) >= count:
break break
if current_thread().peer not in self.pending:
self.addPending()
if (self.pending[current_thread().peer]['requested'] >= time.time() - self.frequency or \
self.pending[current_thread().peer]['received'] >= time.time() - self.frequency) and \
len(self.pending[current_thread().peer]['objects']) >= count:
break
# requested too long ago or not at all
if self.hashes[objectHash]['requested'] < time.time() - self.frequency:
# already requested from this thread but haven't received yet
if objectHash in self.pending[current_thread().peer]['objects']:
if self.pending[current_thread().peer]['requested'] >= time.time() - self.frequency or self.pending[current_thread().peer]['received'] >= time.time() - self.frequency:
continue
elif current_thread().peer not in self.hashes[objectHash]['peers']:
continue
objectHashes.append(objectHash)
self.hashes[objectHash]['requested'] = time.time()
self.pending[current_thread().peer]['requested'] = time.time()
self.addPending(objectHash)
except (RuntimeError, KeyError, ValueError): except (RuntimeError, KeyError, ValueError):
# the for cycle sometimes breaks if you remove elements # the for cycle sometimes breaks if you remove elements
pass pass
return objectHashes return objectHashes
def delete(self, objectHash): def delete(self, objectHash, justReceived=False):
with self.lock: with self.lock:
if objectHash in self.hashes: if objectHash in self.hashes:
del self.hashes[objectHash] del self.hashes[objectHash]
if objectHash in self.pending[current_thread().peer]['objects']:
self.pending[current_thread().peer]['objects'].remove(objectHash)
if justReceived:
self.pending[current_thread().peer]['received'] = time.time()
def stop(self): def stop(self):
with self.lock: with self.lock:
self.hashes = {} self.hashes = {}
self.pending = {}
def threadEnd(self): def threadEnd(self):
with self.lock: with self.lock:
for objectHash in self.hashes: for objectHash in self.hashes:
self.removeObjectFromCurrentThread(objectHash) self.removeObjectFromCurrentThread(objectHash)
try:
del self.pending[current_thread().peer]
except KeyError:
pass