LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases 282 Packages Wiki Activity

fix requesting same blob over and over if only peer(s) say they don't have it

Browse source
This commit is contained in:
Jack Robison 2019-11-26 14:53:14 -05:00
parent 87ec2c7378
commit 669f3394c7
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
1 changed files with 19 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
lbry/lbry/blob_exchange/downloader.py
View file

@ -27,7 +27,7 @@ class BlobDownloader:
self.ignored: typing.Dict['KademliaPeer', int] = {} self.ignored: typing.Dict['KademliaPeer', int] = {}
self.scores: typing.Dict['KademliaPeer', int] = {} self.scores: typing.Dict['KademliaPeer', int] = {}
self.failures: typing.Dict['KademliaPeer', int] = {} self.failures: typing.Dict['KademliaPeer', int] = {}
self.connection_failures: typing.List['KademliaPeer'] = [] self.connection_failures: typing.Set['KademliaPeer'] = set()
self.connections: typing.Dict['KademliaPeer', 'BlobExchangeClientProtocol'] = {} self.connections: typing.Dict['KademliaPeer', 'BlobExchangeClientProtocol'] = {}
self.is_running = asyncio.Event(loop=self.loop) self.is_running = asyncio.Event(loop=self.loop)
@ -48,7 +48,7 @@ class BlobDownloader:
connection_id=connection_id, connection_manager=self.blob_manager.connection_manager connection_id=connection_id, connection_manager=self.blob_manager.connection_manager
) )
if not bytes_received and not protocol and peer not in self.connection_failures: if not bytes_received and not protocol and peer not in self.connection_failures:
self.connection_failures.append(peer) self.connection_failures.add(peer)
if not protocol and peer not in self.ignored: if not protocol and peer not in self.ignored:
self.ignored[peer] = self.loop.time() self.ignored[peer] = self.loop.time()
log.debug("drop peer %s:%i", peer.address, peer.tcp_port) log.debug("drop peer %s:%i", peer.address, peer.tcp_port)
@ -87,32 +87,42 @@ class BlobDownloader:
if blob.get_is_verified(): if blob.get_is_verified():
return blob return blob
self.is_running.set() self.is_running.set()
tried_for_this_blob: typing.Set['KademliaPeer'] = set()
try: try:
while not blob.get_is_verified() and self.is_running.is_set(): while not blob.get_is_verified() and self.is_running.is_set():
batch: typing.Set['KademliaPeer'] = set() batch: typing.Set['KademliaPeer'] = set(self.connections.keys())
while not self.peer_queue.empty(): while not self.peer_queue.empty():
batch.update(self.peer_queue.get_nowait()) batch.update(self.peer_queue.get_nowait())
if batch:
self.peer_queue.put_nowait(list(batch))
log.debug( log.debug(
"running, %d peers, %d ignored, %d active, %s connections", "%s running, %d peers, %d ignored, %d active, %s connections", blob_hash[:6],
len(batch), len(self.ignored), len(self.active_connections), len(self.connections) len(batch), len(self.ignored), len(self.active_connections), len(self.connections)
) )
re_add: typing.Set['KademliaPeer'] = set()
for peer in sorted(batch, key=lambda peer: self.scores.get(peer, 0), reverse=True): for peer in sorted(batch, key=lambda peer: self.scores.get(peer, 0), reverse=True):
if peer in self.ignored or peer in self.active_connections: if peer in self.ignored:
continue
if peer in tried_for_this_blob:
continue
if peer in self.active_connections:
if peer not in re_add:
re_add.add(peer)
continue continue
if not self.should_race_continue(blob): if not self.should_race_continue(blob):
break break
log.debug("request %s from %s:%i", blob_hash[:8], peer.address, peer.tcp_port) log.debug("request %s from %s:%i", blob_hash[:8], peer.address, peer.tcp_port)
just_probe = len(self.connections) == 0 t = self.loop.create_task(self.request_blob_from_peer(blob, peer, connection_id))
t = self.loop.create_task(self.request_blob_from_peer(blob, peer, connection_id, just_probe))
self.active_connections[peer] = t self.active_connections[peer] = t
tried_for_this_blob.add(peer)
if not re_add:
self.peer_queue.put_nowait(list(batch))
await self.new_peer_or_finished() await self.new_peer_or_finished()
self.cleanup_active() self.cleanup_active()
log.debug("downloaded %s", blob_hash[:8]) log.debug("downloaded %s", blob_hash[:8])
return blob return blob
finally: finally:
blob.close() blob.close()
if self.loop.is_running():
self.loop.call_soon(self.cleanup_active)
def close(self): def close(self):
self.connection_failures.clear() self.connection_failures.clear()