mikhail/llama.cpp-mtp-turboquant
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

CI : support IOT device (IQ9) (#22987)

Browse Source 
* update test scripts

* align CI behavior between linux and android

* remove automatically cancel in 15min

* enable cancel-in-progress

* fix ty check issue

* update and fix pylint issue

* update runner such that we are not restricted by the 15min limit rule

* fix flake8 lint issue

* update runner according to review feedback

* code update according to review feedback

* switch from llama-cli to llama-completion binary with -no-cnv flag
This commit is contained in:
Zack Li
2026-05-14 13:58:34 -07:00
committed by GitHub
parent 834a243664
commit d81e63dcfd
7 changed files with 793 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/snapdragon/qdc/run_qdc_jobs.py
+368 -85
View File
@@ -1,4 +1,4 @@
"""Run llama.cpp Hexagon Android tests in a single QDC Appium job.
"""Run llama.cpp Hexagon tests in a single QDC job.
Bundles test scripts into one artifact and submits a single QDC job:
@@ -10,6 +10,10 @@ Results are written to $GITHUB_STEP_SUMMARY when set (GitHub Actions).
Prerequisites:
pip install /path/to/qualcomm_device_cloud_sdk*.whl
Platform is inferred from --device:
android Appium + pytest (Android phones: SM8750 / SM8650 / SM8850)
linux BASH (Linux IoT: QCS9075M)
Required environment variables:
QDC_API_KEY API key from QDC UI -> Users -> Settings -> API Keys
@@ -23,6 +27,7 @@ Usage:
from __future__ import annotations
import argparse
import enum
import logging
import os
import re
@@ -30,15 +35,35 @@ import shutil
import sys
import tempfile
import time
import urllib.request
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable
from qualcomm_device_cloud_sdk.api import qdc_api # ty: ignore[unresolved-import]
from qualcomm_device_cloud_sdk.logging import configure_logging # ty: ignore[unresolved-import]
from qualcomm_device_cloud_sdk.models import ArtifactType, JobMode, JobState, JobSubmissionParameter, JobType, TestFramework # ty: ignore[unresolved-import]
from qualcomm_device_cloud_sdk.api import qdc_api
from qualcomm_device_cloud_sdk.logging import configure_logging
from qualcomm_device_cloud_sdk.models import (
ArtifactType,
JobMode,
JobState,
JobSubmissionParameter,
JobType,
TestFramework,
)
# configure_logging only sets up the SDK logger; basicConfig is needed for
# our own log.info to reach stdout.
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(name)s %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()],
)
configure_logging(level=logging.INFO, handlers=[logging.StreamHandler()])
# Silence per-poll GET/status spam from the SDK and its HTTP client.
logging.getLogger("qualcomm_device_cloud").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
log = logging.getLogger(__name__)
POLL_INTERVAL = 30
@@ -47,23 +72,56 @@ LOG_UPLOAD_TIMEOUT = 600
CAPACITY_TIMEOUT = 1800
CAPACITY_POLL = 60
MAX_CONCURRENT_JOBS = 5
DEFAULT_RETRIES = 0
RETRY_DELAY = 300
TERMINAL_STATES = {JobState.COMPLETED, JobState.CANCELED}
NON_TERMINAL_STATES = {JobState.DISPATCHED, JobState.RUNNING, JobState.SETUP, JobState.SUBMITTED}
_SCRIPTS_DIR = Path(__file__).parent
_TESTS_DIR = _SCRIPTS_DIR / "tests"
_RUN_BENCH = _TESTS_DIR / "run_bench_tests_posix.py"
_RUN_BACKEND_OPS = _TESTS_DIR / "run_backend_ops_posix.py"
_UTILS = _TESTS_DIR / "utils.py"
_CONFTEST = _TESTS_DIR / "conftest.py"
_REQUIREMENTS = _SCRIPTS_DIR / "requirements.txt"
class DeviceUnavailableError(Exception):
"""Raised when the QDC device resource is not available (retryable)."""
_SCRIPTS_DIR = Path(__file__).parent
_TESTS_DIR = _SCRIPTS_DIR / "tests"
# --- Shared test assets -------------------------------------------------------
_UTILS = _TESTS_DIR / "utils.py"
_CONFTEST = _TESTS_DIR / "conftest.py"
_PYTEST_LINE_RE = re.compile(
r"(?:[\w/]+\.py::)?(?:\w+::)?([\w\[\].-]+)\s+(PASSED|FAILED|ERROR|SKIPPED)"
)
_EXCLUDED_LOGS = {"qdc_android_whole_host-000.log", "qdc_kernel_host-000.log"}
_EXCLUDED_LOGS = {
"qdc_android_whole_host-000.log",
"qdc_kernel_host-000.log",
"qdc_LE_whole_host-000.log",
"qdc_LE_kernel_host-000.log",
"script.log",
}
_NON_TERMINAL_STATE_VALUES = {s.value for s in NON_TERMINAL_STATES}
# --- Android (Appium + pytest) assets ----------------------------------------
_RUN_BENCH = _TESTS_DIR / "run_bench_tests_posix.py"
_RUN_BACKEND_OPS = _TESTS_DIR / "run_backend_ops_posix.py"
_REQUIREMENTS = _SCRIPTS_DIR / "requirements.txt"
_UPSTREAM_ADB_SCRIPTS = (
"https://raw.githubusercontent.com/ggml-org/llama.cpp/master/scripts/snapdragon/adb"
)
_ADB_SCRIPT_NAMES = [
"run-bench.sh",
"run-cli.sh",
"run-completion.sh",
"run-tool.sh",
]
# --- Linux (BASH) assets ------------------------------------------------------
_RUN_LINUX_TEMPLATE = _TESTS_DIR / "linux" / "run_linux.sh"
_LINUX_ENTRY_SCRIPT = "/bin/bash /data/local/tmp/TestContent/run_linux.sh"
# =============================================================================
# Artifact builders (per platform)
# =============================================================================
@dataclass
class JobResult:
@@ -73,35 +131,58 @@ class JobResult:
failure_details: dict[str, str] = field(default_factory=dict)
def build_artifact_zip(
def _write_lf(path: Path, content: str) -> None:
"""Write text with LF line endings (required by /bin/bash on Linux)."""
with open(path, "w", encoding="utf-8", newline="\n") as f:
f.write(content)
def _build_android_artifact(
pkg_dir: Path,
stage_dir: Path,
*,
test_mode: str = "bench",
model_url: str | None = None,
test_mode: str,
model_url: str | None,
) -> Path:
"""Bundle everything into a single QDC artifact zip.
"""Android zip (Appium/pytest). Extracted by QDC under /qdc/appium/.
Zip structure (extracted by QDC to /qdc/appium/ on the runner):
Zip structure:
llama_cpp_bundle/ installed package (adb pushed to /data/local/tmp/)
run-{bench,cli,completion,tool}.sh upstream adb wrappers (patched)
tests/
utils.py shared helpers (paths, run_adb_command, …)
conftest.py shared pytest fixtures (driver)
test_bench_posix.py bench + cli tests (<<MODEL_URL>> substituted)
AND/OR
test_backend_ops_posix.py test-backend-ops -b HTP0
utils.py shared adb helpers
conftest.py Appium pytest fixtures
test_bench_posix.py bench + cli tests (for --test bench or all)
test_backend_ops_posix.py test-backend-ops on HTP0
requirements.txt
pytest.ini addopts = --junitxml=results.xml
"""
shutil.copytree(pkg_dir, stage_dir / "llama_cpp_bundle")
bundle_dir = stage_dir / "llama_cpp_bundle"
shutil.copytree(pkg_dir, bundle_dir)
# Download upstream adb scripts so they land at /qdc/appium/ on the QDC
# runner. They wrap `adb shell` internally. Patch in `chmod +x bin/* lib/*`
# right after `cd $basedir` so device binaries are executable.
for name in _ADB_SCRIPT_NAMES:
url = f"{_UPSTREAM_ADB_SCRIPTS}/{name}"
dest = stage_dir / name
log.info("Downloading %s", url)
urllib.request.urlretrieve(url, str(dest))
content = dest.read_text()
content = content.replace(
"cd $basedir;",
"cd $basedir; chmod +x bin/* lib/* 2>/dev/null;",
)
dest.write_text(content)
dest.chmod(0o755)
tests_dir = stage_dir / "tests"
tests_dir.mkdir()
shutil.copy(_UTILS, tests_dir / "utils.py")
shutil.copy(_UTILS, tests_dir / "utils.py")
shutil.copy(_CONFTEST, tests_dir / "conftest.py")
if test_mode in ("bench", "all"):
assert model_url is not None, "--model-url is required for bench/all test modes"
assert model_url is not None
(tests_dir / "test_bench_posix.py").write_text(
_RUN_BENCH.read_text().replace("<<MODEL_URL>>", model_url)
)
@@ -109,33 +190,140 @@ def build_artifact_zip(
shutil.copy(_RUN_BACKEND_OPS, tests_dir / "test_backend_ops_posix.py")
shutil.copy(_REQUIREMENTS, stage_dir / "requirements.txt")
(stage_dir / "pytest.ini").write_text("[pytest]\naddopts = --junitxml=results.xml\n")
(stage_dir / "pytest.ini").write_text(
"[pytest]\naddopts = --junitxml=results.xml\n"
)
zip_base = str(stage_dir / "artifact")
shutil.make_archive(zip_base, "zip", stage_dir)
return Path(f"{zip_base}.zip")
def _build_linux_artifact(
pkg_dir: Path,
stage_dir: Path,
test_mode: str,
model_url: str | None,
) -> Path:
"""Linux IoT zip (BASH framework). Extracted by QDC to /data/local/tmp/TestContent/.
Zip structure:
run_linux.sh entry script (placeholder-substituted, LF line endings)
llama_cpp_bundle/ installed package
"""
bundle_dir = stage_dir / "llama_cpp_bundle"
shutil.copytree(pkg_dir, bundle_dir)
template = _RUN_LINUX_TEMPLATE.read_text(encoding="utf-8")
rendered = template.replace("{MODEL_URL}", model_url or "").replace(
"{TEST_MODE}", test_mode
)
script_path = stage_dir / "run_linux.sh"
_write_lf(script_path, rendered)
script_path.chmod(0o755)
zip_base = str(stage_dir / "artifact")
shutil.make_archive(zip_base, "zip", stage_dir)
return Path(f"{zip_base}.zip")
# =============================================================================
# Platform enum + strategy table
# =============================================================================
class Platform(enum.Enum):
ANDROID = "android"
LINUX = "linux"
@dataclass(frozen=True)
class PlatformSpec:
test_framework: TestFramework
entry_script: str | None
build_artifact: Callable[[Path, Path, str, str | None], Path]
job_name_fmt: str
PLATFORM_SPECS: dict[Platform, PlatformSpec] = {
Platform.ANDROID: PlatformSpec(
test_framework=TestFramework.APPIUM,
entry_script=None,
build_artifact=_build_android_artifact,
job_name_fmt="{base}",
),
Platform.LINUX: PlatformSpec(
test_framework=TestFramework.BASH,
entry_script=_LINUX_ENTRY_SCRIPT,
build_artifact=_build_linux_artifact,
job_name_fmt="{base} (Linux)",
),
}
DEVICE_PLATFORM: dict[str, Platform] = {
"SM8750": Platform.ANDROID,
"SM8650": Platform.ANDROID,
"SM8850": Platform.ANDROID,
"QCS9075M": Platform.LINUX,
}
# =============================================================================
# Shared QDC job plumbing
# =============================================================================
def wait_for_job(client, job_id: str, timeout: int) -> str:
elapsed = 0
last_state = None
consecutive_errors = 0
max_consecutive_errors = 5
while elapsed < timeout:
raw = qdc_api.get_job_status(client, job_id)
try:
raw = qdc_api.get_job_status(client, job_id)
consecutive_errors = 0
except Exception as e:
consecutive_errors += 1
log.warning(
"Transient error polling job %s (%d/%d): %s",
job_id,
consecutive_errors,
max_consecutive_errors,
e,
)
if consecutive_errors >= max_consecutive_errors:
raise
time.sleep(POLL_INTERVAL)
elapsed += POLL_INTERVAL
continue
try:
status = JobState(raw)
except ValueError:
status = raw
if status in TERMINAL_STATES:
return raw.lower()
log.info("Job %s: %s", job_id, raw)
if raw != last_state:
log.info("Job %s: %s", job_id, raw)
last_state = raw
time.sleep(POLL_INTERVAL)
elapsed += POLL_INTERVAL
# Abort to free the QDC concurrency slot instead of leaking it.
try:
qdc_api.abort_job(client, job_id)
log.warning("Aborted job %s after timeout to free concurrency slot", job_id)
except Exception as e:
log.warning("Failed to abort job %s: %s", job_id, e)
raise TimeoutError(f"Job {job_id} did not finish within {timeout}s")
def wait_for_log_upload(client, job_id: str) -> None:
elapsed = 0
while elapsed <= LOG_UPLOAD_TIMEOUT:
status = (qdc_api.get_job_log_upload_status(client, job_id) or "").lower()
try:
status = (qdc_api.get_job_log_upload_status(client, job_id) or "").lower()
except Exception as e:
log.warning("get_job_log_upload_status failed: %s — will retry", e)
status = ""
if status in {"completed", "failed"}:
return
log.info("Waiting for log upload (status=%s) ...", status)
@@ -150,17 +338,33 @@ def wait_for_capacity(client, max_jobs: int = MAX_CONCURRENT_JOBS) -> None:
while elapsed < CAPACITY_TIMEOUT:
jobs_page = qdc_api.get_jobs_list(client, page_number=0, page_size=50)
if jobs_page is None:
log.warning("Could not retrieve job list; proceeding without capacity check")
log.warning(
"Could not retrieve job list; proceeding without capacity check"
)
return
items = getattr(jobs_page, "data", []) or []
active = sum(1 for j in items if getattr(j, "state", None) in _NON_TERMINAL_STATE_VALUES)
active = sum(
1 for j in items if getattr(j, "state", None) in _NON_TERMINAL_STATE_VALUES
)
if active < max_jobs:
log.info("Active QDC jobs: %d / %d — proceeding", active, max_jobs)
return
log.info("Active QDC jobs: %d / %d — waiting %ds ...", active, max_jobs, CAPACITY_POLL)
log.info(
"Active QDC jobs: %d / %d — waiting %ds ...",
active,
max_jobs,
CAPACITY_POLL,
)
time.sleep(CAPACITY_POLL)
elapsed += CAPACITY_POLL
log.warning("Capacity wait timed out after %ds; proceeding anyway", CAPACITY_TIMEOUT)
raise TimeoutError(
f"Capacity wait timed out after {CAPACITY_TIMEOUT}s"
)
# ---------------------------------------------------------------------------
# Log parsing helpers
# ---------------------------------------------------------------------------
def _parse_junit_xml(content: str) -> tuple[dict[str, bool], dict[str, str]]:
@@ -192,10 +396,26 @@ def _parse_pytest_output(content: str) -> dict[str, bool]:
def fetch_logs_and_parse_tests(
client, job_id: str
client, job_id: str, max_retries: int = 5, retry_delay: int = 30
) -> tuple[dict[str, bool], dict[str, str], dict[str, str]]:
"""Returns (test_results, raw_logs, failure_details)."""
log_files = qdc_api.get_job_log_files(client, job_id)
log_files = None
for attempt in range(1, max_retries + 1):
try:
log_files = qdc_api.get_job_log_files(client, job_id)
break
except Exception as e:
if attempt < max_retries:
log.warning(
"get_job_log_files failed (attempt %d/%d): %s — retrying in %ds",
attempt, max_retries, e, retry_delay,
)
time.sleep(retry_delay)
else:
log.error(
"get_job_log_files failed after %d attempts: %s", max_retries, e
)
return {}, {}, {}
if not log_files:
log.warning("No log files returned for job %s", job_id)
return {}, {}, {}
@@ -207,8 +427,8 @@ def fetch_logs_and_parse_tests(
with tempfile.TemporaryDirectory() as tmpdir:
for lf in log_files:
log.info("Downloading log file: %s", lf.filename)
zip_path = os.path.join(tmpdir, "log.zip")
log.info("Downloading log file: %s", lf.filename)
qdc_api.download_job_log_files(client, lf.filename, zip_path)
try:
shutil.unpack_archive(zip_path, tmpdir, "zip")
@@ -226,12 +446,15 @@ def fetch_logs_and_parse_tests(
elif fname.endswith(".log"):
if fname in _EXCLUDED_LOGS:
continue
log.info("--- %s ---", fname)
log.info("%s", content)
log.info("--- %s ---\n%s", fname, content)
raw_logs[fname] = content
pytest_fallback.update(_parse_pytest_output(content))
return (test_results if test_results else pytest_fallback), raw_logs, failure_details
return (
(test_results if test_results else pytest_fallback),
raw_logs,
failure_details,
)
def write_summary(result: JobResult, title: str = "QDC Test Results") -> None:
@@ -289,30 +512,106 @@ def write_summary(result: JobResult, title: str = "QDC Test Results") -> None:
f.write("\n".join(lines) + "\n")
# =============================================================================
# CLI + main
# =============================================================================
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument("--pkg-dir", required=True, type=Path,
p.add_argument("--pkg-dir", required=True, type=Path,
help="Installed llama.cpp package directory (contains bin/ and lib/)")
p.add_argument("--model-url",
help="Direct URL to the GGUF model file (required for --test bench)")
p.add_argument("--device", required=True,
p.add_argument("--device", required=True,
help="QDC chipset name, e.g. SM8750")
p.add_argument("--test", choices=["bench", "backend-ops", "all"], default="bench",
help="Test suite to run (default: bench)")
p.add_argument("--job-timeout", type=int, default=JOB_TIMEOUT, metavar="SECONDS",
help=f"Max seconds to wait for job completion (default: {JOB_TIMEOUT})")
p.add_argument("--retries", type=int, default=DEFAULT_RETRIES, metavar="N",
help="Number of retries when device is unavailable (default: 0)")
p.add_argument("--retry-delay", type=int, default=RETRY_DELAY, metavar="SECONDS",
help=f"Seconds to wait between retries (default: {RETRY_DELAY})")
args = p.parse_args()
if args.test in ("bench", "all") and not args.model_url:
p.error("--model-url is required when --test bench or --test all")
return args
def _submit_and_run_job(client, args, spec, target_id, artifact_id) -> JobResult:
"""Submit a QDC job and wait for results.
Raises DeviceUnavailableError for transient device/resource issues that
are worth retrying. Returns JobResult for definitive outcomes (pass or
test failure).
"""
try:
wait_for_capacity(client)
except TimeoutError:
raise DeviceUnavailableError("Capacity wait timed out — device busy")
job_name = spec.job_name_fmt.format(base="llama.cpp Hexagon tests")
job_id = qdc_api.submit_job(
public_api_client=client,
target_id=target_id,
job_name=job_name,
external_job_id=None,
job_type=JobType.AUTOMATED,
job_mode=JobMode.APPLICATION,
timeout=max(1, args.job_timeout // 60),
test_framework=spec.test_framework,
entry_script=spec.entry_script,
job_artifacts=[artifact_id],
monkey_events=None,
monkey_session_timeout=None,
job_parameters=[JobSubmissionParameter.WIFIENABLED],
)
if job_id is None:
raise DeviceUnavailableError("Job submission failed — device may be unavailable")
log.info("Job submitted: %s (device=%s)", job_id, args.device)
try:
job_status = wait_for_job(client, job_id, timeout=args.job_timeout)
except TimeoutError as e:
raise DeviceUnavailableError(str(e))
log.info("Job %s finished: %s", job_id, job_status)
wait_for_log_upload(client, job_id)
tests, raw_logs, failure_details = fetch_logs_and_parse_tests(client, job_id)
job_ok = job_status == JobState.COMPLETED.value.lower()
if not job_ok and not tests:
raise DeviceUnavailableError(
f"Job did not complete (status={job_status}) and produced no test results"
)
passed = job_ok and all(tests.values()) if tests else job_ok
if spec.test_framework == TestFramework.BASH and not tests:
log.error("No test results recovered (state=%s). Script likely never ran.", job_status)
passed = False
if not passed:
log.error("Job did not complete successfully or tests failed (status=%s)", job_status)
return JobResult(passed=passed, tests=tests, raw_logs=raw_logs, failure_details=failure_details)
def main() -> int:
args = parse_args()
platform = DEVICE_PLATFORM.get(args.device)
if platform is None:
log.error(
"Unknown device %r. Known: %s",
args.device, ", ".join(sorted(DEVICE_PLATFORM.keys())),
)
return 1
spec = PLATFORM_SPECS[platform]
api_key = os.environ.get("QDC_API_KEY")
if not api_key:
log.error("QDC_API_KEY environment variable must be set")
@@ -334,10 +633,9 @@ def main() -> int:
return 1
with tempfile.TemporaryDirectory() as tmpdir:
log.info("Building artifact ...")
zip_path = build_artifact_zip(
args.pkg_dir, Path(tmpdir),
test_mode=args.test, model_url=args.model_url,
log.info("Building %s artifact (test=%s) ...", platform.value, args.test)
zip_path = spec.build_artifact(
args.pkg_dir, Path(tmpdir), args.test, args.model_url
)
log.info("Uploading artifact (%d MB) ...", zip_path.stat().st_size // 1_000_000)
artifact_id = qdc_api.upload_file(client, str(zip_path), ArtifactType.TESTSCRIPT)
@@ -346,46 +644,31 @@ def main() -> int:
log.error("Artifact upload failed")
return 1
wait_for_capacity(client)
job_id = qdc_api.submit_job(
public_api_client=client,
target_id=target_id,
job_name="llama.cpp Hexagon tests",
external_job_id=None,
job_type=JobType.AUTOMATED,
job_mode=JobMode.APPLICATION,
timeout=max(1, args.job_timeout // 60),
test_framework=TestFramework.APPIUM,
entry_script=None,
job_artifacts=[artifact_id],
monkey_events=None,
monkey_session_timeout=None,
job_parameters=[JobSubmissionParameter.WIFIENABLED],
)
if job_id is None:
log.error("Job submission failed")
max_attempts = 1 + args.retries
for attempt in range(1, max_attempts + 1):
try:
result = _submit_and_run_job(client, args, spec, target_id, artifact_id)
break
except DeviceUnavailableError as e:
if attempt < max_attempts:
log.warning(
"Attempt %d/%d failed (device unavailable): %s — retrying in %ds",
attempt, max_attempts, e, args.retry_delay,
)
time.sleep(args.retry_delay)
else:
log.error(
"Attempt %d/%d failed (device unavailable): %s — no retries left",
attempt, max_attempts, e,
)
write_summary(
JobResult(passed=False, tests={}),
title=f"QDC Device Unavailable ({args.device})",
)
return 1
else:
return 1
log.info("Job submitted: %s (device=%s)", job_id, args.device)
try:
job_status = wait_for_job(client, job_id, timeout=args.job_timeout)
except TimeoutError as e:
log.error("%s", e)
write_summary(JobResult(passed=False, tests={}), title=f"QDC Job Timed Out ({args.device})")
return 1
log.info("Job %s finished: %s", job_id, job_status)
wait_for_log_upload(client, job_id)
tests, raw_logs, failure_details = fetch_logs_and_parse_tests(client, job_id)
passed = job_status == JobState.COMPLETED.value.lower()
if tests:
passed = passed and all(tests.values())
if not passed:
log.error("Job did not complete successfully or tests failed (status=%s)", job_status)
result = JobResult(passed=passed, tests=tests, raw_logs=raw_logs, failure_details=failure_details)
if args.test == "backend-ops":
title = f"Backend Ops — HTP0 ({args.device})"
elif args.test == "all":
@@ -394,7 +677,7 @@ def main() -> int:
title = f"QDC Test Results ({args.device})"
write_summary(result, title=title)
return 0 if passed else 1
return 0 if result.passed else 1
if __name__ == "__main__":