Function start_server

scripts/benchmark_discovery.py:545–589

(args: argparse.Namespace, socket_path: Path)

def start_server(args: argparse.Namespace, socket_path: Path) -> subprocess.Popen[str]:
    socket_path.unlink(missing_ok=True)
    command = [
        args.jcode,
        "--socket",
        str(socket_path),
        "--no-selfdev",
        "--no-update",
    ]
    if args.provider:
        command += ["--provider", args.provider]
    command += ["serve", "--server-name", f"discovery-benchmark-{os.getpid()}"]
    environment = benchmark_environment(socket_path)
    log_path = socket_path.parent / "server.log"
    log = log_path.open("w", encoding="utf-8")
    process = subprocess.Popen(
        command,
        stdout=log,
        stderr=subprocess.STDOUT,
        text=True,
        env=environment,
        start_new_session=True,
    )
    log.close()
    deadline = time.monotonic() + 20
    while time.monotonic() < deadline:
        try:
            probe = subprocess.run(
                [args.jcode, "--socket", str(socket_path), "debug", "server:info"],
                capture_output=True,
                text=True,
                env=environment,
                timeout=2,
            )
            if probe.returncode == 0:
                return process
        except subprocess.TimeoutExpired:
            pass
        if process.poll() not in (None, 0):
            details = log_path.read_text(encoding="utf-8", errors="replace")[-4000:]
            raise BenchmarkError(f"benchmark server exited early: {details}")
        time.sleep(0.2)
    terminate_process(process)
    details = log_path.read_text(encoding="utf-8", errors="replace")[-4000:]
    raise BenchmarkError(f"benchmark server did not become ready on {socket_path}: {details}")


def stop_server(args: argparse.Namespace, socket_path: Path, process: subprocess.Popen[str]) -> None: