Add support for parallel execution in host

We define a list of processes which are currently running inside host
and we populate it each time run_command() implementation is called.
In this way, we keep a list of running processes and all of them can
be killed by running stop() method.

Signed-off-by: Andrea Cervesato <[email protected]>

Author: acerv

ltp/host.py

@@ -33,7 +33,7 @@ def __init__(self) -> None:
         self._initialized = False
         self._cmd_lock = threading.Lock()
         self._fetch_lock = threading.Lock()
-        self._proc = None
+        self._procs = []
         self._stop = False
         self._cwd = None
         self._env = None
@@ -85,18 +85,23 @@ def _inner_stop(self, sig: int, timeout: float = 30) -> None:
 
         self._stop = True
 
-        if self._proc:
-            self._logger.info("Terminating process with %s", sig)
-            self._proc.send_signal(sig)
-
         with Timeout(timeout) as timer:
-            while self._fetch_lock.locked():
-                time.sleep(1e-6)
-                timer.check(err_msg="Timeout waiting for command to stop")
+            if self._procs:
+                self._logger.info(
+                    "Terminating %d process(es) with %s",
+                    len(self._procs), sig)
+
+                for proc in self._procs:
+                    proc.send_signal(sig)
+
+                    while proc.poll() is None:
+                        time.sleep(1e-6)
+                        timer.check(
+                            err_msg="Timeout waiting for command to stop")
 
-            while self._cmd_lock.locked():
+            while self._fetch_lock.locked():
                 time.sleep(1e-6)
-                timer.check(err_msg="Timeout waiting for command to stop")
+                timer.check(err_msg="Timeout waiting to fetch file")
 
         self._logger.info("Process terminated")
 
@@ -114,14 +119,18 @@ def force_stop(
             iobuffer: IOBuffer = None) -> None:
         self._inner_stop(signal.SIGKILL, timeout)
 
-    def _read_stdout(self, size: int, iobuffer: IOBuffer = None) -> str:
+    def _read_stdout(
+            self,
+            proc: subprocess.Popen,
+            size: int,
+            iobuffer: IOBuffer = None) -> str:
         """
         Read data from stdout.
         """
         if not self.is_running:
             return None
 
-        data = os.read(self._proc.stdout.fileno(), size)
+        data = os.read(proc.stdout.fileno(), size)
         rdata = data.decode(encoding="utf-8", errors="replace")
         rdata = rdata.replace('\r', '')
 
@@ -143,82 +152,82 @@ def run_command(self,
         if not self.is_running:
             raise SUTError("SUT is not running")
 
-        with self._cmd_lock:
-            t_secs = max(timeout, 0)
-
-            self._logger.info("Executing command (timeout=%d): %s",
-                              t_secs,
-                              command)
-
-            # pylint: disable=consider-using-with
-            self._proc = subprocess.Popen(
-                command,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                cwd=self._cwd,
-                env=self._env,
-                shell=True)
-
-            ret = None
-            t_start = time.time()
-            t_end = 0
-            stdout = ""
+        t_secs = max(timeout, 0)
+
+        self._logger.info(
+            "Executing command (timeout=%.3f): %s",
+            t_secs,
+            command)
+
+        # pylint: disable=consider-using-with
+        proc = subprocess.Popen(
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            cwd=self._cwd,
+            env=self._env,
+            shell=True)
+
+        self._procs.append(proc)
+
+        ret = None
+        t_start = time.time()
+        t_end = 0
+        stdout = ""
+
+        try:
+            poller = select.epoll()
+            poller.register(
+                proc.stdout.fileno(),
+                select.POLLIN |
+                select.POLLPRI |
+                select.POLLHUP |
+                select.POLLERR)
+
+            with Timeout(timeout) as timer:
+                while True:
+                    events = poller.poll(0.1)
+                    for fdesc, _ in events:
+                        if fdesc != proc.stdout.fileno():
+                            break
 
-            try:
-                poller = select.epoll()
-                poller.register(
-                    self._proc.stdout.fileno(),
-                    select.POLLIN |
-                    select.POLLPRI |
-                    select.POLLHUP |
-                    select.POLLERR)
+                        data = self._read_stdout(proc, 1024, iobuffer)
+                        if data:
+                            stdout += data
 
-                with Timeout(timeout) as timer:
-                    while True:
-                        events = poller.poll(0.1)
-                        for fdesc, _ in events:
-                            if fdesc != self._proc.stdout.fileno():
-                                break
-
-                            data = self._read_stdout(1024, iobuffer)
-                            if data:
-                                stdout += data
+                    if proc.poll() is not None:
+                        break
 
-                        if self._proc.poll() is not None:
-                            break
+                    timer.check(
+                        err_msg="Timeout during command execution",
+                        exc=SUTTimeoutError)
 
-                        timer.check(
-                            err_msg="Timeout during command execution",
-                            exc=SUTTimeoutError)
+            t_end = time.time() - t_start
 
-                t_end = time.time() - t_start
+            # once the process stopped, we still might have some data
+            # inside the stdout buffer
+            while not self._stop:
+                data = self._read_stdout(proc, 1024, iobuffer)
+                if not data:
+                    break
 
-                # once the process stopped, we still might have some data
-                # inside the stdout buffer
-                while not self._stop:
-                    data = self._read_stdout(1024, iobuffer)
-                    if not data:
-                        break
+                stdout += data
+        finally:
+            self._procs.remove(proc)
 
-                    stdout += data
-            except subprocess.TimeoutExpired as err:
-                self._proc.kill()
-                raise SUTError(err)
-            finally:
-                ret = {
-                    "command": command,
-                    "stdout": stdout,
-                    "returncode": self._proc.returncode,
-                    "timeout": t_secs,
-                    "exec_time": t_end,
-                }
+            ret = {
+                "command": command,
+                "stdout": stdout,
+                "returncode": proc.returncode,
+                "timeout": t_secs,
+                "exec_time": t_end,
+            }
 
-                self._logger.debug("return data=%s", ret)
-                self._proc = None
+            self._logger.debug("return data=%s", ret)
 
-            self._logger.info("Command executed")
+        self._logger.info("Command executed")
 
-            return ret
+        return ret
 
     def fetch_file(
             self,

ltp/tests/test_host.py

@@ -1,7 +1,13 @@
 """
 Test SUT implementations.
 """
+import os
+import time
+import threading
+from concurrent.futures import ThreadPoolExecutor
 import pytest
+from ltp.sut import IOBuffer
+from ltp.sut import SUTTimeoutError
 from ltp.host import HostSUT
 from ltp.tests.sut import _TestSUT
 from ltp.tests.sut import Printer
@@ -59,3 +65,97 @@ def test_env(self, tmpdir):
         ret = sut.run_command("cat $FILE", timeout=2, iobuffer=Printer())
         assert ret["returncode"] == 0
         assert ret["stdout"] == "runltp-ng tests"
+
+    def test_multiple_commands(self, sut):
+        """
+        Execute run_command multiple times.
+        """
+        sut.communicate()
+
+        def _runner(index):
+            return sut.run_command(f"echo -n {index}", timeout=15)
+
+        results = []
+
+        with ThreadPoolExecutor(max_workers=os.cpu_count()) as executor:
+            for result in executor.map(_runner, range(100)):
+                results.append(result)
+
+        for i in range(100):
+            data = results[i]
+
+            assert data["command"] == f"echo -n {i}"
+            assert data["timeout"] == 15
+            assert data["returncode"] == 0
+            assert data["stdout"] == f"{i}"
+            assert 0 < data["exec_time"] < time.time()
+
+    def test_multiple_commands_timeout(self, sut):
+        """
+        Execute run_command multiple times with low timeout.
+        """
+        sut.communicate()
+
+        def _runner(_):
+            with pytest.raises(SUTTimeoutError):
+                sut.run_command("sleep 1", timeout=0.1)
+
+        with ThreadPoolExecutor(max_workers=os.cpu_count()) as executor:
+            executor.map(_runner, range(100))
+
+    def test_multiple_commands_stop(self, sut):
+        """
+        Execute run_command multiple time, then call stop().
+        """
+        class MyBuffer(IOBuffer):
+            """
+            For each echo command, we store 1 inside `executed` list.
+            At the end of all commands executions we know how many
+            commands are sleeping by checking `executed` list.
+            """
+            executed = []
+
+            def write(self, _: str) -> None:
+                self.executed.append(1)
+
+            def flush(self) -> None:
+                pass
+
+        buffer = MyBuffer()
+        results = []
+        cpu_count = os.cpu_count()
+        exec_count = cpu_count - 1
+        sut.communicate()
+
+        def _threaded():
+            def _runner(index):
+                return sut.run_command(
+                    f"echo -n {index}; sleep 3",
+                    timeout=5,
+                    iobuffer=buffer)
+
+            with ThreadPoolExecutor(max_workers=cpu_count) as executor:
+                for result in executor.map(_runner, range(exec_count)):
+                    results.append(result)
+
+        thread = threading.Thread(target=_threaded, daemon=True)
+        thread.start()
+
+        while len(buffer.executed) < exec_count:
+            time.sleep(0.001)
+            continue
+
+        sut.force_stop()
+
+        while len(results) < exec_count:
+            time.sleep(0.001)
+            continue
+
+        for i in range(exec_count):
+            data = results[i]
+
+            assert data["command"] == f"echo -n {i}; sleep 3"
+            assert data["timeout"] == 5
+            assert data["returncode"] != 0
+            assert data["stdout"] == f"{i}"
+            assert 0 < data["exec_time"] < time.time()