Objects to be downloaded fixes
- tries to avoid calling senddata it it would block receiveDataThread, allowing fore more asynchronous operation - request objects in chunks of 100 (CPU performance optimisation)
This commit is contained in:
parent
f079ff5b99
commit
dbe15d0b99
|
@ -217,16 +217,16 @@ 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
|
||||||
while Missing().len() > 0 and not self.sendDataThreadQueue.full():
|
if Missing().len() > 0 and self.sendDataThreadQueue.qsize() < 10:
|
||||||
objectHash = Missing().pull()
|
for objectHash in Missing().pull(100):
|
||||||
if objectHash is None:
|
if self.sendDataThreadQueue.full():
|
||||||
break
|
break
|
||||||
if objectHash in Inventory():
|
if objectHash in Inventory():
|
||||||
logger.debug('Inventory already has object listed in inv message.')
|
logger.debug('Inventory already has object listed in inv message.')
|
||||||
Missing().delete(objectHash)
|
Missing().delete(objectHash)
|
||||||
else:
|
else:
|
||||||
# We don't have the object in our inventory. Let's request it.
|
# We don't have the object in our inventory. Let's request it.
|
||||||
self.sendgetdata(objectHash)
|
self.sendgetdata(objectHash)
|
||||||
self.processData()
|
self.processData()
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -101,23 +101,29 @@ class Missing(object):
|
||||||
with self.lock:
|
with self.lock:
|
||||||
return len(self.hashes)
|
return len(self.hashes)
|
||||||
|
|
||||||
def pull(self):
|
def pull(self, count=1):
|
||||||
|
if count < 1:
|
||||||
|
raise ValueError("Must be at least one")
|
||||||
with self.lock:
|
with self.lock:
|
||||||
now = time.time()
|
now = time.time()
|
||||||
since = now - 300 # once every 5 minutes
|
since = now - 300 # once every 5 minutes
|
||||||
try:
|
try:
|
||||||
objectHash = random.choice({k:v for k, v in self.hashes.iteritems() if current_thread().peer in self.hashes[k]['peers'] and self.hashes[k]['requested'] < since}.keys())
|
matchingHashes = {k:v for k, v in self.hashes.iteritems() if current_thread().peer in self.hashes[k]['peers'] and self.hashes[k]['requested'] < since}
|
||||||
|
if count > len(matchingHashes):
|
||||||
|
count = len(matchingHashes)
|
||||||
|
objectHashes = random.sample(matchingHashes, count)
|
||||||
except (IndexError, KeyError): # list is empty
|
except (IndexError, KeyError): # list is empty
|
||||||
return None
|
return None
|
||||||
try:
|
for objectHash in objectHashes:
|
||||||
self.hashes[objectHash]['peers'].remove(current_thread().peer)
|
try:
|
||||||
except ValueError:
|
self.hashes[objectHash]['peers'].remove(current_thread().peer)
|
||||||
pass
|
except ValueError:
|
||||||
if len(self.hashes[objectHash]['peers']) == 0:
|
pass
|
||||||
self.delete(objectHash)
|
if len(self.hashes[objectHash]['peers']) == 0:
|
||||||
else:
|
self.delete(objectHash)
|
||||||
self.hashes[objectHash]['requested'] = now
|
else:
|
||||||
return objectHash
|
self.hashes[objectHash]['requested'] = now
|
||||||
|
return objectHashes
|
||||||
|
|
||||||
def delete(self, objectHash):
|
def delete(self, objectHash):
|
||||||
with self.lock:
|
with self.lock:
|
||||||
|
|
Reference in New Issue
Block a user