Exit the sim after the desired number of code.py runs

Make UART PTY reliable by waiting for all bytes to be read from
the subsidiary PTY.

Author: tannewt

ports/zephyr-cp/supervisor/port.c

@@ -17,6 +17,15 @@
 #include <zephyr/kernel.h>
 #include <zephyr/sys/reboot.h>
 
+#if defined(CONFIG_ARCH_POSIX)
+#include <limits.h>
+#include <stdlib.h>
+
+#include "cmdline.h"
+#include "posix_board_if.h"
+#include "posix_native_task.h"
+#endif
+
 #include "lib/tlsf/tlsf.h"
 #include <zephyr/device.h>
 
@@ -37,6 +46,32 @@ static K_EVENT_DEFINE(main_needed);
 
 static struct k_timer tick_timer;
 
+#if defined(CONFIG_ARCH_POSIX)
+// Number of VM runs before exiting.
+// <= 0 means run forever.
+// INT32_MAX means option was not provided.
+static int32_t native_sim_vm_runs = INT32_MAX;
+static uint32_t native_sim_reset_port_count = 0;
+
+static struct args_struct_t native_sim_reset_port_args[] = {
+    {
+        .option = "vm-runs",
+        .name = "count",
+        .type = 'i',
+        .dest = &native_sim_vm_runs,
+        .descript = "Exit native_sim after this many VM runs. "
+            "Example: --vm-runs=2"
+    },
+    ARG_TABLE_ENDMARKER
+};
+
+static void native_sim_register_cmdline_opts(void) {
+    native_add_command_line_opts(native_sim_reset_port_args);
+}
+
+NATIVE_TASK(native_sim_register_cmdline_opts, PRE_BOOT_1, 0);
+#endif
+
 static void _tick_function(struct k_timer *timer_id) {
     supervisor_tick();
 }
@@ -61,6 +96,16 @@ void reset_port(void) {
     #if CIRCUITPY_BLEIO
     bleio_reset();
     #endif
+
+    #if defined(CONFIG_ARCH_POSIX)
+    native_sim_reset_port_count++;
+    if (native_sim_vm_runs != INT32_MAX &&
+        native_sim_vm_runs > 0 &&
+        native_sim_reset_port_count >= (uint32_t)(native_sim_vm_runs + 1)) {
+        printk("posix: exiting after %d VM runs\n", native_sim_vm_runs);
+        posix_exit(0);
+    }
+    #endif
 }
 
 void reset_to_bootloader(void) {

ports/zephyr-cp/tests/__init__.py

@@ -1,6 +1,7 @@
 import serial
 import subprocess
 import threading
+import time
 
 
 class StdSerial:
@@ -25,6 +26,12 @@ def close(self):
             self.stdin.close()
         self.stdout.close()
 
+    @property
+    def in_waiting(self):
+        if self.stdout is None:
+            return 0
+        return len(self.stdout.peek())
+
 
 class SerialSaver:
     """Capture serial output in a background thread so output isn't missed."""
@@ -58,6 +65,13 @@ def _reader_loop(self):
             with self._cv:
                 self.all_output += text
                 self._cv.notify_all()
+        in_waiting = 0
+        try:
+            in_waiting = self.serial.in_waiting
+        except OSError:
+            pass
+        if in_waiting > 0:
+            self.all_output += self.serial.read().decode("utf-8", errors="replace")
 
     def wait_for(self, text, timeout=10):
         with self._cv:
@@ -79,11 +93,11 @@ def close(self):
             return
 
         self._stop.set()
+        self._reader.join(timeout=1.0)
         try:
             self.serial.close()
         except Exception:
             pass
-        self._reader.join(timeout=1.0)
         self.serial = None
 
     def write(self, text):
@@ -131,6 +145,7 @@ def shutdown(self):
         self.debug_serial.close()
 
     def wait_until_done(self):
-        self._proc.wait(timeout=self._timeout)
-        self.serial.close()
-        self.debug_serial.close()
+        start_time = time.monotonic()
+        while self._proc.poll() is None and time.monotonic() - start_time < self._timeout:
+            time.sleep(0.01)
+        self.shutdown()

ports/zephyr-cp/tests/bsim/conftest.py

@@ -86,7 +86,7 @@ def shutdown(self) -> None:
 @pytest.fixture
 def bsim_phy(request, bsim_phy_binary, native_sim_env, sim_id):
     duration_marker = request.node.get_closest_marker("duration")
-    sim_length = float(duration_marker.args[0]) if duration_marker else 20.0
+    duration = float(duration_marker.args[0]) if duration_marker else 20.0
 
     devices = 1
     if "circuitpython2" in request.fixturenames or "zephyr_sample" in request.fixturenames:
@@ -97,12 +97,7 @@ def bsim_phy(request, bsim_phy_binary, native_sim_env, sim_id):
         sample_device_id = int(sample_marker.kwargs.get("device_id", 1))
         devices = max(devices, sample_device_id + 1)
 
-    bsim_marker = request.node.get_closest_marker("bsim")
-    if bsim_marker is not None:
-        devices = int(bsim_marker.kwargs.get("devices", devices))
-        sim_length = float(bsim_marker.kwargs.get("sim_length", sim_length))
-
-    sim_length_us = int(sim_length * 1e6)
+    sim_length_us = int(duration * 1e6)
     cmd = [
         "stdbuf",
         "-oL",
@@ -136,7 +131,7 @@ def bsim_phy(request, bsim_phy_binary, native_sim_env, sim_id):
             raise RuntimeError("bsim PHY exited immediately")
         # Assume bsim is running
 
-    phy = BsimPhyInstance(proc, phy_output, timeout=sim_length)
+    phy = BsimPhyInstance(proc, phy_output, timeout=duration)
     yield phy
     phy.shutdown()
 
@@ -195,7 +190,12 @@ def zephyr_sample(request, bsim_phy, native_sim_env, sim_id):
     if not binary.exists():
         pytest.skip(f"Zephyr sample binary not found: {binary}")
 
-    cmd = [str(binary), f"-s={sim_id}", f"-d={device_id}"]
+    cmd = [
+        str(binary),
+        f"-s={sim_id}",
+        f"-d={device_id}",
+        "-disconnect_on_exit=1",
+    ]
     logger.info("Running: %s", " ".join(cmd))
     proc = subprocess.Popen(
         cmd,

ports/zephyr-cp/tests/bsim/test_bsim_basics.py

@@ -14,16 +14,13 @@
 
 @pytest.mark.circuitpy_drive({"code.py": BSIM_CODE})
 @pytest.mark.circuitpy_drive({"code.py": BSIM_CODE})
+@pytest.mark.duration(3)
 def test_bsim_dual_instance_connect(bsim_phy, circuitpython1, circuitpython2):
     """Run two bsim instances on the same sim id and verify UART output."""
-    print("in the test")
 
-    # Wait for both devices to produce their expected output before
-    # tearing down the simulation.
-    circuitpython1.serial.wait_for("bsim ready")
-    circuitpython2.serial.wait_for("bsim ready")
-
-    bsim_phy.finish_sim()
+    # Wait for both devices to complete before checking output.
+    circuitpython1.wait_until_done()
+    circuitpython2.wait_until_done()
 
     output0 = circuitpython1.serial.all_output
     output1 = circuitpython2.serial.all_output

ports/zephyr-cp/tests/bsim/test_bsim_ble_advertising.py

@@ -4,7 +4,6 @@
 """BLE advertising tests for nrf5340bsim."""
 
 import logging
-import time
 
 import pytest
 
@@ -39,8 +38,9 @@
 print("adv run start")
 adapter.start_advertising(advertisement, connectable=False)
 print("adv running")
-while True:
-    time.sleep(0.2)
+time.sleep(10)
+adapter.stop_advertising()
+print("adv run done")
 """
 
 
@@ -50,54 +50,34 @@ def test_bsim_advertise_and_scan(bsim_phy, circuitpython, zephyr_sample):
     """Advertise from CircuitPython and verify Zephyr observer sees traffic."""
     observer = zephyr_sample
 
-    start_time = time.time()
-    while time.time() - start_time < 12.0:
-        observer_output = observer.serial.all_output
-        adv_ready = "adv started" in circuitpython.serial.all_output
-        if (
-            "Device found:" in observer_output
-            and "AD data len 10" in observer_output
-            and adv_ready
-        ):
-            break
-        time.sleep(0.05)
+    circuitpython.wait_until_done()
 
+    cp_output = circuitpython.serial.all_output
     observer_output = observer.serial.all_output
-    assert "adv start" in circuitpython.serial.all_output
-    assert "adv started" in circuitpython.serial.all_output
+    assert "adv start" in cp_output
+    assert "adv started" in cp_output
+    assert "adv stop" in cp_output
     assert "Device found:" in observer_output
     assert "AD data len 10" in observer_output
 
 
-@pytest.mark.zephyr_sample("bluetooth/observer", timeout=20.0)
+@pytest.mark.zephyr_sample("bluetooth/observer")
+@pytest.mark.code_py_runs(2)
+@pytest.mark.duration(25)
 @pytest.mark.circuitpy_drive({"code.py": BSIM_ADV_INTERRUPT_RELOAD_CODE})
 def test_bsim_advertise_ctrl_c_reload(bsim_phy, circuitpython, zephyr_sample):
     """Ensure advertising resumes after Ctrl-C and a reload."""
     observer = zephyr_sample
 
-    start_time = time.time()
-    sent_ctrl_c = False
-    sent_reload = False
-    observer_count_before = 0
-
-    while time.time() - start_time < 22.0:
-        cp_output = circuitpython.serial.all_output
-        observer_output = observer.serial.all_output
-        device_found_count = observer_output.count("Device found:")
-
-        if not sent_ctrl_c and "adv running" in cp_output and device_found_count > 0:
-            circuitpython.serial.write("\x03")
-            sent_ctrl_c = True
-            observer_count_before = device_found_count
-
-        if sent_ctrl_c and not sent_reload and "KeyboardInterrupt" in cp_output:
-            circuitpython.serial.write("\x04")
-            sent_reload = True
+    circuitpython.serial.wait_for("adv running")
+    observer.serial.wait_for("Device found:")
+    observer_count_before = observer.serial.all_output.count("Device found:")
 
-        if sent_reload and cp_output.count("adv running") >= 2:
-            break
+    circuitpython.serial.write("\x03")
+    circuitpython.serial.wait_for("KeyboardInterrupt")
 
-        time.sleep(0.05)
+    circuitpython.serial.write("\x04")
+    circuitpython.wait_until_done()
 
     cp_output = circuitpython.serial.all_output
     observer_output = observer.serial.all_output
@@ -107,5 +87,6 @@ def test_bsim_advertise_ctrl_c_reload(bsim_phy, circuitpython, zephyr_sample):
     assert "adv run start" in cp_output
     assert "KeyboardInterrupt" in cp_output
     assert cp_output.count("adv running") >= 2
-    assert observer_output.count("Device found:") >= observer_count_before
+    assert cp_output.count("adv run done") >= 1
+    assert observer_output.count("Device found:") >= observer_count_before + 1
     assert "Already advertising" not in cp_output

ports/zephyr-cp/tests/bsim/test_bsim_ble_name.py

@@ -3,8 +3,6 @@
 
 """BLE name tests for nrf5340bsim."""
 
-import time
-
 import pytest
 
 pytestmark = pytest.mark.circuitpython_board("native_nrf5340bsim")
@@ -22,10 +20,6 @@
 @pytest.mark.circuitpy_drive({"code.py": BSIM_NAME_CODE})
 def test_bsim_set_name(bsim_phy, circuitpython):
     """Set the BLE name and read it back on bsim."""
-    start_time = time.time()
-    while time.time() - start_time < 3.0:
-        if "name CPNAME" in circuitpython.serial.all_output:
-            break
-        time.sleep(0.05)
+    circuitpython.wait_until_done()
 
     assert "name CPNAME" in circuitpython.serial.all_output