munet: simplify async child watcher setting for old pythons

Older pytest-asyncio did things with event_loops that we tried to work with, but
in the end this didn't work and pytest-asyncio was improved to not require it
anymore.

Now we just have to make sure we have a non-threaded child watcher for asyncio
when we are unsharing inline with a new pid namespace as we are not allowed to
create threads in the main process under these conditions.

Signed-off-by: Christian Hopps <chopps@labn.net>

Author: choppsv1

munet/__main__.py

@@ -17,7 +17,6 @@
 from . import cli
 from . import parser
 from .args import add_launch_args
-from .base import get_event_loop
 from .cleanup import cleanup_previous
 from .cleanup import is_running_in_rundir
 from .compat import PytestConfig
@@ -200,22 +199,16 @@ def main(*args):
         logger.critical("No nodes defined in config file")
         return 1
 
-    loop = None
     status = 4
     try:
         parser.validate_config(config, logger, args)
         if args.validate_only:
             return 0
-        # Executes the cmd for each node.
-        loop = get_event_loop()
-        status = loop.run_until_complete(async_main(args, config))
+        status = asyncio.get_event_loop().run_until_complete(async_main(args, config))
     except KeyboardInterrupt:
         logger.info("Exiting, received KeyboardInterrupt in main")
     except Exception as error:
         logger.info("Exiting, unexpected exception %s", error, exc_info=True)
-    finally:
-        if loop:
-            loop.close()
 
     return status
 

munet/base.py

@@ -268,34 +268,34 @@ def _get_exec_path(binary, cmdf, cache):
     return None
 
 
-def get_event_loop():
-    """Configure and return our non-thread using event loop.
+def ensure_non_threaded_event_watcher():
+    # Skip all this if >= python 3.12
+    if sys.version_info >= (3, 12):
+        return
 
-    This function configures a new child watcher to not use threads.
-    Threads cannot be used when we inline unshare a PID namespace.
-    """
     policy = asyncio.get_event_loop_policy()
-    loop = policy.get_event_loop()
-    if not hasattr(os, "pidfd_open"):
-        return loop
-
     owatcher = policy.get_child_watcher()
+    try:
+        want_class = asyncio.PidfdChildWatcher  # pylint: disable=no-member
+    except Exception:
+        want_class = asyncio.SafeChildWatcher  # pylint: disable=deprecated-class
+
     logging.debug(
-        "event_loop_fixture: global policy %s, current loop %s, current watcher %s",
+        "ensure_non_threaded_event_watcher: global policy %s, current watcher %s",
         policy,
-        loop,
         owatcher,
     )
 
+    # It's already non-threaded
+    if isinstance(owatcher, (want_class, asyncio.SafeChildWatcher)):
+        return
+
     policy.set_child_watcher(None)
     owatcher.close()
 
-    try:
-        watcher = asyncio.PidfdChildWatcher()  # pylint: disable=no-member
-    except Exception:
-        watcher = asyncio.SafeChildWatcher()                # pylint: disable=deprecated-class
+    watcher = want_class()
+    # Reget the loop (why, does it change with the watcher.close()?)
     loop = policy.get_event_loop()
-
     logging.debug(
         "event_loop_fixture: attaching new watcher %s to loop and setting in policy",
         watcher,
@@ -305,8 +305,6 @@ def get_event_loop():
     policy.set_event_loop(loop)
     assert asyncio.get_event_loop_policy().get_child_watcher() is watcher
 
-    return loop
-
 
 class Commander:  # pylint: disable=R0904
     """An object that can execute commands."""
@@ -2188,11 +2186,16 @@ def __init__(
 
             linux.unshare(uflags)
 
-            if not pid:
+            if not self.pid_ns:
                 p = None
                 self.pid = None
                 self.nsenter_fork = False
             else:
+                # If we are unsharing inline and creating a new pid namespace we need to
+                # ensure the default event loop watcher is a non-threaded version.
+                if not kvok or sys.version_info < (3, 12):
+                    ensure_non_threaded_event_watcher()
+
                 # Need to fork to create the PID namespace, but we need to continue
                 # running from the parent so that things like pytest work. We'll execute
                 # a mutini process to manage the child init 1 duties.
@@ -2331,7 +2334,7 @@ def __init__(
         # process using self.pid
         #
 
-        if pid:
+        if self.pid_ns:
             nsenter_fork = True
         elif unet and unet.nsenter_fork:
             # if unet created a pid namespace we need to enter it since we aren't
@@ -2376,7 +2379,7 @@ def __init__(
 
         # We need to remount the procfs for the new PID namespace, since we aren't using
         # unshare(1) which does that for us.
-        if pid and unshare_inline:
+        if self.pid_ns and unshare_inline:
             assert mount
             self.cmd_raises_nsonly("mount -t proc proc /proc")
 
@@ -2443,7 +2446,7 @@ def __init__(
                     self.bind_mount(s[0], s[1])
 
         # this will fail if running inside the namespace with PID
-        if pid:
+        if self.pid_ns:
             o = self.cmd_nostatus_nsonly("ls -l /proc/1/ns")
         else:
             o = self.cmd_nostatus_nsonly("ls -l /proc/self/ns")

munet/mutest/__main__.py

@@ -24,7 +24,6 @@
 from munet import parser
 from munet.args import add_testing_args
 from munet.base import Bridge
-from munet.base import get_event_loop
 from munet.cli import async_cli
 from munet.compat import PytestConfig
 from munet.mutest import userapi as uapi
@@ -473,8 +472,7 @@ def main():
     loop = None
     status = 4
     try:
-        loop = get_event_loop()
-        status = loop.run_until_complete(async_main(args))
+        status = asyncio.get_event_loop().run_until_complete(async_main(args))
     except KeyboardInterrupt:
         logging.info("Exiting (main), received KeyboardInterrupt in main")
     except Exception as error:

munet/testing/fixtures.py

@@ -23,7 +23,6 @@
 
 from ..base import BaseMunet
 from ..base import Bridge
-from ..base import get_event_loop
 from ..cleanup import cleanup_current
 from ..native import L3NodeMixin
 from ..parser import async_build_topology
@@ -164,17 +163,6 @@ def module_autouse(request):
             raise Exception("Base Munet was not cleaned up/deleted")
 
 
-@pytest.fixture(scope="session")
-def event_loop():
-    """Create an instance of the default event loop for the session."""
-    loop = get_event_loop()
-    try:
-        logging.info("event_loop_fixture: yielding with new event loop watcher")
-        yield loop
-    finally:
-        loop.close()
-
-
 @pytest.fixture(scope="module")
 def rundir_module():
     root_path = os.environ.get("MUNET_RUNDIR", "/tmp/unet-test")

tests/control/test_cmds.py

@@ -13,10 +13,7 @@
 import pytest
 
 
-# All tests are coroutines
-pytestmark = pytest.mark.asyncio
-
-# How does this double assignment work, what's going on?
+# Run tests with unshare_inline and not.
 pytestmark = pytest.mark.parametrize("unet", [True, False], indirect=["unet"])