riscv-arch-test/run_tests.py at act4 · riscv/riscv-arch-test

History

executable file

340 lines (283 loc) · 13.5 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

#!/usr/bin/env python3

##################################

# run_tests.py

# jcarlin@hmc.edu Jan 2026

# SPDX-License-Identifier: Apache-2.0

# Run all ELFs from a directory using the input command in parallel

##################################

import argparse

import os

import re

import shlex

import signal

import subprocess

import sys

from functools import partial

from multiprocessing import Pool

from pathlib import Path

_SUMMARY_RE = re.compile(r'RVCP-SUMMARY: TEST (PASSED|FAILED|SIGRUN) - Test File ".*"')

_DEBUG_PLACEHOLDER_RE = re.compile(r"\{debug:([^}]*)\}")

_TRACEFILE_PLACEHOLDER = "__TRACEFILE__"

_SUMMARYFILE_PLACEHOLDER = "__SUMMARYFILE__"

# Simulator-reported failure reasons worth surfacing verbatim in FAIL messages, e.g.

# whisper's "Error: Failed stop: Hart 0: Core entered critical-error state ..." or

# qemu's "qemu: fatal: M-mode double trap"

_ERROR_LINE_RE = re.compile(r"Error:|Failed stop:|critical[-_ ]error|fatal", re.IGNORECASE)

# ANSI color codes — disabled when stdout is not a terminal

USE_COLOR = sys.stdout.isatty()

def _expand_debug_placeholders(command: str, *, debug: bool) -> str:

"""Expand {debug:...} placeholders in a command string.

When debug is True, {debug:--flag1 --flag2} expands to '--flag1 --flag2'.

When debug is False, the placeholder is removed entirely.

"""

result = _DEBUG_PLACEHOLDER_RE.sub(lambda m: m.group(1) if debug else "", command)

return " ".join(result.split())

def _color(code: str, text: str) -> str:

if USE_COLOR:

return f"\033[{code}m{text}\033[0m"

return text

def red(text: str) -> str:

return _color("1;31", text)

def green(text: str) -> str:

return _color("1;32", text)

def bold(text: str) -> str:

return _color("1", text)

def yellow(text: str) -> str:

return _color("1;33", text)

def bold_cyan(text: str) -> str:

return _color("1;36", text)

def dim(text: str) -> str:

return _color("2", text)

def _simulator_error_lines(log_file: Path, limit: int = 3) -> list[str]:

"""Extract simulator-reported error lines from a run log (skipping the 2-line header)."""

try:

lines = log_file.read_text(errors="replace").splitlines()

except OSError:

return []

# "Monitoring net ...critical_error" is imperas's startup registration line, not an error

return [line.strip() for line in lines[2:] if _ERROR_LINE_RE.search(line) and "Monitoring net" not in line][:limit]

def run_test(

command: str, log_dir: Path, elf_dir: Path, elf_path: Path, verbose: bool, timeout: int

) -> tuple[bool, Path, str, str]:

"""Run a single ELF and return (failed, elf_path, rvcp_summary_line, fail_message)."""

# Create log, trace, and summary file paths mirroring the ELF subdirectory hierarchy

rel = elf_path.relative_to(elf_dir)

log_file = log_dir / rel.with_suffix(".log")

trace_file = log_dir / rel.with_suffix(".trace.log")

summary_file = log_dir / rel.with_suffix(".summary.log")

log_file.parent.mkdir(parents=True, exist_ok=True)

# Substitute __TRACEFILE__ / __SUMMARYFILE__ placeholders with per-test paths.

# __TRACEFILE__: directs simulator trace output to a separate file so it doesn't

# interleave with RVCP-SUMMARY lines in the main log.

# __SUMMARYFILE__: for simulators that can redirect console/UART output (containing

# RVCP-SUMMARY) to a file but cannot redirect trace output. When present,

# run_tests reads RVCP-SUMMARY from this file instead of the main log.

has_summary_file = _SUMMARYFILE_PLACEHOLDER in command

# Split command first, then substitute placeholders at token level so paths

# containing spaces remain a single argument.

tokens = shlex.split(command)

tokens = [

tok.replace(_TRACEFILE_PLACEHOLDER, str(trace_file)).replace(_SUMMARYFILE_PLACEHOLDER, str(summary_file))

for tok in tokens

]

# Build display command from the substituted tokens

test_command = shlex.join(tokens)

# Extract leading KEY=VALUE env var assignments

env_overrides: dict[str, str] = {}

while tokens and "=" in tokens[0] and not tokens[0].startswith("-"):

key, _, value = tokens.pop(0).partition("=")

env_overrides[key] = value

env = {**os.environ, **env_overrides} if env_overrides else None

cmd = tokens

full_cmd = [*cmd, str(elf_path)]

display_cmd = f"{test_command} {elf_path}"

# Safe to print from the worker: --verbose forces --jobs 1, so nothing races with it.

if verbose:

print(f"\nRunning {display_cmd}", flush=True)

timed_out = False

with log_file.open("w") as f:

print("This log file generated by running:", file=f)

print(f"{display_cmd}\n", file=f)

f.flush()

proc = subprocess.Popen(

full_cmd,

stdin=subprocess.DEVNULL,

stdout=f,

stderr=subprocess.STDOUT,

env=env,

start_new_session=True, # new process group so killpg reaches QEMU children

)

try:

proc.wait(timeout=timeout)

except subprocess.TimeoutExpired:

timed_out = True

# SIGTERM first so simulators with graceful handlers (e.g. imperas

# --stoponcontrolc) can flush trace output; escalate if they don't exit.

try:

os.killpg(os.getpgid(proc.pid), signal.SIGTERM)

except (ProcessLookupError, PermissionError):

proc.terminate()

try:

proc.wait(timeout=5)

except subprocess.TimeoutExpired:

try:

os.killpg(os.getpgid(proc.pid), signal.SIGKILL)

except (ProcessLookupError, PermissionError):

proc.kill()

proc.wait()

returncode = proc.returncode

# Build trace/summary file references for failure output

trace_msg = f"\n Trace: {dim(str(trace_file))}" if trace_file.exists() else ""

summary_msg = f"\n Summary: {dim(str(summary_file))}" if has_summary_file else ""

error_msg = "".join(f"\n {dim(line)}" for line in _simulator_error_lines(log_file))

if timed_out:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — timed out after {timeout}s, simulator process group killed"

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

return True, elf_path, f"TIMEOUT after {timeout}s", message

# Check exit code

exit_failed = returncode != 0

# Check for RVCP-SUMMARY lines. When __SUMMARYFILE__ was used, read from the

# dedicated summary file (simulator UART output) instead of the main log.

if has_summary_file and summary_file.exists():

summary_text = summary_file.read_text(errors="replace")

else:

summary_text = log_file.read_text(errors="replace")

summaries = _SUMMARY_RE.findall(summary_text)

rvcp_lines = [line for line in summary_text.splitlines() if "RVCP-SUMMARY:" in line]

rvcp_summary = rvcp_lines[0] if rvcp_lines else "No RVCP-SUMMARY line found"

summary_failed = "FAILED" in summaries

summary_sigrun = "SIGRUN" in summaries

no_summary = len(summaries) == 0

# Overall failure for test

failed = exit_failed or summary_failed or summary_sigrun or no_summary

# Build failure message for test

message = ""

if summary_sigrun:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — RVCP-SUMMARY reports SIGRUN"

f"\n ELF was not built with RVTEST_SELFCHECK enabled (non-selfchecking test)."

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

elif exit_failed and no_summary:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — exit code {returncode} indicates failure, no RVCP-SUMMARY line found"

f"\n Likely abnormal termination (killed, crash, timeout) or bug in RVMODEL_IO_WRITE macro."

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

elif exit_failed and summary_failed:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — exit code {returncode}"

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

elif summary_failed and not exit_failed:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — RVCP-SUMMARY: TEST FAILED but exit code {returncode} indicates success"

f"\n If this is an ImperasFPM test, it is due to ImperasFPM not yet supporting failure exit code. Otherwise likely bug in RVMODEL_HALT_FAIL macro."

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

elif exit_failed and not summary_failed:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — RVCP-SUMMARY: TEST PASSED but exit code {returncode} indicates failure"

f"\n Likely bug in RVMODEL_HALT_PASS macro."

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

elif no_summary and not exit_failed:

message = (

f" {red('FAIL')} {bold(elf_path.name)} — exit code 0 but no RVCP-SUMMARY line found"

f"\n Test may have been killed externally or hung without producing output."

f"\n Log: {dim(str(log_file))}{trace_msg}{summary_msg}{error_msg}"

)

return failed, elf_path, rvcp_summary, message

def main() -> int:

parser = argparse.ArgumentParser(description="Run all ELF files using the provided command")

parser.add_argument(

"command", type=str, help="Command to run each ELF (elf path will be appended). E.g., 'spike --isa=rv64gc'"

)

parser.add_argument("elf_dir", type=Path, help="Path to ELF directory (e.g., work/spike-rv64/elfs)")

parser.add_argument("-j", "--jobs", type=int, default=os.cpu_count(), help="Number of parallel jobs")

parser.add_argument(

"-v",

"--verbose",

action="store_true",

help="Print the full command for every ELF, implies --debug, serializes to 1 job",

)

parser.add_argument(

"-d", "--debug", action="store_true", help="Enable debug mode: expand {debug:...} placeholders in the command"

)

parser.add_argument(

"--timeout",

type=int,

default=5 * 60,

metavar="SECONDS",

help="Per-test timeout in seconds (default: 300). Kills the entire simulator process group on expiry.",

)

args = parser.parse_args()

if args.timeout <= 0:

parser.error("--timeout must be a positive integer")

# Verbose implies debug and serialized execution

if args.verbose:

args.debug = True

args.jobs = 1

# Expand {debug:...} placeholders based on --debug flag

command = _expand_debug_placeholders(args.command, debug=args.debug)

# Set up directories

elf_dir = args.elf_dir.resolve()

log_dir = elf_dir.parent / "logs"

log_dir.mkdir(parents=True, exist_ok=True)

summary_log = elf_dir.parent / "summary.log"

# Derive config name from elf_dir (e.g., work/<config-name>/elfs -> config-name)

config_name = elf_dir.parent.name

# Print banner for this config

banner = f"══════ {config_name} ══════"

print(f"\n{bold_cyan(banner)}")

print(f" {bold('Running ELFs from')}: {elf_dir}")

banner_command = command.replace(_TRACEFILE_PLACEHOLDER, "<trace_file>")

banner_command = banner_command.replace(_SUMMARYFILE_PLACEHOLDER, "<summary_file>")

print(f" {bold('Using command')}: {banner_command} <elf_path>")

print(f" {bold('Overall summary available at')}: {summary_log}")

if _TRACEFILE_PLACEHOLDER in command:

print(f" {bold('Trace output')}: {log_dir}/<test>.trace.log")

if _SUMMARYFILE_PLACEHOLDER in command:

print(f" {bold('RVCP summary source')}: {log_dir}/<test>.summary.log")

# Find all ELFs

elf_files = sorted(elf_dir.rglob("*.elf"))

if not elf_files:

print(yellow(" No ELF files found"))

sys.exit(0)

partial_run_test = partial(run_test, command, log_dir, elf_dir, verbose=args.verbose, timeout=args.timeout)

failed = 0

entries: list[tuple[str, str]] = []

# Run individual tests

with Pool(args.jobs) as pool, summary_log.open("w") as f:

for fail_status, elf_path, rvcp_summary, fail_message in pool.imap_unordered(partial_run_test, elf_files):

if fail_status:

failed += 1

# Printed here, in the parent, so concurrent workers cannot interleave with

# each other mid-message (see run_test's docstring).

if fail_message:

print(fail_message, flush=True)

rel_log = str(elf_path.relative_to(elf_dir).with_suffix(".log"))

entries.append((rel_log, rvcp_summary))

print(f"{rel_log} {rvcp_summary}", file=f, flush=True)

pool.close()

pool.join()

# Rewrite the top-level summary log sorted and column-aligned now that all results are in.

entries.sort()

col_width = max((len(p) for p, _ in entries), default=0)

with summary_log.open("w") as f:

for rel_log, rvcp_summary in entries:

print(f"{rel_log:<{col_width}} {rvcp_summary}", file=f)

# Print overall results

passed = len(elf_files) - failed

print()

if failed:

print(red(f" RESULT: {failed} failed, {passed} passed out of {len(elf_files)} tests."))

else:

print(green(f" RESULT: All {len(elf_files)} tests passed."))

return 1 if failed else 0

if __name__ == "__main__":

sys.exit(main())

FilesExpand file tree

run_tests.py

Latest commit

History

run_tests.py

File metadata and controls