Function stream_compile_and_run_tests

ci/meson/streaming.py:547–774

Stream test compilation and execution with overlap. Tests start executing as soon as they finish linking, while remaining tests continue compiling. This reduces total wall-clock time compared to a sequential compile-then-run approach. Args: build_dir: Meson build direc

(
    build_dir: Path,
    test_callback: Callable[[Path], TestResult],
    target: Optional[str] = None,
    verbose: bool = False,
    compile_timeout: int = 600,
    build_optimizer: Optional[BuildOptimizer] = None,
    test_file_filter: Optional[str] = None,
    build_timer=None,
    max_failures: int = 10,
)

def stream_compile_and_run_tests(
    build_dir: Path,
    test_callback: Callable[[Path], TestResult],
    target: Optional[str] = None,
    verbose: bool = False,
    compile_timeout: int = 600,
    build_optimizer: Optional[BuildOptimizer] = None,
    test_file_filter: Optional[str] = None,
    build_timer=None,
    max_failures: int = 10,
) -> StreamingResult:
    """
    Stream test compilation and execution with overlap.

    Tests start executing as soon as they finish linking, while remaining
    tests continue compiling. This reduces total wall-clock time compared
    to a sequential compile-then-run approach.

    Args:
        build_dir: Meson build directory
        test_callback: Function called with each completed test path.
                      Returns a TestResult indicating pass/fail and captured output.
        target: Specific target to build (None = all)
        verbose: Show detailed progress messages (default: False)
        compile_timeout: Timeout in seconds for compilation (default: 600)
        build_optimizer: Optional BuildOptimizer for DLL relink suppression.
        test_file_filter: Optional .hpp filename to filter test execution (e.g., "backbeat.hpp")
        build_timer: Optional BuildTimer for recording test_execution_done checkpoint.
        max_failures: Maximum number of test failures before halting (default: 10, 0 = unlimited).
    """
    # Pass test file filter to the callback via an attribute so it can
    # inject it into the subprocess environment (os.environ is stale after
    # _streaming_env was copied).
    if test_file_filter:
        setattr(test_callback, "_test_file_filter", test_file_filter)

    # Prepare concurrent test execution infrastructure.
    # Tests are submitted to the executor as they finish linking (during
    # compilation), overlapping test execution with ongoing compilation.
    max_workers = min(os.cpu_count() or 4, 16)
    halt_event = threading.Event()
    _kill_all = getattr(test_callback, "kill_all", None)

    @dataclass
    class _WorkerResult:
        """Internal result from a worker thread."""

        test_path: Path
        test_result: TestResult
        error_msg: str = ""

    def _run_one_test(test_path: Path) -> _WorkerResult:
        """Run a single test in a worker thread."""
        if halt_event.is_set():
            return _WorkerResult(test_path, TestResult(success=False))
        try:
            return _WorkerResult(test_path, test_callback(test_path))
        except KeyboardInterrupt as ki: