Fix Windows passthrough NTTTCP readiness

Fail Windows-host passthrough NTTTCP tests when receiver output cannot be parsed instead of reusing sender totals.

Add bounded listener readiness checks for Linux and Windows NTTTCP receivers, and fall back to reboot when systemctl reboot exits unsuccessfully.

Author: vyadavmsft

lisa/microsoft/testsuites/device_passthrough/functional_tests.py

@@ -143,9 +143,7 @@ def _vendor_device_from_host_device(
             }
 
         cloud_hypervisor = platform
-        bdf = (
-            f"{device.domain}:{device.bus}:{device.slot}.{device.function}"
-        ).lower()
+        bdf = f"{device.domain}:{device.bus}:{device.slot}.{device.function}".lower()
         cat = cloud_hypervisor.host_node.tools[Cat]
         vendor_raw = cat.read(f"/sys/bus/pci/devices/{bdf}/vendor", sudo=True).strip()
         device_raw = cat.read(f"/sys/bus/pci/devices/{bdf}/device", sudo=True).strip()

lisa/microsoft/testsuites/performance/networkperf_passthrough.py

@@ -29,13 +29,13 @@
 from lisa.sut_orchestrator import CLOUD_HYPERVISOR, HYPERV
 from lisa.testsuite import TestResult
 from lisa.tools import Dhclient, Kill, PowerShell, Sysctl
+from lisa.tools.ip import Ip
 from lisa.tools.iperf3 import (
     IPERF_TCP_BUFFER_LENGTHS,
     IPERF_TCP_CONCURRENCY,
     IPERF_UDP_BUFFER_LENGTHS,
     IPERF_UDP_CONCURRENCY,
 )
-from lisa.tools.ip import Ip
 from lisa.tools.ntttcp import NTTTCP_TCP_CONCURRENCY, NTTTCP_UDP_CONCURRENCY, Ntttcp
 from lisa.util import (
     LisaException,
@@ -1172,12 +1172,17 @@ def _perf_ntttcp_with_windows_server(
                     if udp_mode
                     else server_ntttcp.create_ntttcp_result(receiver_result)
                 )
-            except (AssertionError, LisaException):
-                client.log.debug(
-                    "NTTTCP receiver output was not parseable; using sender "
-                    "totals for receiver-side metrics."
-                )
-                parsed_server_result = parsed_client_result
+            except (AssertionError, LisaException) as parse_error:
+                receiver_node = "Linux guest" if udp_mode else "Windows host"
+                raise LisaException(
+                    f"Failed to parse NTTTCP receiver output from {receiver_node} "
+                    f"for {test_case_name} with {test_thread} connections. "
+                    "Verify that the receiver completed and emitted NTTTCP "
+                    "totals before publishing performance data. "
+                    f"Exit code: {receiver_result.exit_code}. "
+                    f"Stdout: {receiver_result.stdout[:2000]}. "
+                    f"Stderr: {receiver_result.stderr[:2000]}"
+                ) from parse_error
             if udp_mode:
                 perf_message = client_ntttcp.create_ntttcp_udp_performance_message(
                     parsed_server_result,

lisa/tools/ntttcp.py

@@ -19,7 +19,7 @@
 from lisa.tools import Firewall, Gcc, Git, Lscpu, Make, Sed
 from lisa.tools.powershell import PowerShell
 from lisa.tools.taskset import TaskSet
-from lisa.util import LisaException, constants
+from lisa.util import LisaException, check_till_timeout, constants
 from lisa.util.process import ExecutableResult, Process
 
 from .sysctl import Sysctl
@@ -245,18 +245,48 @@ def run_as_server_async(
             shell=True,
             sudo=True,
         )
-        # NTTTCP for Linux 1.4.0
-        # ---------------------------------------------------------
-        # 01:16:35 INFO: no role specified. use receiver role
-        # 01:16:35 INFO: 65 threads created
-        # above output means ntttcp server is ready. In no-sync mode, the
-        # receiver may not print the same readiness line before clients start.
         if no_sync:
-            time.sleep(5)
+            self._wait_server_port_ready(process, udp_mode)
         else:
             process.wait_output("threads created")
         return process
 
+    def _wait_server_port_ready(
+        self,
+        process: Process,
+        udp_mode: bool,
+        port: int = 5001,
+    ) -> None:
+        def is_ready() -> bool:
+            if not process.is_running():
+                raise LisaException(
+                    "ntttcp receiver exited before the listen port became ready"
+                )
+            return self._is_server_port_open(udp_mode, port)
+
+        protocol = "UDP" if udp_mode else "TCP"
+        check_till_timeout(
+            is_ready,
+            timeout_message=f"wait for ntttcp {protocol} receiver port {port} open",
+            timeout=30,
+        )
+
+    def _is_server_port_open(self, udp_mode: bool, port: int) -> bool:
+        ss_options = "-lun" if udp_mode else "-ltn"
+        netstat_options = "-lnu" if udp_mode else "-ltn"
+        result = self.node.execute(
+            cmd=(
+                f"(ss {ss_options} 2>/dev/null || "
+                f"netstat {netstat_options} 2>/dev/null || true) "
+                f"| grep -E '[:.]?{port}[[:space:]]'"
+            ),
+            shell=True,
+            sudo=True,
+            no_info_log=True,
+            no_error_log=True,
+        )
+        return result.exit_code == 0
+
     def run_as_server(
         self,
         nic_name: str,
@@ -979,7 +1009,7 @@ def create_ntttcp_result(
 
 class WindowsNtttcp(Ntttcp):
     _download_url = (
-        "https://github.com/microsoft/ntttcp/releases/latest/download/ntttcp.exe"
+        "https://github.com/microsoft/ntttcp/releases/latest/download/" "ntttcp.exe"
     )
     _total_mbps_pattern = re.compile(
         r"(?im)^\s*TOTAL\s+(?P<throughput>[0-9]+(?:\.[0-9]+)?)\s*$"
@@ -1003,8 +1033,7 @@ def dependencies(self) -> List[Type[Tool]]:
 
     def setup_system(self, udp_mode: bool = False, set_task_max: bool = True) -> None:
         self.node.tools[PowerShell].run_cmdlet(
-            "Set-NetFirewallProfile -Profile Domain,Public,Private "
-            "-Enabled False",
+            "Set-NetFirewallProfile -Profile Domain,Public,Private " "-Enabled False",
             fail_on_error=False,
         )
 
@@ -1037,19 +1066,31 @@ def run_as_server_async(
             shell=True,
             sudo=True,
         )
-        if udp_mode:
-            time.sleep(5)
-        else:
-            self.node.tools[PowerShell].run_cmdlet(
-                "for ($i = 0; $i -lt 10; $i++) { "
-                "if (Get-NetTCPConnection -State Listen -LocalPort 5001 "
-                "-ErrorAction SilentlyContinue) { exit 0 }; "
-                "Start-Sleep -Seconds 1 }; exit 1",
-                force_run=True,
-                timeout=15,
-            )
+        self._wait_receiver_port_ready(udp_mode)
         return process
 
+    def _wait_receiver_port_ready(
+        self,
+        udp_mode: bool,
+        port: int = 5001,
+        timeout: int = 30,
+    ) -> None:
+        endpoint_cmdlet = (
+            f"Get-NetUDPEndpoint -LocalPort {port} -ErrorAction SilentlyContinue"
+            if udp_mode
+            else f"Get-NetTCPConnection -State Listen -LocalPort {port} "
+            "-ErrorAction SilentlyContinue"
+        )
+        self.node.tools[PowerShell].run_cmdlet(
+            "$deadline = (Get-Date).AddSeconds(" + str(timeout) + "); "
+            "do { "
+            f"if ({endpoint_cmdlet}) {{ exit 0 }}; "
+            "Start-Sleep -Milliseconds 500 "
+            "} while ((Get-Date) -lt $deadline); exit 1",
+            force_run=True,
+            timeout=timeout + 5,
+        )
+
     def run_as_client(
         self,
         nic_name: str,

lisa/tools/reboot.py

@@ -60,6 +60,8 @@ def _get_last_boot_time(self) -> datetime:
 
     def _wait_ssh_session_stable(self, time_out: int) -> None:
         timer = create_timer()
+        consecutive_successes = 0
+        last_error = ""
         while timer.elapsed(False) < time_out:
             try:
                 self.node.close()
@@ -69,21 +71,26 @@ def _wait_ssh_session_stable(self, time_out: int) -> None:
                     timeout=10,
                     no_info_log=True,
                 ).assert_exit_code()
-                sleep(10)
-                self.node.close()
-                self.node.execute(
-                    "echo lisa reboot ready",
-                    shell=True,
-                    timeout=10,
-                    no_info_log=True,
-                ).assert_exit_code()
-                return
+                consecutive_successes += 1
+                if consecutive_successes >= 2:
+                    return
             except Exception as e:
+                consecutive_successes = 0
+                last_error = str(e)
                 self._log.debug(f"waiting for stable ssh session after reboot: {e}")
-                sleep(5)
+            sleep(2)
         raise LisaException(
-            f"cannot get stable ssh session after reboot in {time_out} seconds"
+            f"cannot get stable ssh session after reboot in {time_out} seconds. "
+            f"Last error: {last_error}"
+        )
+
+    def _run_reboot_command(self) -> None:
+        command_result = self.node.execute(
+            "command -v reboot", shell=True, sudo=True, no_info_log=True
         )
+        if command_result.exit_code == 0:
+            self._command = command_result.stdout.strip()
+        self.run(force_run=True, sudo=True, timeout=10)
 
     def reboot_and_check_panic(self, log_path: Path) -> None:
         try:
@@ -132,19 +139,22 @@ def reboot(self, time_out: int = 300) -> None:
                 "command -v systemctl", shell=True, sudo=True, no_info_log=True
             )
             if systemctl_result.exit_code == 0:
-                self.node.execute(
+                reboot_result = self.node.execute(
                     "systemctl reboot -i",
                     shell=True,
                     sudo=True,
                     timeout=10,
                 )
+                if reboot_result.exit_code != 0:
+                    self._log.debug(
+                        "systemctl reboot failed with exit code "
+                        f"{reboot_result.exit_code}; falling back to reboot. "
+                        f"stdout: {reboot_result.stdout}, "
+                        f"stderr: {reboot_result.stderr}"
+                    )
+                    self._run_reboot_command()
             else:
-                command_result = self.node.execute(
-                    "command -v reboot", shell=True, sudo=True, no_info_log=True
-                )
-                if command_result.exit_code == 0:
-                    self._command = command_result.stdout
-                self.run(force_run=True, sudo=True, timeout=10)
+                self._run_reboot_command()
         except Exception as e:
             # it doesn't matter to exceptions here. The system may reboot fast
             self._log.debug(f"ignorable exception on rebooting: {e}")