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

make peer search asynchronous

Browse source
This commit is contained in:
Victor Shyba 2018-10-04 15:39:12 -03:00 committed by Lex Berezhny
parent 234eaf8890
commit 24a11d3f21
1 changed files with 24 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
lbrynet/dht/peerfinder.py
View file

@ -27,8 +27,8 @@ class DHTPeerFinder(DummyPeerFinder):
self.dht_node = dht_node self.dht_node = dht_node
self.peer_manager = peer_manager self.peer_manager = peer_manager
self.peers = {} self.peers = {}
self._ongoing_searchs = {}
@defer.inlineCallbacks
def find_peers_for_blob(self, blob_hash, timeout=None, filter_self=True): def find_peers_for_blob(self, blob_hash, timeout=None, filter_self=True):
""" """
Find peers for blob in the DHT Find peers for blob in the DHT
@ -40,25 +40,27 @@ class DHTPeerFinder(DummyPeerFinder):
Returns: Returns:
list of peers for the blob list of peers for the blob
""" """
if blob_hash not in self.peers: self.peers.setdefault(blob_hash, {(self.dht_node.externalIP, self.dht_node.peerPort,)})
self.peers[blob_hash] = [(self.dht_node.externalIP, self.dht_node.peerPort)] if not blob_hash in self._ongoing_searchs:
self._ongoing_searchs[blob_hash] = self._execute_peer_search(blob_hash, timeout)
peers = set(self._filter_self(blob_hash) if filter_self else self.peers[blob_hash])
return defer.succeed([self.peer_manager.get_peer(*peer) for peer in peers])
@defer.inlineCallbacks
def _execute_peer_search(self, blob_hash, timeout):
try:
bin_hash = binascii.unhexlify(blob_hash) bin_hash = binascii.unhexlify(blob_hash)
finished_deferred = self.dht_node.iterativeFindValue(bin_hash, exclude=self.peers[blob_hash]) finished_deferred = self.dht_node.iterativeFindValue(bin_hash, exclude=self.peers[blob_hash])
timeout = timeout or conf.settings['peer_search_timeout'] timeout = timeout or conf.settings['peer_search_timeout']
if timeout: if timeout:
finished_deferred.addTimeout(timeout, self.dht_node.clock) finished_deferred.addTimeout(timeout, self.dht_node.clock)
try:
peer_list = yield finished_deferred peer_list = yield finished_deferred
self.peers[blob_hash].update(set((host, port) for _, host, port in peer_list))
except defer.TimeoutError: except defer.TimeoutError:
log.debug("DHT timed out while looking peers for blob %s after %s seconds", log.debug("DHT timed out while looking peers for blob %s after %s seconds", blob_hash, timeout)
blob_hash, timeout) finally:
peer_list = [] del self._ongoing_searchs[blob_hash]
peers = set(peer_list) def _filter_self(self, blob_hash):
results = [] my_host, my_port = self.dht_node.externalIP, self.dht_node.peerPort
for node_id, host, port in peers: return set((host, port) for host, port in self.peers[blob_hash] if (host, port) != (my_host, my_port))
if filter_self and (host, port) == (self.dht_node.externalIP, self.dht_node.peerPort):
continue
results.append(self.peer_manager.get_peer(host, port))
self.peers[blob_hash].append((host, port))
defer.returnValue(results)