util/helper: introduce ensure_event_loop()

Calling asyncio.get_event_loop() with no current event loop is deprecated
since Python 3.10 and will be an error in some future Python release
[1]. Using it causes errors in IPython when using a RemotePlace.

Instead of using asyncio.get_event_loop(), add a new helper function
ensure_event_loop(): Try to retrieve the OS thread's event loop from a
ContextVar. If the ContextVar is not set yet, try to get the current loop
from asyncio.get_running_loop() and store it in the ContextVar. If
there is no loop, create a new one, set it as the current event loop for
the current thread and store in the ContextVar. Finally return the found
or newly created loop.

This behavior mimics what asyncio.get_event_loop() did. It is also used
by jupyter_core [2].

[1] https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.get_event_loop
[2] jupyter/jupyter_core#387

Signed-off-by: Bastian Krause <[email protected]>

labgrid/util/helper.py

@@ -1,3 +1,5 @@
+import asyncio
+from contextvars import ContextVar
 import fcntl
 import os
 import logging
@@ -30,6 +32,24 @@ def get_user():
     import pwd
     return pwd.getpwuid(os.getuid())[0]
 
+
+_loop: ContextVar[asyncio.AbstractEventLoop | None] = ContextVar("_loop", default=None)
+
+
+def ensure_event_loop():
+    """Get the event loop for this thread, or create a new event loop."""
+    loop = _loop.get()
+    if loop is not None and not loop.is_closed():
+        return loop
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+    _loop.set(loop)
+    return loop
+
+
 def set_nonblocking(fd):
     flags = fcntl.fcntl(fd, fcntl.F_GETFL)
     fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)