fix orchestration framework to run on Python3.11

Starting with Python3.11, asyncio.wait() doesn't accept coroutines anymore. Instead, we have to wrap every coroutine in a tasks.

experiments/simbricks/orchestration/exectools.py

@@ -100,14 +100,14 @@ class Component(object):
                 return
 
     async def _waiter(self):
-        out_handlers = asyncio.ensure_future(
-            asyncio.wait([
-                self._read_stream(self._proc.stdout, self._consume_out),
+        stdout_handler = asyncio.create_task(
+            self._read_stream(self._proc.stdout, self._consume_out)
+        )
+        stderr_handler = asyncio.create_task(
             self._read_stream(self._proc.stderr, self._consume_err)
-            ])
         )
         rc = await self._proc.wait()
-        await out_handlers
+        await asyncio.wait([stdout_handler, stderr_handler])
         await self.terminated(rc)
 
     async def send_input(self, bs, eof=False):
@@ -158,20 +158,28 @@ class Component(object):
         """Attempts to stop this component by sending signals in the following
         order: interrupt, terminate, kill."""
         await self.interrupt()
-        _, pending = await asyncio.wait([self._proc.wait()], timeout=delay)
-        if len(pending) != 0:
+        try:
+            await asyncio.wait_for(self._proc.wait(), delay)
+            return
+        except TimeoutError:
             print(
                 f'terminating component {self.cmd_parts[0]} '
-                f'pid {self._proc.pid}'
+                f'pid {self._proc.pid}',
+                flush=True
             )
             await self.terminate()
-            _, pending = await asyncio.wait([self._proc.wait()], timeout=delay)
-            if len(pending) != 0:
+
+        try:
+            await asyncio.wait_for(self._proc.wait(), delay)
+            return
+        except TimeoutError:
             print(
                 f'killing component {self.cmd_parts[0]} '
-                    f'pid {self._proc.pid}'
+                f'pid {self._proc.pid}',
+                flush=True
             )
             await self.kill()
+
         await self._proc.wait()
 
     async def started(self):
@@ -339,7 +347,9 @@ class Executor(object):
     async def await_files(self, paths, *args, **kwargs):
         xs = []
         for p in paths:
-            xs.append(self.await_file(p, *args, **kwargs))
+            waiter = asyncio.create_task(self.await_file(p, *args, **kwargs))
+            xs.append(waiter)
+
         await asyncio.wait(xs)
 
 

experiments/simbricks/orchestration/runners.py

@@ -125,7 +125,9 @@ class ExperimentBaseRunner(ABC):
             if self.verbose:
                 print('preparing config tar:', path)
             host.node_config.make_tar(path)
-            copies.append(self.sim_executor(host).send_file(path, self.verbose))
+            executor = self.sim_executor(host)
+            task = asyncio.create_task(executor.send_file(path, self.verbose))
+            copies.append(task)
         await asyncio.wait(copies)
 
         # prepare all simulators in parallel
@@ -133,11 +135,12 @@ class ExperimentBaseRunner(ABC):
         for sim in self.exp.all_simulators():
             prep_cmds = list(sim.prep_cmds(self.env))
             executor = self.sim_executor(sim)
-            sims.append(
+            task = asyncio.create_task(
                 executor.run_cmdlist(
                     'prepare_' + self.exp.name, prep_cmds, verbose=self.verbose
                 )
             )
+            sims.append(task)
         await asyncio.wait(sims)
 
     async def wait_for_sims(self):
@@ -156,14 +159,14 @@ class ExperimentBaseRunner(ABC):
             ts.prepare()
             while ts.is_active():
                 # start ready simulators in parallel
-                starts = []
+                starting = []
                 sims = []
                 for sim in ts.get_ready():
-                    starts.append(self.start_sim(sim))
+                    starting.append(asyncio.create_task(self.start_sim(sim)))
                     sims.append(sim)
 
                 # wait for starts to complete
-                await asyncio.wait(starts)
+                await asyncio.wait(starting)
 
                 for sim in sims:
                     ts.done(sim)
@@ -190,7 +193,7 @@ class ExperimentBaseRunner(ABC):
             # "interrupt, terminate, kill" all processes
             scs = []
             for _, sc in self.running:
-                scs.append(sc.int_term_kill())
+                scs.append(asyncio.create_task(sc.int_term_kill()))
             await asyncio.wait(scs)
 
             # wait for all processes to terminate
@@ -200,7 +203,7 @@ class ExperimentBaseRunner(ABC):
             # remove all sockets
             scs = []
             for (executor, sock) in self.sockets:
-                scs.append(executor.rmtree(sock))
+                scs.append(asyncio.create_task(executor.rmtree(sock)))
             if scs:
                 await asyncio.wait(scs)