citizenaspirant/PyBitmessage
1
0
Fork 0
Code Issues 1 Pull Requests Releases Wiki Activity

Expire objects that we're unable to receive

Browse Source 
- sometimes a node would send an "inv" about an object but then not
  provide it when requested. This could be that it expired in the
  meantime or it was an attack or a bug. This patch will forget that the
  object exists if was requested too many times and not received.
This commit is contained in:
Peter Šurda 2017-01-16 17:08:47 +01:00
parent ca6bc9981c
commit 9f89df6d1c
Signed by untrusted user: PeterSurda
GPG Key ID: 0C5F50C0B5F37D87
1 changed files with 22 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/inventory.py
View File

@ -91,7 +91,10 @@ class Missing(object):
self.lock = RLock() self.lock = RLock()
self.hashes = {} self.hashes = {}
self.stopped = False self.stopped = False
# don't request the same object more frequently than this
self.frequency = 60 self.frequency = 60
# after requesting and not receiving an object more than this times, consider it expired
self.maxRequestCount = 6
self.pending = {} self.pending = {}
def add(self, objectHash): def add(self, objectHash):
@ -99,7 +102,7 @@ class Missing(object):
return return
with self.lock: with self.lock:
if objectHash not in self.hashes: if objectHash not in self.hashes:
self.hashes[objectHash] = {'peers':[], 'requested':0} self.hashes[objectHash] = {'peers':[], 'requested':0, 'requestedCount':0}
self.hashes[objectHash]['peers'].append(current_thread().peer) self.hashes[objectHash]['peers'].append(current_thread().peer)
def addPending(self, objectHash=None): def addPending(self, objectHash=None):
@ -144,16 +147,31 @@ class Missing(object):
self.pending[current_thread().peer]['received'] >= time.time() - self.frequency) and \ self.pending[current_thread().peer]['received'] >= time.time() - self.frequency) and \
len(self.pending[current_thread().peer]['objects']) >= count: len(self.pending[current_thread().peer]['objects']) >= count:
break break
# requested too long ago or not at all # requested too long ago or not at all from any thread
if self.hashes[objectHash]['requested'] < time.time() - self.frequency: if self.hashes[objectHash]['requested'] < time.time() - self.frequency:
# already requested from this thread but haven't received yet # ready requested from this thread but haven't received yet
if objectHash in self.pending[current_thread().peer]['objects']: 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: # if still sending or receiving, request next
if self.pending[current_thread().peer]['received'] >= time.time() - self.frequency or \
self.pending[current_thread().peer]['requested'] >= time.time() - self.frequency:
continue continue
# haven't requested or received anything recently, re-request (i.e. continue)
# the current node doesn't have the object
elif current_thread().peer not in self.hashes[objectHash]['peers']: elif current_thread().peer not in self.hashes[objectHash]['peers']:
continue continue
# already requested too many times, remove all signs of this object
if self.hashes[objectHash]['requestedCount'] >= self.maxRequestCount:
with self.lock:
del self.hashes[objectHash]
for thread in self.pending.keys():
if objectHash in self.pending[thread]['objects']:
self.pending[thread]['objects'].remove(objectHash)
continue
# all ok, request
objectHashes.append(objectHash) objectHashes.append(objectHash)
self.hashes[objectHash]['requested'] = time.time() self.hashes[objectHash]['requested'] = time.time()
with self.lock:
self.hashes[objectHash]['requestedCount'] += 1
self.pending[current_thread().peer]['requested'] = time.time() self.pending[current_thread().peer]['requested'] = time.time()
self.addPending(objectHash) self.addPending(objectHash)
except (RuntimeError, KeyError, ValueError): except (RuntimeError, KeyError, ValueError):