RanchiMall/electrumx
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Avoid callback in peer manager

Browse Source 
Spawn separate tasks for each peer we test
This commit is contained in:
Neil Booth 2018-07-24 14:16:48 +08:00
parent 12c024f3ac
commit ba607544b9
1 changed files with 34 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
electrumx/server/peers.py
View File

@ -347,7 +347,7 @@ class PeerManager(object):
else: else:
self.logger.info('no proxy detected') self.logger.info('no proxy detected')
async def _peer_discovery_loop(self): async def _discover_peers(self):
'''Main loop performing peer maintenance. This includes '''Main loop performing peer maintenance. This includes
1) Forgetting unreachable peers. 1) Forgetting unreachable peers.
@ -386,57 +386,46 @@ class PeerManager(object):
# Retry a failed connection if enough time has passed # Retry a failed connection if enough time has passed
return peer.last_try < now - WAKEUP_SECS * 2 ** peer.try_count return peer.last_try < now - WAKEUP_SECS * 2 ** peer.try_count
peers = [peer for peer in self.peers if should_retry(peer)] for peer in self.peers:
if should_retry(peer):
self.tasks.create_task(self._retry_peer(peer))
for peer in peers: async def _retry_peer(self, peer):
peer.try_count += 1 peer.try_count += 1
pairs = peer.connection_port_pairs()
if peer.bad or not pairs:
self._maybe_forget_peer(peer)
else:
self._retry_peer(peer, pairs)
def _retry_peer(self, peer, port_pairs): for kind, port in peer.connection_port_pairs():
peer.last_try = time.time() peer.last_try = time.time()
kwargs = {'loop': self.loop} kwargs = {}
if kind == 'SSL':
kwargs['ssl'] = ssl.SSLContext(ssl.PROTOCOL_TLS)
kind, port = port_pairs[0] host = self.env.cs_host(for_rpc=False)
if kind == 'SSL': if isinstance(host, list):
kwargs['ssl'] = ssl.SSLContext(ssl.PROTOCOL_TLS) host = host[0]
host = self.env.cs_host(for_rpc=False) if self.env.force_proxy or peer.is_tor:
if isinstance(host, list): if not self.proxy:
host = host[0] return
kwargs['proxy'] = self.proxy
kwargs['resolve'] = not peer.is_tor
elif host:
# Use our listening Host/IP for outgoing non-proxy
# connections so our peers see the correct source.
kwargs['local_addr'] = (host, None)
if self.env.force_proxy or peer.is_tor: session = PeerSession(peer, self, kind, peer.host, port, **kwargs)
if not self.proxy: try:
await session.create_connection()
return return
kwargs['proxy'] = self.proxy except Exception as e:
kwargs['resolve'] = not peer.is_tor elapsed = time.time() - peer.last_try
elif host: self.logger.info(f'failed connecting to {peer} at {kind} port '
# Use our listening Host/IP for outgoing non-proxy f'{port} in {elapsed:.1f}s: {e}')
# connections so our peers see the correct source. # Try the next port pair
kwargs['local_addr'] = (host, None) continue
session = PeerSession(peer, self, kind, peer.host, port, **kwargs) self._maybe_forget_peer(peer)
callback = partial(self._on_connected, peer, port_pairs)
self.tasks.create_task(session.create_connection(), callback)
def _on_connected(self, peer, port_pairs, task):
'''Called when a connection attempt succeeds or fails.
If failed, close the session, log it and try remaining port pairs.
'''
if not task.cancelled() and task.exception():
kind, port = port_pairs.pop(0)
elapsed = time.time() - peer.last_try
self.logger.info(f'failed connecting to {peer} at {kind} port '
f'{port} in {elapsed:.1f}s: {task.exception()}')
if port_pairs:
self._retry_peer(peer, port_pairs)
else:
self._maybe_forget_peer(peer)
def _set_verification_status(self, peer, kind, good): def _set_verification_status(self, peer, kind, good):
'''Called when a verification succeeded or failed.''' '''Called when a verification succeeded or failed.'''
@ -478,8 +467,6 @@ class PeerManager(object):
self.logger.info('forgetting {} peer: {}'.format(desc, peer)) self.logger.info('forgetting {} peer: {}'.format(desc, peer))
self.peers.discard(peer) self.peers.discard(peer)
return forget
# #
# External interface # External interface
# #
@ -487,7 +474,7 @@ class PeerManager(object):
if self.env.peer_discovery == self.env.PD_ON: if self.env.peer_discovery == self.env.PD_ON:
self.logger.info(f'beginning peer discovery. Force use of ' self.logger.info(f'beginning peer discovery. Force use of '
f'proxy: {self.env.force_proxy}') f'proxy: {self.env.force_proxy}')
self.tasks.create_task(self._peer_discovery_loop()) self.tasks.create_task(self._discover_peers())
else: else:
self.logger.info('peer discovery is disabled') self.logger.info('peer discovery is disabled')