tlnworker: make add_peer async - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
DIR commit ff0aa90ddf5c461105d0c798fa8f98f4db57d0be
DIR parent 8bb23ea2cd253d1a6f6059f301a8158ed22f8587
HTML Author: SomberNight <somber.night@protonmail.com>
Date: Tue, 16 Oct 2018 16:30:18 +0200
lnworker: make add_peer async
Diffstat:
M electrum/lnworker.py | 27 ++++++++++++++++-----------
1 file changed, 16 insertions(+), 11 deletions(-)
---
DIR diff --git a/electrum/lnworker.py b/electrum/lnworker.py
t@@ -90,7 +90,10 @@ class LNWorker(PrintError):
def _add_peers_from_config(self):
peer_list = self.config.get('lightning_peers', [])
for host, port, pubkey in peer_list:
- self.add_peer(host, int(port), bfh(pubkey))
+ asyncio.run_coroutine_threadsafe(
+ self.add_peer(host, int(port), bfh(pubkey)),
+ self.network.asyncio_loop)
+
def suggest_peer(self):
for node_id, peer in self.peers.items():
t@@ -105,20 +108,20 @@ class LNWorker(PrintError):
with self.lock:
return {x: y for (x, y) in self.channels.items() if y.node_id == node_id}
- def add_peer(self, host, port, node_id):
+ async def add_peer(self, host, port, node_id):
port = int(port)
peer_addr = LNPeerAddr(host, port, node_id)
if node_id in self.peers:
return
self._last_tried_peer[peer_addr] = time.time()
self.print_error("adding peer", peer_addr)
+ peer = Peer(self, peer_addr, request_initial_sync=self.config.get("request_initial_sync", True))
async def _init_peer():
reader, writer = await asyncio.open_connection(peer_addr.host, peer_addr.port)
transport = LNTransport(self.node_keypair.privkey, node_id, reader, writer)
peer.transport = transport
await self.network.main_taskgroup.spawn(peer.main_loop())
asyncio.ensure_future(_init_peer())
- peer = Peer(self, peer_addr, request_initial_sync=self.config.get("request_initial_sync", True))
self.peers[node_id] = peer
self.network.trigger_callback('ln_status')
return peer
t@@ -240,7 +243,9 @@ class LNWorker(PrintError):
socket.getaddrinfo(host, int(port))
except socket.gaierror:
raise ConnStringFormatError(_('Hostname does not resolve (getaddrinfo failed)'))
- peer = self.add_peer(host, port, node_id)
+ peer_future = asyncio.run_coroutine_threadsafe(self.add_peer(host, port, node_id),
+ self.network.asyncio_loop)
+ peer = peer_future.result(timeout)
coro = self._open_channel_coroutine(peer, local_amt_sat, push_amt_sat, password)
f = asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop)
chan = f.result(timeout)
t@@ -452,14 +457,14 @@ class LNWorker(PrintError):
self.print_error('got {} ln peers from dns seed'.format(len(peers)))
return peers
- def reestablish_peers_and_channels(self):
- def reestablish_peer_for_given_channel():
+ async def reestablish_peers_and_channels(self):
+ async def reestablish_peer_for_given_channel():
# try last good address first
peer = self.channel_db.get_last_good_address(chan.node_id)
if peer:
last_tried = self._last_tried_peer.get(peer, 0)
if last_tried + PEER_RETRY_INTERVAL_FOR_CHANNELS < now:
- self.add_peer(peer.host, peer.port, peer.pubkey)
+ await self.add_peer(peer.host, peer.port, peer.pubkey)
return
# try random address for node_id
node_info = self.channel_db.nodes.get(chan.node_id, None)
t@@ -470,7 +475,7 @@ class LNWorker(PrintError):
peer = LNPeerAddr(host, port, chan.node_id)
last_tried = self._last_tried_peer.get(peer, 0)
if last_tried + PEER_RETRY_INTERVAL_FOR_CHANNELS < now:
- self.add_peer(host, port, chan.node_id)
+ await self.add_peer(host, port, chan.node_id)
with self.lock:
channels = list(self.channels.values())
t@@ -480,7 +485,7 @@ class LNWorker(PrintError):
continue
peer = self.peers.get(chan.node_id, None)
if peer is None:
- reestablish_peer_for_given_channel()
+ await reestablish_peer_for_given_channel()
else:
coro = peer.reestablish_channel(chan)
asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop)
t@@ -491,11 +496,11 @@ class LNWorker(PrintError):
while True:
await asyncio.sleep(1)
now = time.time()
- self.reestablish_peers_and_channels()
+ await self.reestablish_peers_and_channels()
if len(self.peers) >= NUM_PEERS_TARGET:
continue
peers = self._get_next_peers_to_try()
for peer in peers:
last_tried = self._last_tried_peer.get(peer, 0)
if last_tried + PEER_RETRY_INTERVAL < now:
- self.add_peer(peer.host, peer.port, peer.pubkey)
+ await self.add_peer(peer.host, peer.port, peer.pubkey)