Objects to be downloaded optimising

- treat requested but never arrived objects as expired. This is how it
  worked before the refactoring. Without this, the list won't go to
  zero.
This commit is contained in:
Peter Šurda 2017-01-15 22:01:10 +01:00
parent 79893fdc23
commit 94f0bdc731
Signed by: PeterSurda
GPG Key ID: 0C5F50C0B5F37D87
2 changed files with 16 additions and 17 deletions

View File

@ -1,5 +1,4 @@
import collections import collections
import random
from threading import current_thread, RLock from threading import current_thread, RLock
import time import time
@ -38,6 +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)
def __delitem__(self, hash): def __delitem__(self, hash):
raise NotImplementedError raise NotImplementedError
@ -101,18 +101,27 @@ class Missing(object):
with self.lock: with self.lock:
return len(self.hashes) return len(self.hashes)
def removeObjectFromCurrentThread(self, objectHash):
with self.lock:
try:
self.hashes[objectHash]['peers'].remove(current_thread().peer)
except ValueError:
pass
if len(self.hashes[objectHash]['peers']) == 0:
self.delete(objectHash)
else:
self.hashes[objectHash]['requested'] = time.time()
def pull(self, count=1): def pull(self, count=1):
if count < 1: if count < 1:
raise ValueError("Must be at least one") raise ValueError("Must be at least one")
with self.lock: with self.lock:
now = time.time() since = time.time() - 60 # once every minute
since = now - 60 # once every minute
objectHashes = [] objectHashes = []
for objectHash in self.hashes.keys(): for objectHash in self.hashes.keys():
if current_thread().peer in self.hashes[objectHash]['peers'] and self.hashes[objectHash]['requested'] < since: if current_thread().peer in self.hashes[objectHash]['peers'] and self.hashes[objectHash]['requested'] < since:
objectHashes.append(objectHash) objectHashes.append(objectHash)
self.hashes[objectHash]['peers'].remove(current_thread().peer) self.removeObjectFromCurrentThread(objectHash)
self.hashes[objectHash]['requested'] = now
if len(objectHashes) >= count: if len(objectHashes) >= count:
break break
return objectHashes return objectHashes
@ -125,9 +134,4 @@ class Missing(object):
def threadEnd(self): def threadEnd(self):
with self.lock: with self.lock:
for objectHash in self.hashes: for objectHash in self.hashes:
try: self.removeObjectFromCurrentThread(objectHash)
self.hashes[objectHash]['peers'].remove(current_thread().peer)
except ValueError:
pass
# current_thread().peer

View File

@ -27,7 +27,7 @@ import highlevelcrypto
#import helper_startup #import helper_startup
from helper_sql import * from helper_sql import *
from helper_threading import * from helper_threading import *
from inventory import Inventory, Missing from inventory import Inventory
import protocol import protocol
import state import state
@ -474,7 +474,6 @@ def _checkAndShareUndefinedObjectWithPeers(data):
return return
inventoryHash = calculateInventoryHash(data) inventoryHash = calculateInventoryHash(data)
Missing().delete(inventoryHash)
if inventoryHash in Inventory(): if inventoryHash in Inventory():
logger.debug('We have already received this undefined object. Ignoring.') logger.debug('We have already received this undefined object. Ignoring.')
return return
@ -498,7 +497,6 @@ def _checkAndShareMsgWithPeers(data):
return return
readPosition += streamNumberLength readPosition += streamNumberLength
inventoryHash = calculateInventoryHash(data) inventoryHash = calculateInventoryHash(data)
Missing().delete(inventoryHash)
if inventoryHash in Inventory(): if inventoryHash in Inventory():
logger.debug('We have already received this msg message. Ignoring.') logger.debug('We have already received this msg message. Ignoring.')
return return
@ -531,7 +529,6 @@ def _checkAndShareGetpubkeyWithPeers(data):
readPosition += streamNumberLength readPosition += streamNumberLength
inventoryHash = calculateInventoryHash(data) inventoryHash = calculateInventoryHash(data)
Missing().delete(inventoryHash)
if inventoryHash in Inventory(): if inventoryHash in Inventory():
logger.debug('We have already received this getpubkey request. Ignoring it.') logger.debug('We have already received this getpubkey request. Ignoring it.')
return return
@ -567,7 +564,6 @@ def _checkAndSharePubkeyWithPeers(data):
tag = '' tag = ''
inventoryHash = calculateInventoryHash(data) inventoryHash = calculateInventoryHash(data)
Missing().delete(inventoryHash)
if inventoryHash in Inventory(): if inventoryHash in Inventory():
logger.debug('We have already received this pubkey. Ignoring it.') logger.debug('We have already received this pubkey. Ignoring it.')
return return
@ -603,7 +599,6 @@ def _checkAndShareBroadcastWithPeers(data):
else: else:
tag = '' tag = ''
inventoryHash = calculateInventoryHash(data) inventoryHash = calculateInventoryHash(data)
Missing().delete(inventoryHash)
if inventoryHash in Inventory(): if inventoryHash in Inventory():
logger.debug('We have already received this broadcast object. Ignoring.') logger.debug('We have already received this broadcast object. Ignoring.')
return return