remote/client: rework event loop handling in start_session()

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 in labgrid.remote.client.start_session() causes errors in
IPython when using a RemotePlace:

  In [1]: from labgrid.resource.remote import RemotePlace
     ...: from labgrid import Target
     ...:
     ...: target = Target("example")
     ...: RemotePlace(target, name="example-place")
  [...]
  RuntimeError: There is no current event loop in thread 'MainThread'.

For labgrid.remote.client.start_session() there is no reliable way of
retrieving the thread's event loop without being called from an async
context (which we cannot assume here). Instead of using
asyncio.get_event_loop(), use a new helper function ensure_event_loop()
that returns the first available loop instance from:

- externally provided event loop
- stashed event loop
- OS thread's running event loop (when called from async code)
- new event loop

The returned loop is stashed for future calls. See also [2] for a
similar approach.

start_session() now accepts a new optional argument "loop" for providing
an external event loop.

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

Signed-off-by: Bastian Krause <bst@pengutronix.de>

labgrid/remote/client.py

@@ -4,6 +4,7 @@
 import argparse
 import asyncio
 import contextlib
+from contextvars import ContextVar
 import enum
 import os
 import pathlib
@@ -1532,8 +1533,44 @@ def print_version(self):
  print(labgrid_version())
 
 
-def start_session(url, extra, debug=False):
- loop = asyncio.get_event_loop()
+_loop: ContextVar["asyncio.AbstractEventLoop | None"] = ContextVar("_loop", default=None)
+
+def ensure_event_loop(external_loop=None):
+ """Get the event loop for this thread, or create a new event loop."""
+ # get stashed loop
+ loop = _loop.get()
+
+ # ignore closed stashed loop
+ if loop and loop.is_closed():
+ loop = None
+
+ if external_loop:
+ # if a loop is stashed, expect it to be the same as the external one
+ if loop:
+ assert loop is external_loop
+ _loop.set(external_loop)
+ return external_loop
+
+ # return stashed loop
+ if loop:
+ return loop
+
+ try:
+ # if called from async code, try to get current's thread loop
+ loop = asyncio.get_running_loop()
+ except RuntimeError:
+ # no previous, external or running loop found, create a new one
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+
+ # stash it
+ _loop.set(loop)
+ return loop
+
+
+def start_session(url, extra, debug=False, loop=None):
+ loop = ensure_event_loop(loop)
+
  if debug:
  loop.set_debug(True)
  session = None
@@ -2044,7 +2081,9 @@ def main():
  coordinator_url = os.environ.get("LG_COORDINATOR", "127.0.0.1:20408")
 
  logging.debug('Starting session with "%s"', coordinator_url)
- session = start_session(coordinator_url, extra, args.debug)
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ session = start_session(coordinator_url, extra, debug=args.debug, loop=loop)
  logging.debug("Started session")
 
  try: