munet: make Commander.spawn() async

spawn(), which conducts a send/expect process is
blocking due to the abscence of async_=True to
pexpect.expect() within the method's main
send/expect loop. This is undesirable due the
presence of other ongoing initialization
coroutines, since it will block their progress
for the entire time needed to complete either
the single nodes send/expect loop (or wait for
it to timeout).

This is particularly problematic in cases where
multiple VMs requiring custom console expect/send
prompts are required (e.g. routers). Since each
QEMU VM sets up both console logging and completes
an expect/send loop without yielding control to
the coroutine of the other QEMU VMs, only one VM
can ever have logging configured (and thus advance
in an expect/send loop) at any given time. Not
only is this inefficient given that all QEMU VMs
are running and waiting for input, but this can
be fatal if console logging is not set up on each
VM in time. If some mandatory console output is
missed, then there is no way to recover and the
setup of the QEMU VM will time out.

However, setting async_=True in pexpect.expect()
leads to errors when mixed with PopenSpawn.
To bypass this issue, we do not use pexpect.expect()
with async_=True, but repeatedly call
pexpect.expect() with a timeout set at 0.1 (Until
a timeout that we track expires).

The end result of this change is that both logging
is set up on all nodes immediately in QEMU nodes
and that independent progress can be made in each
VM's expect/send loop.

Signed-off-by: Liam Brady <[email protected]>

Author: liambrady

munet/base.py

@@ -603,7 +603,7 @@ def _spawn(self, cmd, skip_pre_cmd=False, use_pty=False, echo=False, **kwargs):
             p = pexpect.spawn(actual_cmd[0], actual_cmd[1:], echo=echo, **defaults)
         return p, actual_cmd
 
-    def spawn(
+    async def spawn(
         self,
         cmd,
         spawned_re,
@@ -616,7 +616,7 @@ def spawn(
         trace=None,
         **kwargs,
     ):
-        """Create a spawned send/expect process.
+        """Create an async spawned send/expect process.
 
         Args:
             cmd: list of args to exec/popen with, or an already open socket
@@ -677,12 +677,18 @@ def spawn(
 
         # Now send a CRLF to cause the prompt (or whatever else) to re-issue
         p.send("\n")
-        try:
-            patterns = [spawned_re, *expects]
+        patterns = [spawned_re, *expects]
 
-            self.logger.debug("%s: expecting: %s", self, patterns)
+        self.logger.debug("%s: expecting: %s", self, patterns)
 
-            while index := p.expect(patterns):
+        timeout = kwargs.get("timeout", 120)
+        timeout_ts = datetime.datetime.now() + datetime.timedelta(seconds=timeout)
+        while timeout_ts > datetime.datetime.now():
+            try:
+                await asyncio.sleep(0)  # Avoid blocking other coroutines
+                index = p.expect(patterns, timeout=0.1)
+                if index == 0:
+                    break
                 if trace:
                     assert p.match is not None
                     self.logger.debug(
@@ -697,35 +703,38 @@ def spawn(
                     p.send(sends[index - 1])
 
                 self.logger.debug("%s: expecting again: %s", self, patterns)
-            self.logger.debug(
-                "%s: got spawned_re: '%s' matching '%s'",
-                self,
-                p.match.group(0),
-                spawned_re,
-            )
-            return p
-        except pexpect.TIMEOUT:
+            except pexpect.TIMEOUT:
+                continue
+            except pexpect.EOF as eoferr:
+                if p.isalive():
+                    raise
+                rc = p.status
+                before = indent(p.before)
+                error = CalledProcessError(rc, ac, output=before)
+                self.logger.error(
+                    "%s: EOF looking for spawned_re '%s' before EOF:\n%s",
+                    self,
+                    spawned_re,
+                    before,
+                )
+                p.close()
+                raise error from eoferr
+
+        if p.match is None:
             self.logger.error(
                 "%s: TIMEOUT looking for spawned_re '%s' expect buffer so far:\n%s",
                 self,
                 spawned_re,
                 indent(p.buffer),
             )
-            raise
-        except pexpect.EOF as eoferr:
-            if p.isalive():
-                raise
-            rc = p.status
-            before = indent(p.before)
-            error = CalledProcessError(rc, ac, output=before)
-            self.logger.error(
-                "%s: EOF looking for spawned_re '%s' before EOF:\n%s",
-                self,
-                spawned_re,
-                before,
-            )
-            p.close()
-            raise error from eoferr
+            raise pexpect.TIMEOUT("spawn() timed out within Munet")
+        self.logger.debug(
+            "%s: got spawned_re: '%s' matching '%s'",
+            self,
+            p.match.group(0),
+            spawned_re,
+        )
+        return p
 
     async def shell_spawn(
         self,
@@ -761,7 +770,7 @@ async def shell_spawn(
         combined_prompt = r"({}|{})".format(re.escape(PEXPECT_PROMPT), prompt)
 
         assert not is_file_like(cmd) or not use_pty
-        p = self.spawn(
+        p = await self.spawn(
             cmd,
             combined_prompt,
             expects=expects,

munet/native.py

@@ -911,7 +911,7 @@ async def monitor(
         logfile_read = open(lfname, "a+", encoding="utf-8")
         logfile_read.write("-- start read logging for: '{}' --\n".format(sock))
 
-        p = self.spawn(sock, prompt, logfile=logfile, logfile_read=logfile_read)
+        p = await self.spawn(sock, prompt, logfile=logfile, logfile_read=logfile_read)
         from .base import ShellWrapper  # pylint: disable=C0415
 
         p.send("\n")