Skip to content

Commit b6a14a2

Browse files
committed
asyncio: stop using get_event_loop(). introduce ~singleton loop.
asyncio.get_event_loop() became deprecated in python3.10. (see python/cpython#83710) ``` .../electrum/electrum/daemon.py:470: DeprecationWarning: There is no current event loop self.asyncio_loop = asyncio.get_event_loop() .../electrum/electrum/network.py:276: DeprecationWarning: There is no current event loop self.asyncio_loop = asyncio.get_event_loop() ``` Also, according to that thread, "set_event_loop() [... is] not deprecated by oversight". So, we stop using get_event_loop() and set_event_loop() in our own code. Note that libraries we use (such as the stdlib for python <3.10), might call get_event_loop, which then relies on us having called set_event_loop e.g. for the GUI thread. To work around this, a custom event loop policy providing a get_event_loop implementation is used. Previously, we have been using a single asyncio event loop, created with util.create_and_start_event_loop, and code in many places got a reference to this loop using asyncio.get_event_loop(). Now, we still use a single asyncio event loop, but it is now stored as a global in util._asyncio_event_loop (access with util.get_asyncio_loop()). I believe these changes also fix spesmilo#5376
1 parent 334da24 commit b6a14a2

13 files changed

+111
-58
lines changed

electrum/commands.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,7 @@ def _run(self, method, args, password_getter=None, **kwargs):
187187
kwargs.pop('wallet')
188188

189189
coro = f(*args, **kwargs)
190-
fut = asyncio.run_coroutine_threadsafe(coro, asyncio.get_event_loop())
190+
fut = asyncio.run_coroutine_threadsafe(coro, util.get_asyncio_loop())
191191
result = fut.result()
192192

193193
if self._callback:

electrum/daemon.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,7 @@ def request(config: SimpleConfig, endpoint, args=(), timeout=60):
124124
rpc_user, rpc_password = get_rpc_credentials(config)
125125
server_url = 'http://%s:%d' % (host, port)
126126
auth = aiohttp.BasicAuth(login=rpc_user, password=rpc_password)
127-
loop = asyncio.get_event_loop()
127+
loop = util.get_asyncio_loop()
128128
async def request_coroutine():
129129
if socktype == 'unix':
130130
connector = aiohttp.UnixConnector(path=path)
@@ -467,7 +467,7 @@ def __init__(self, config: SimpleConfig, fd=None, *, listen_jsonrpc=True):
467467
if 'wallet_path' in config.cmdline_options:
468468
self.logger.warning("Ignoring parameter 'wallet_path' for daemon. "
469469
"Use the load_wallet command instead.")
470-
self.asyncio_loop = asyncio.get_event_loop()
470+
self.asyncio_loop = util.get_asyncio_loop()
471471
self.network = None
472472
if not config.get('offline'):
473473
self.network = Network(config, daemon=self)

electrum/exchange_rate.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,7 @@ def get_historical_rates(self, ccy: str, cache_dir: str) -> None:
148148
if h is None:
149149
h = self.read_historical_rates(ccy, cache_dir)
150150
if h is None or h['timestamp'] < time.time() - 24*3600:
151-
asyncio.get_event_loop().create_task(self.get_historical_rates_safe(ccy, cache_dir))
151+
util.get_asyncio_loop().create_task(self.get_historical_rates_safe(ccy, cache_dir))
152152

153153
def history_ccys(self) -> Sequence[str]:
154154
return []
@@ -471,7 +471,7 @@ async def query_all_exchanges_for_their_ccys_over_network():
471471
for name, klass in exchanges.items():
472472
exchange = klass(None, None)
473473
await group.spawn(get_currencies_safe(name, exchange))
474-
loop = asyncio.get_event_loop()
474+
loop = util.get_asyncio_loop()
475475
try:
476476
loop.run_until_complete(query_all_exchanges_for_their_ccys_over_network())
477477
except Exception as e:

electrum/gui/kivy/main_window.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -394,7 +394,7 @@ def __init__(self, **kwargs):
394394
self.is_exit = False
395395
self.wallet = None # type: Optional[Abstract_Wallet]
396396
self.pause_time = 0
397-
self.asyncio_loop = asyncio.get_event_loop()
397+
self.asyncio_loop = util.get_asyncio_loop()
398398
self.password = None
399399
self._use_single_password = False
400400
self.resume_dialog = None

electrum/lnworker.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -496,7 +496,7 @@ async def add_peer(self, connect_str: str) -> Peer:
496496
# Try DNS-resolving the host (if needed). This is simply so that
497497
# the caller gets a nice exception if it cannot be resolved.
498498
try:
499-
await asyncio.get_event_loop().getaddrinfo(host, port)
499+
await asyncio.get_running_loop().getaddrinfo(host, port)
500500
except socket.gaierror:
501501
raise ConnStringFormatError(_('Hostname does not resolve (getaddrinfo failed)'))
502502
# add peer

electrum/network.py

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -273,7 +273,7 @@ def __init__(self, config: SimpleConfig, *, daemon: 'Daemon' = None):
273273
init_retry_delay_urgent=1,
274274
)
275275

276-
self.asyncio_loop = asyncio.get_event_loop()
276+
self.asyncio_loop = util.get_asyncio_loop()
277277
assert self.asyncio_loop.is_running(), "event loop not running"
278278

279279
assert isinstance(config, SimpleConfig), f"config should be a SimpleConfig instead of {type(config)}"
@@ -381,9 +381,11 @@ async def stop_gossip(self, *, full_shutdown: bool = False):
381381
self.channel_db = None
382382
self.path_finder = None
383383

384-
def run_from_another_thread(self, coro, *, timeout=None):
385-
assert util.get_running_loop() != self.asyncio_loop, 'must not be called from network thread'
386-
fut = asyncio.run_coroutine_threadsafe(coro, self.asyncio_loop)
384+
@classmethod
385+
def run_from_another_thread(cls, coro, *, timeout=None):
386+
loop = util.get_asyncio_loop()
387+
assert util.get_running_loop() != loop, 'must not be called from asyncio thread'
388+
fut = asyncio.run_coroutine_threadsafe(coro, loop)
387389
return fut.result(timeout)
388390

389391
@staticmethod
@@ -1321,7 +1323,7 @@ def send_http_on_proxy(cls, method, url, **kwargs):
13211323
assert util.get_running_loop() != network.asyncio_loop
13221324
loop = network.asyncio_loop
13231325
else:
1324-
loop = asyncio.get_event_loop()
1326+
loop = util.get_asyncio_loop()
13251327
coro = asyncio.run_coroutine_threadsafe(cls._send_http_on_proxy(method, url, **kwargs), loop)
13261328
# note: _send_http_on_proxy has its own timeout, so no timeout here:
13271329
return coro.result()

electrum/sql_db.py

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ class SqlDB(Logger):
2626
def __init__(self, asyncio_loop: asyncio.BaseEventLoop, path, commit_interval=None):
2727
Logger.__init__(self)
2828
self.asyncio_loop = asyncio_loop
29-
asyncio.set_event_loop(asyncio_loop)
3029
self.stopping = False
3130
self.stopped_event = asyncio.Event()
3231
self.path = path

electrum/tests/test_lnpeer.py

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import electrum
1616
import electrum.trampoline
1717
from electrum import bitcoin
18+
from electrum import util
1819
from electrum import constants
1920
from electrum.network import Network
2021
from electrum.ecc import ECPrivkey
@@ -62,7 +63,7 @@ def __init__(self, tx_queue):
6263
user_config = {}
6364
user_dir = tempfile.mkdtemp(prefix="electrum-lnpeer-test-")
6465
self.config = simple_config.SimpleConfig(user_config, read_user_dir_function=lambda: user_dir)
65-
self.asyncio_loop = asyncio.get_event_loop()
66+
self.asyncio_loop = util.get_asyncio_loop()
6667
self.channel_db = ChannelDB(self)
6768
self.channel_db.data_loaded.set()
6869
self.path_finder = LNPathFinder(self.channel_db)
@@ -1361,4 +1362,4 @@ async def f():
13611362

13621363

13631364
def run(coro):
1364-
return asyncio.run_coroutine_threadsafe(coro, loop=asyncio.get_event_loop()).result()
1365+
return asyncio.run_coroutine_threadsafe(coro, loop=util.get_asyncio_loop()).result()

electrum/tests/test_lnrouter.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import shutil
55
import asyncio
66

7+
from electrum import util
78
from electrum.util import bh2u, bfh, create_and_start_event_loop
89
from electrum.lnutil import ShortChannelID
910
from electrum.lnonion import (OnionHopsDataSingle, new_onion_packet,
@@ -64,7 +65,7 @@ def prepare_graph(self):
6465
"""
6566
class fake_network:
6667
config = self.config
67-
asyncio_loop = asyncio.get_event_loop()
68+
asyncio_loop = util.get_asyncio_loop()
6869
trigger_callback = lambda *args: None
6970
register_callback = lambda *args: None
7071
interface = None

electrum/tests/test_lntransport.py

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import asyncio
22

3+
from electrum import util
34
from electrum.ecc import ECPrivkey
45
from electrum.lnutil import LNPeerAddr
56
from electrum.lntransport import LNResponderTransport, LNTransport
@@ -11,6 +12,15 @@
1112

1213
class TestLNTransport(ElectrumTestCase):
1314

15+
def setUp(self):
16+
super().setUp()
17+
self.asyncio_loop, self._stop_loop, self._loop_thread = util.create_and_start_event_loop()
18+
19+
def tearDown(self):
20+
self.asyncio_loop.call_soon_threadsafe(self._stop_loop.set_result, 1)
21+
self._loop_thread.join(timeout=1)
22+
super().tearDown()
23+
1424
@needs_test_with_all_chacha20_implementations
1525
def test_responder(self):
1626
# local static
@@ -38,11 +48,11 @@ async def read(self, num_bytes):
3848
assert num_bytes == 66
3949
return bytes.fromhex('00b9e3a702e93e3a9948c2ed6e5fd7590a6e1c3a0344cfc9d5b57357049aa22355361aa02e55a8fc28fef5bd6d71ad0c38228dc68b1c466263b47fdf31e560e139ba')
4050
transport = LNResponderTransport(ls_priv, Reader(), Writer())
41-
asyncio.get_event_loop().run_until_complete(transport.handshake(epriv=e_priv))
51+
asyncio.run_coroutine_threadsafe(
52+
transport.handshake(epriv=e_priv), self.asyncio_loop).result()
4253

4354
@needs_test_with_all_chacha20_implementations
4455
def test_loop(self):
45-
loop = asyncio.get_event_loop()
4656
responder_shaked = asyncio.Event()
4757
server_shaked = asyncio.Event()
4858
responder_key = ECPrivkey.generate_random_key()
@@ -96,4 +106,4 @@ async def f():
96106
server.close()
97107
await server.wait_closed()
98108

99-
loop.run_until_complete(f())
109+
asyncio.run_coroutine_threadsafe(f(), self.asyncio_loop).result()

0 commit comments

Comments
 (0)