From 3fc72585f252492a5058d3976e49a42062360afc Mon Sep 17 00:00:00 2001 From: cangtianhuang <1903374751@qq.com> Date: Thu, 16 Jul 2026 13:53:23 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E2=9A=A1=EF=B8=8F=20Classify=20comp=20outp?= =?UTF-8?q?ut,=20refactor=20log=20writer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- engineV2.py | 16 +- engineV4.py | 22 +- tester/accuracy_stable.py | 120 ++- tester/api_config/log_writer.py | 873 +++++++++++---------- tools/error_stat/error_stat.py | 54 +- tools/prof/paddleapitest_matmul_heatmap.py | 3 +- 6 files changed, 616 insertions(+), 472 deletions(-) diff --git a/engineV2.py b/engineV2.py index ddc3f751..63a10b96 100644 --- a/engineV2.py +++ b/engineV2.py @@ -493,9 +493,7 @@ def check_gpu_memory(gpu_ids, num_workers_per_gpu, required_memory): # required def init_worker_gpu(gpu_worker_list, lock, available_gpus, max_workers_per_gpu, options): - if options.log_dir: - set_test_log_path(options.log_dir) - set_engineV2() + init_log(options.log_dir, worker_tmp_logs=True) my_pid = os.getpid() def pid_exists(pid): @@ -985,8 +983,6 @@ def main(): if options.bitwise_alignment: options.atol = 0.0 options.rtol = 0.0 - if options.log_dir: - set_test_log_path(options.log_dir) if options.api_config: try: @@ -1008,8 +1004,7 @@ def main(): globals().update(_load_test_classes(options)) - # set log_writer - set_engineV2() + init_log(options.log_dir, worker_tmp_logs=True) options.api_config = options.api_config.strip() print( @@ -1097,10 +1092,7 @@ def main(): return config_files = [options.api_config_file] - # set log_writer before resume/checkpoint handling - if options.log_dir: - set_test_log_path(options.log_dir) - set_engineV2() + init_log(options.log_dir, worker_tmp_logs=True) # when engineV2 was interrupted, resume from .tmp dir aggregate_logs(cleanup=True) @@ -1247,7 +1239,7 @@ def cleanup_handler(*args): elif err.exitcode in (-signal.SIGKILL, -signal.SIGTERM): checkpoint_ready = False print( - f"[warn] Worker was externally killed for {config} " + f"[warn] Worker was externally killed " f"(exit={err.exitcode}); case will be retried on next run.", flush=True, ) diff --git a/engineV4.py b/engineV4.py index 0eeb3086..e5097016 100644 --- a/engineV4.py +++ b/engineV4.py @@ -135,9 +135,7 @@ class WorkerSlot: def _init_worker_runtime(slot_index, gpu_id, options, *, redirect_output): - if options.log_dir: - set_test_log_path(options.log_dir) - set_engineV2() + init_log(options.log_dir, worker_tmp_logs=True) if gpu_id is not None: os.environ["CUDA_VISIBLE_DEVICES"] = str(gpu_id) @@ -258,9 +256,7 @@ def _build_sanitizer_case_command(api_config_str, options, log_dir, sanitizer_cm def _sanitizer_worker_loop(slot_index, gpu_id, input_queue, result_queue, options): - if options.log_dir: - set_test_log_path(options.log_dir) - set_engineV2() + init_log(options.log_dir, worker_tmp_logs=True) redirect_stdio() child_process = None @@ -1672,8 +1668,6 @@ def main(): if options.bitwise_alignment: options.atol = 0.0 options.rtol = 0.0 - if options.log_dir: - set_test_log_path(options.log_dir) if options._sanitizer_child: try: @@ -1714,8 +1708,7 @@ def main(): globals().update(_load_test_classes(options)) - # set log_writer - set_engineV2() + init_log(options.log_dir, worker_tmp_logs=True) options.api_config = options.api_config.strip() print( @@ -1803,6 +1796,8 @@ def main(): return config_files = [options.api_config_file] + init_log(options.log_dir, worker_tmp_logs=True) + # when engineV2 was interrupted, resume from .tmp dir aggregate_logs(cleanup=True) if options.use_compute_sanitizer: @@ -1878,11 +1873,6 @@ def main(): if options.test_cpu: print(f"Using {cpu_count()} CPU(s) for paddle in CPU mode.", flush=True) - # set log_writer - if options.log_dir: - set_test_log_path(options.log_dir) - set_engineV2() - # initialize worker pool (per-worker queue architecture) pool = WorkerPool(available_gpus, max_workers_per_gpu, options) @@ -1964,7 +1954,7 @@ def cleanup_handler(*args): if external_kill: print( - f"[warn] Worker was externally killed for {config} " + f"[warn] Worker was externally killed " f"(exit={exitcode}); case will be retried on next run.", flush=True, ) diff --git a/tester/accuracy_stable.py b/tester/accuracy_stable.py index 21c72190..3356edd8 100644 --- a/tester/accuracy_stable.py +++ b/tester/accuracy_stable.py @@ -8,12 +8,30 @@ import torch from .accuracy import process_grad_output, process_output -from .api_config.log_writer import has_terminal_log, log_accuracy_stable, write_to_log +from .api_config.log_writer import ( + ALL_DIMENSIONS, + COMP_TO_DIMENSION, + has_comp_terminal_log, + has_terminal_log, + log_accuracy_stable, + write_to_comp_log, + write_to_log, +) from .base import CUDA_ERROR, CUDA_OOM, APITestBase from .paddle_to_torch import get_converter class APITestAccuracyStable(APITestBase): + # 执行阶段错误广播映射: (iter_idx, source) -> 受影响的 comp 列表 + _TORCH_AFFECTED_COMPS = { + 0: ["T1P1", "T1P2", "T1T2", "T1P1B", "T1P2B", "T1T2B"], + 1: ["T2P2", "T2P1", "T1T2", "T2P2B", "T2P1B", "T1T2B"], + } + _PADDLE_AFFECTED_COMPS = { + 0: ["T1P1", "T2P1", "P1P2", "T1P1B", "T2P1B", "P1P2B"], + 1: ["T2P2", "T1P2", "P1P2", "T2P2B", "T1P2B", "P1P2B"], + } + def __init__(self, api_config, **kwargs): super().__init__(api_config) self.test_amp = kwargs.get("test_amp", False) @@ -21,6 +39,11 @@ def __init__(self, api_config, **kwargs): torch.set_printoptions(profile="short", edgeitems=2, threshold=100, linewidth=120) torch.set_default_device("cuda") + def _broadcast_to_comp_dimensions(self, log_type, affected_comps): + """将执行阶段错误广播到所有受影响的 comp 维度""" + for comp in affected_comps: + write_to_comp_log(comp, log_type, self.api_config.config) + def _reset_random_state(self, seed: int = 42): """Reset numpy / paddle / torch (CPU+CUDA) RNGs so random APIs (uniform, normal, randn, bernoulli, dropout, ...) produce @@ -99,7 +122,7 @@ def test(self): # ======== torch ======== self._reset_random_state() torch_output, torch_out_grads, torch_grad_success = self.get_torch_output( - convert_result + convert_result, _i ) if torch_output is None: return @@ -107,7 +130,7 @@ def test(self): # ======== paddle ======== self._reset_random_state() - paddle_output, paddle_out_grads = self.get_paddle_output(torch_grad_success) + paddle_output, paddle_out_grads = self.get_paddle_output(torch_grad_success, _i) if paddle_output is None: return paddle.device.cuda.empty_cache() @@ -141,11 +164,18 @@ def test(self): self.compare(paddle_output_pair[0], paddle_output_pair[1], "P1P2") self.compare(paddle_grad_pair[0], paddle_grad_pair[1], "P1P2B") + # 逐维度写 pass + for dimension in ALL_DIMENSIONS: + if not has_comp_terminal_log(dimension, self.api_config.config): + # 取该维度的任一 comp 代表写 pass + rep_comp = next(c for c, d in COMP_TO_DIMENSION.items() if d == dimension) + write_to_comp_log(rep_comp, "pass", self.api_config.config) + # 主日志 pass（让 engine 的 has_terminal_log 能正确判断） if not has_terminal_log(self.api_config.config): print(f"[pass] {self.api_config.config}", flush=True) write_to_log("pass", self.api_config.config) - def get_torch_output(self, convert_result): + def get_torch_output(self, convert_result, iter_idx=0): # ======== run torch forward ========: torch_output = None try: @@ -183,11 +213,11 @@ def get_torch_output(self, convert_result): err_str = str(err) if any(cuda_err in err_str for cuda_err in CUDA_OOM): print(f"[oom] {self.api_config.config}\n{err_str}", flush=True) - write_to_log("oom", self.api_config.config) + self._broadcast_to_comp_dimensions("oom", self._TORCH_AFFECTED_COMPS[iter_idx]) raise print(f"[torch_error] {self.api_config.config}\n{err_str}", flush=True) traceback.print_exc() - write_to_log("torch_error", self.api_config.config) + self._broadcast_to_comp_dimensions("torch_error", self._TORCH_AFFECTED_COMPS[iter_idx]) if any(cuda_err in err_str for cuda_err in CUDA_ERROR): raise return None, None, None @@ -215,21 +245,25 @@ def get_torch_output(self, convert_result): f"[config_input] {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("config_input", self.api_config.config) + self._broadcast_to_comp_dimensions( + "config_input", self._TORCH_AFFECTED_COMPS[iter_idx] + ) return None, None, None if any(cuda_err in err_str for cuda_err in CUDA_OOM): print( f"[oom] phase=backward {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("oom", self.api_config.config) + self._broadcast_to_comp_dimensions("oom", self._TORCH_AFFECTED_COMPS[iter_idx]) raise if any(cuda_err in err_str for cuda_err in CUDA_ERROR): print( f"[torch_error] phase=backward {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("torch_error", self.api_config.config) + self._broadcast_to_comp_dimensions( + "torch_error", self._TORCH_AFFECTED_COMPS[iter_idx] + ) raise print(err_str, flush=True) @@ -242,7 +276,9 @@ def get_torch_output(self, convert_result): flush=True, ) traceback.print_exc() - write_to_log("torch_error", self.api_config.config) + self._broadcast_to_comp_dimensions( + "torch_error", self._TORCH_AFFECTED_COMPS[iter_idx] + ) raise def process_torch_outputs(obj): @@ -256,7 +292,7 @@ def process_torch_outputs(obj): torch_out_grads = process_torch_outputs(torch_out_grads) return torch_output, torch_out_grads, torch_grad_success - def get_paddle_output(self, torch_grad_success): + def get_paddle_output(self, torch_grad_success, iter_idx=0): # ======== run paddle forward ======== paddle_output = None try: @@ -307,26 +343,30 @@ def get_paddle_output(self, torch_grad_success): err_str = str(err) if self.should_ignore_paddle_error(err_str): print(f"[pass] {self.api_config.config}", flush=True) - write_to_log("pass", self.api_config.config) + self._broadcast_to_comp_dimensions("pass", self._PADDLE_AFFECTED_COMPS[iter_idx]) return None, None if any(cuda_err in err_str for cuda_err in CUDA_ERROR): print(f"[paddle_cuda] {self.api_config.config}\n{err_str}", flush=True) - write_to_log("paddle_cuda", self.api_config.config) + self._broadcast_to_comp_dimensions( + "paddle_cuda", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) raise if any(cuda_err in err_str for cuda_err in CUDA_OOM): print(f"[oom] {self.api_config.config}\n{err_str}", flush=True) - write_to_log("oom", self.api_config.config) + self._broadcast_to_comp_dimensions("oom", self._PADDLE_AFFECTED_COMPS[iter_idx]) raise print(f"[paddle_error] {self.api_config.config}\n{err_str}", flush=True) traceback.print_exc() - write_to_log("paddle_error", self.api_config.config) + self._broadcast_to_comp_dimensions( + "paddle_error", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) return None, None try: paddle.base.core.eager._for_test_check_cuda_error() except Exception as err: print(f"[paddle_cuda] {self.api_config.config}\n{err!s}", flush=True) - write_to_log("paddle_cuda", self.api_config.config) + self._broadcast_to_comp_dimensions("paddle_cuda", self._PADDLE_AFFECTED_COMPS[iter_idx]) raise # ======== run paddle backward ======== @@ -351,31 +391,39 @@ def get_paddle_output(self, torch_grad_success): f"[config_input] {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("config_input", self.api_config.config) + self._broadcast_to_comp_dimensions( + "config_input", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) return None, None if self.should_ignore_paddle_error(err_str): print(f"[pass] {self.api_config.config}", flush=True) - write_to_log("pass", self.api_config.config) + self._broadcast_to_comp_dimensions( + "pass", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) return None, None if any(cuda_err in err_str for cuda_err in CUDA_ERROR): print( f"[paddle_cuda] phase=backward {self.api_config.config}\n{err_str}", ) - write_to_log("paddle_cuda", self.api_config.config) + self._broadcast_to_comp_dimensions( + "paddle_cuda", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) raise if any(cuda_err in err_str for cuda_err in CUDA_OOM): print( f"[oom] phase=backward {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("oom", self.api_config.config) + self._broadcast_to_comp_dimensions("oom", self._PADDLE_AFFECTED_COMPS[iter_idx]) raise print( f"[paddle_error] phase=backward {self.api_config.config}\n{err_str}", flush=True, ) traceback.print_exc() - write_to_log("paddle_error", self.api_config.config) + self._broadcast_to_comp_dimensions( + "paddle_error", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) return None, None try: @@ -385,7 +433,9 @@ def get_paddle_output(self, torch_grad_success): f"[paddle_cuda] phase=backward {self.api_config.config}\n{err!s}", flush=True, ) - write_to_log("paddle_cuda", self.api_config.config) + self._broadcast_to_comp_dimensions( + "paddle_cuda", self._PADDLE_AFFECTED_COMPS[iter_idx] + ) raise def process_paddle_outputs(obj): @@ -409,13 +459,13 @@ def compare(self, input1, input2, comp): f"[oom] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("oom", self.api_config.config) + write_to_comp_log(comp, "oom", self.api_config.config) else: print( f"[paddle_accuracy] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return else: print( @@ -423,7 +473,7 @@ def compare(self, input1, input2, comp): f"{type(input1)} / {type(input2)}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return elif isinstance(input1, (list, tuple)): if not isinstance(input2, (list, tuple)): @@ -432,7 +482,7 @@ def compare(self, input1, input2, comp): f"{type(input1)} / {type(input2)}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return if len(input1) != len(input2): print( @@ -441,7 +491,7 @@ def compare(self, input1, input2, comp): f"{type(input2)} : {len(input2)}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return for idx, (item1, item2) in enumerate(zip(input1, input2, strict=False)): if isinstance(item1, (paddle.Tensor, torch.Tensor)) and isinstance( @@ -456,13 +506,13 @@ def compare(self, input1, input2, comp): f"[oom] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("oom", self.api_config.config) + write_to_comp_log(comp, "oom", self.api_config.config) else: print( f"[paddle_accuracy] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return elif not isinstance(item1, (paddle.Tensor, torch.Tensor)) and not isinstance( item2, (paddle.Tensor, torch.Tensor) @@ -476,13 +526,13 @@ def compare(self, input1, input2, comp): f"[oom] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("oom", self.api_config.config) + write_to_comp_log(comp, "oom", self.api_config.config) else: print( f"[paddle_accuracy] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return else: print( @@ -490,7 +540,7 @@ def compare(self, input1, input2, comp): f"{type(item1)} / {type(item2)}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return else: try: @@ -502,13 +552,13 @@ def compare(self, input1, input2, comp): f"[oom] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("oom", self.api_config.config) + write_to_comp_log(comp, "oom", self.api_config.config) else: print( f"[paddle_accuracy] comp={comp} {self.api_config.config}\n{err_str}", flush=True, ) - write_to_log("paddle_accuracy", self.api_config.config) + write_to_comp_log(comp, "paddle_accuracy", self.api_config.config) return def assert_accuracy(self, tensor1, tensor2, comp, idx=0): @@ -622,6 +672,6 @@ def error_msg(msg): dtype, comp, ) - write_to_log("paddle_bitwise", config) + write_to_comp_log(comp, "paddle_bitwise", config) else: raise diff --git a/tester/api_config/log_writer.py b/tester/api_config/log_writer.py index 468211cc..45855c1b 100644 --- a/tester/api_config/log_writer.py +++ b/tester/api_config/log_writer.py @@ -6,6 +6,7 @@ import re import shutil from pathlib import Path +from typing import Literal import pandas as pd @@ -17,7 +18,24 @@ TMP_LOG_PATH = TEST_LOG_PATH / ".tmp" # 日志类型和对应的文件，可在下方进行注册 -LOG_PREFIXES = { +LogType = Literal[ + "checkpoint", + "pass", + "skip", + "paddle_error", + "paddle_accuracy", + "paddle_bitwise", + "paddle_cuda", + "paddle_crash", + "oom", + "timeout", + "torch_error", + "config_input", + "config_parse", + "config_convert", +] + +LOG_PREFIXES: dict[LogType, str] = { "checkpoint": "checkpoint", "pass": "api_config_pass", "skip": "api_config_skip", @@ -36,11 +54,32 @@ TERMINAL_LOG_TYPES = frozenset(LOG_PREFIXES) - {"checkpoint"} -_is_engineV2 = False +# === comp 维度配置（accuracy_stable 模式） === +COMP_TO_DIMENSION = { + "T1P1": "accuracy", + "T2P2": "accuracy", + "T1P2": "accuracy", + "T2P1": "accuracy", + "T1P1B": "accuracy_backward", + "T2P2B": "accuracy_backward", + "T1P2B": "accuracy_backward", + "T2P1B": "accuracy_backward", + "T1T2": "torch_stable", + "T1T2B": "torch_stable_backward", + "P1P2": "paddle_stable", + "P1P2B": "paddle_stable_backward", +} +ALL_DIMENSIONS = sorted(set(COMP_TO_DIMENSION.values())) +TOL_HEADER = ["API", "config", "dtype", "mode", "max_abs_diff", "max_rel_diff"] +STABLE_HEADER = ["API", "config", "dtype", "comp", "max_abs_diff", "max_rel_diff"] + +_use_worker_tmp_logs = False _process_file_handlers = {} _aggregated_offsets = {} _process_terminal_configs = {} +# 每维度的 terminal configs 追踪: dimension -> {config_line -> log_type} +_comp_terminal_configs: dict[str, dict[str, str]] = {} # Command line arguments configuration # Used in engine.py @@ -59,17 +98,25 @@ def set_cfg(cfg): CMD_CONFIG = cfg -def set_test_log_path(log_dir): - global TEST_LOG_PATH, TMP_LOG_PATH - TEST_LOG_PATH = DIR_PATH / log_dir - TEST_LOG_PATH.mkdir(parents=True, exist_ok=True) - TMP_LOG_PATH = TEST_LOG_PATH / ".tmp" +def _reset_runtime(): + close_process_files() + _aggregated_offsets.clear() + _process_terminal_configs.clear() + _comp_terminal_configs.clear() -def set_engineV2(): - global _is_engineV2 - _is_engineV2 = True - TMP_LOG_PATH.mkdir(exist_ok=True) +def init_log(log_dir=None, *, worker_tmp_logs=False): + global TEST_LOG_PATH, TMP_LOG_PATH, _use_worker_tmp_logs + _reset_runtime() + if log_dir: + TEST_LOG_PATH = DIR_PATH / log_dir + else: + TEST_LOG_PATH = DIR_PATH / "tester/api_config/test_log" + TEST_LOG_PATH.mkdir(parents=True, exist_ok=True) + TMP_LOG_PATH = TEST_LOG_PATH / ".tmp" + _use_worker_tmp_logs = worker_tmp_logs + if _use_worker_tmp_logs: + TMP_LOG_PATH.mkdir(exist_ok=True) def get_tmp_log_path(): @@ -121,17 +168,14 @@ def write_checkpoint(line): _process_terminal_configs.pop(line, None) -def write_terminal_log(log_type, line): +def write_terminal_log(log_type: LogType, line): write_to_log(log_type, line) write_checkpoint(line) -def get_log_file(log_type: str): - """获取指定日志类型和PID对应的日志文件路径""" - if log_type not in LOG_PREFIXES: - raise ValueError(f"Invalid log type: {log_type}") +def _get_log_file(log_type: LogType): prefix = LOG_PREFIXES[log_type] - if not _is_engineV2: + if not _use_worker_tmp_logs: cfg = get_cfg() filename = f"{prefix}{cfg.id}.txt" if cfg else f"{prefix}.txt" return TEST_LOG_PATH / filename @@ -139,27 +183,88 @@ def get_log_file(log_type: str): return TMP_LOG_PATH / f"{prefix}_{pid}.txt" -def write_to_log(log_type, line): +def _open_handler(file_path): + if file_path not in _process_file_handlers: + _process_file_handlers[file_path] = file_path.open("a", buffering=1) + return _process_file_handlers[file_path] + + +def _write_line(file_path, line): + try: + _open_handler(file_path).write(line + "\n") + return True + except Exception as err: + print(f"Error writing to {file_path}: {err}", flush=True) + return False + + +def write_to_log(log_type: LogType, line): """添加单条日志到当前进程的日志文件""" line = line.strip() if not line: return terminal_log_type = _process_terminal_configs.get(line) - if _is_engineV2 and log_type == "pass" and terminal_log_type not in (None, "pass"): + if _use_worker_tmp_logs and log_type == "pass" and terminal_log_type not in (None, "pass"): return - file_path = get_log_file(log_type) try: - if file_path not in _process_file_handlers: - _process_file_handlers[file_path] = file_path.open("a", buffering=1) - handler = _process_file_handlers[file_path] - handler.write(line + "\n") - if log_type in TERMINAL_LOG_TYPES and _is_engineV2: - _process_terminal_configs[line] = log_type + file_path = _get_log_file(log_type) except Exception as err: - print(f"Error writing to {file_path}: {err}", flush=True) + print(f"Error resolving log file for {log_type}: {err}", flush=True) + return + if _write_line(file_path, line) and log_type in TERMINAL_LOG_TYPES and _use_worker_tmp_logs: + _process_terminal_configs[line] = log_type + + +def has_comp_terminal_log(dimension, line): + """检查某个 comp 维度下是否已有终态分类""" + line = line.strip() + dim_configs = _comp_terminal_configs.get(dimension) + if dim_configs is None: + return False + return line in dim_configs + + +def _get_comp_file(dimension, log_type: LogType): + prefix = LOG_PREFIXES[log_type] + if _use_worker_tmp_logs: + comp_dir = TMP_LOG_PATH / "comp" / dimension + comp_dir.mkdir(parents=True, exist_ok=True) + return comp_dir / f"{prefix}_{os.getpid()}.txt" + comp_dir = TEST_LOG_PATH / "comp" / dimension + comp_dir.mkdir(parents=True, exist_ok=True) + cfg = get_cfg() + filename = f"{prefix}{cfg.id}.txt" if cfg else f"{prefix}.txt" + return comp_dir / filename + + +def write_to_comp_log(comp, log_type: LogType, line): + """写入 comp 维度的日志文件，同时更新主 _process_terminal_configs""" + try: + dimension = COMP_TO_DIMENSION[comp] + file_path = _get_comp_file(dimension, log_type) + except Exception as err: + print(f"Error resolving comp log file for {comp}/{log_type}: {err}", flush=True) + return + + line = line.strip() + if not line: + return + dim_configs = _comp_terminal_configs.setdefault(dimension, {}) + existing = dim_configs.get(line) + if existing is not None and existing != "pass" and log_type != existing: + return + + if not _write_line(file_path, line): + return -def read_log(log_type): + if log_type in TERMINAL_LOG_TYPES: + dim_configs[line] = log_type + if log_type in TERMINAL_LOG_TYPES and _use_worker_tmp_logs: + _process_terminal_configs[line] = log_type + + +def read_log(log_type: LogType): """读取文件所有行，返回集合""" if log_type not in LOG_PREFIXES: raise ValueError(f"Invalid log type: {log_type}") @@ -228,346 +333,364 @@ def _read_pending_bytes(file_path, end=False): return data, offset, file_size -def _commit_aggregate_offset(file_path, offset, clear=False): +def _save_offset(file_path, offset, clear=False): if clear: _aggregated_offsets.pop(file_path, None) else: _aggregated_offsets[file_path] = offset -def aggregate_logs(end=False, cleanup=False): - """聚合所有相同类型的日志文件""" - should_cleanup_tmp = end or cleanup - tmp_exists = TMP_LOG_PATH.exists() - if not tmp_exists and not should_cleanup_tmp: - TMP_LOG_PATH.mkdir(exist_ok=True) - return +def _save_offsets(pending_offsets, cleanup): + for file_path, offset in pending_offsets.items(): + _save_offset(file_path, offset, clear=cleanup) + if cleanup: + file_path.unlink() + - all_success = True +def _read_lines(log_files, cleanup): + all_lines = set() + pending_offsets = {} + for file_path in log_files: + try: + data, _, end_offset = _read_pending_bytes(file_path, end=cleanup) + pending_offsets[file_path] = end_offset + all_lines.update(line.strip() for line in data.decode().splitlines() if line.strip()) + except Exception as err: + print(f"Error reading {file_path}: {err}", flush=True) + return set(), {}, False + return all_lines, pending_offsets, True + + +def _agg_text(log_files, out_file, cleanup): + if not log_files: + return True + all_lines, pending_offsets, success = _read_lines(log_files, cleanup) + if not success: + return False + try: + if all_lines: + with out_file.open("a") as f: + f.writelines(f"{line}\n" for line in sorted(all_lines)) + except Exception as err: + print(f"Error writing to {out_file}: {err}", flush=True) + out_file.unlink(missing_ok=True) + return False + _save_offsets(pending_offsets, cleanup) + return True + + +def _agg_results(cleanup, tmp_exists): + success = True for prefix in LOG_PREFIXES.values(): log_files = list(TMP_LOG_PATH.glob(f"{prefix}_*.txt")) if tmp_exists else [] - if not log_files: - continue + out_file = TEST_LOG_PATH / f"{prefix}.txt" + success = _agg_text(log_files, out_file, cleanup) and success + return success - prefix_success = True - all_lines = set() - pending_offsets = {} - for file_path in log_files: - try: - data, start_offset, end_offset = _read_pending_bytes( - file_path, end=should_cleanup_tmp - ) - pending_offsets[file_path] = end_offset - all_lines.update( - line.strip() for line in data.decode().splitlines() if line.strip() - ) - except Exception as err: - print(f"Error reading {file_path}: {err}", flush=True) - prefix_success = False - break - if not prefix_success: - all_success = False - continue - aggregated_file = TEST_LOG_PATH / f"{prefix}.txt" - try: - if all_lines: - with aggregated_file.open("a") as f: - f.writelines(f"{line}\n" for line in sorted(all_lines)) - except Exception as err: - print(f"Error writing to {aggregated_file}: {err}", flush=True) - prefix_success = False - - if not prefix_success: - aggregated_file.unlink(missing_ok=True) - all_success = False - else: - for file_path, offset in pending_offsets.items(): - _commit_aggregate_offset(file_path, offset, clear=should_cleanup_tmp) - if should_cleanup_tmp: - file_path.unlink() - - log_success = True - log_file = TEST_LOG_PATH / "log_inorder.log" - tmp_log_files = sorted(TMP_LOG_PATH.glob("log_*.log")) if tmp_exists else [] - BUFFER_SIZE = 4 * 1024 * 1024 - pending_log_offsets = {} +def _agg_inorder(cleanup, tmp_exists): + log_files = sorted(TMP_LOG_PATH.glob("log_*.log")) if tmp_exists else [] + if not log_files: + return True + + out_file = TEST_LOG_PATH / "log_inorder.log" + pending_offsets = {} try: - with log_file.open("ab") as out_f: - for file_path in tmp_log_files: + with out_file.open("ab") as out_f: + for file_path in log_files: try: - data, start_offset, end_offset = _read_pending_bytes( - file_path, end=should_cleanup_tmp - ) - pending_log_offsets[file_path] = end_offset + data, _, end_offset = _read_pending_bytes(file_path, end=cleanup) + pending_offsets[file_path] = end_offset in_f = io.BytesIO(data) while True: - lines = in_f.readlines(BUFFER_SIZE) + lines = in_f.readlines(4 * 1024 * 1024) if not lines: break for line in lines: - if len(line) > 200000: # 如果行长度超过200000字节,截断 - # print( - # f"Truncating long line ({len(line)} bytes) in {file_path.name}" - # ) - out_f.write(line[:200000] + b"\n") - else: - out_f.write(line) + out_f.write(line[:200000] + b"\n" if len(line) > 200000 else line) except Exception as err: print(f"Error reading {file_path}: {err}", flush=True) - log_success = False - break + out_file.unlink(missing_ok=True) + return False except Exception as err: - print(f"Error writing to {log_file}: {err}", flush=True) - log_success = False + print(f"Error writing to {out_file}: {err}", flush=True) + out_file.unlink(missing_ok=True) + return False - if not log_success: - log_file.unlink(missing_ok=True) - all_success = False - else: - for file_path, offset in pending_log_offsets.items(): - _commit_aggregate_offset(file_path, offset, clear=should_cleanup_tmp) - if should_cleanup_tmp: - file_path.unlink() + _save_offsets(pending_offsets, cleanup) + return True - tol_success = True - tol_file = TEST_LOG_PATH / "tol.csv" - tmp_tol_files = sorted(TMP_LOG_PATH.glob("tol_*.csv")) if tmp_exists else [] - if tmp_tol_files: - pending_tol_offsets = {} + +def _agg_csv(log_files, out_file, header, cleanup): + if not log_files: + return True + + pending_offsets = {} + try: + is_new = not out_file.exists() or out_file.stat().st_size == 0 + with out_file.open("a", newline="") as out_f: + writer = csv.writer(out_f) + if is_new: + writer.writerow(header) + for file_path in log_files: + try: + data, start_offset, end_offset = _read_pending_bytes(file_path, end=cleanup) + pending_offsets[file_path] = end_offset + reader = csv.reader(io.StringIO(data.decode())) + if start_offset == 0: + next(reader, None) + for row in reader: + if row: + writer.writerow(row) + except Exception as err: + print(f"Error reading {file_path}: {err}", flush=True) + out_file.unlink(missing_ok=True) + return False + except Exception as err: + print(f"Error writing to {out_file}: {err}", flush=True) + out_file.unlink(missing_ok=True) + return False + + _save_offsets(pending_offsets, cleanup) + return True + + +def _sort_csv(file_path, columns): + if not file_path.exists(): + return + try: + df = pd.read_csv(file_path, on_bad_lines="warn") + df = df.sort_values(by=columns, ignore_index=True) + df.to_csv(file_path, index=False, na_rep="nan") + except Exception as err: + print(f"Error arranging {file_path}: {err}", flush=True) + + +def _count_logs(): + log_counts = {} + checkpoint_file = TEST_LOG_PATH / "checkpoint.txt" + api_configs = set() + try: + with checkpoint_file.open("r") as f: + api_configs = {line.strip() for line in f if line.strip()} + log_counts["checkpoint"] = len(api_configs) + except Exception as err: + print(f"Error reading {checkpoint_file}: {err}", flush=True) + + for log_type, prefix in LOG_PREFIXES.items(): + if log_type == "checkpoint": + continue + log_file = TEST_LOG_PATH / f"{prefix}.txt" + if not log_file.exists(): + continue try: - tol_is_new = not tol_file.exists() or tol_file.stat().st_size == 0 - with tol_file.open("a", newline="") as out_f: - writer = csv.writer(out_f) - if tol_is_new: - writer.writerow( - [ - "API", - "config", - "dtype", - "mode", - "max_abs_diff", - "max_rel_diff", - ] - ) - for file_path in tmp_tol_files: - try: - data, start_offset, end_offset = _read_pending_bytes( - file_path, end=should_cleanup_tmp - ) - pending_tol_offsets[file_path] = end_offset - reader = csv.reader(io.StringIO(data.decode())) - if start_offset == 0: - next(reader, None) - for row in reader: - if row: # 确保行不为空 - writer.writerow(row) - except Exception as err: - print(f"Error reading {file_path}: {err}", flush=True) - tol_success = False - break + with log_file.open("r") as f: + lines = {line.strip() for line in f if line.strip()} + api_configs -= lines + log_counts[log_type] = len(lines) except Exception as err: - print(f"Error writing to {tol_file}: {err}", flush=True) - tol_success = False - - if not tol_success: - tol_file.unlink(missing_ok=True) - all_success = False - else: - for file_path, offset in pending_tol_offsets.items(): - _commit_aggregate_offset(file_path, offset, clear=should_cleanup_tmp) - if should_cleanup_tmp: - file_path.unlink() - - stable_success = True - stable_file = TEST_LOG_PATH / "stable.csv" - tmp_stable_files = sorted(TMP_LOG_PATH.glob("stable_*.csv")) if tmp_exists else [] - if tmp_stable_files: - pending_stable_offsets = {} + print(f"Error reading {log_file}: {err}", flush=True) + + if api_configs: + log_counts["incomplete"] = len(api_configs) + incomplete_file = TEST_LOG_PATH / "api_config_incomplete.txt" try: - stable_is_new = not stable_file.exists() or stable_file.stat().st_size == 0 - with stable_file.open("a", newline="") as out_f: - writer = csv.writer(out_f) - if stable_is_new: - writer.writerow( - [ - "API", - "config", - "dtype", - "comp", - "max_abs_diff", - "max_rel_diff", - ] - ) - for file_path in tmp_stable_files: - try: - data, start_offset, end_offset = _read_pending_bytes( - file_path, end=should_cleanup_tmp - ) - pending_stable_offsets[file_path] = end_offset - reader = csv.reader(io.StringIO(data.decode())) - if start_offset == 0: - next(reader, None) - for row in reader: - if row: # 确保行不为空 - writer.writerow(row) - except Exception as err: - print(f"Error reading {file_path}: {err}", flush=True) - stable_success = False - break + with incomplete_file.open("w") as f: + f.writelines(f"{line}\n" for line in sorted(api_configs)) except Exception as err: - print(f"Error writing to {stable_file}: {err}", flush=True) - stable_success = False - - if not stable_success: - stable_file.unlink(missing_ok=True) - all_success = False - else: - for file_path, offset in pending_stable_offsets.items(): - _commit_aggregate_offset(file_path, offset, clear=should_cleanup_tmp) - if should_cleanup_tmp: - file_path.unlink() - - if ( - should_cleanup_tmp - and all_success - and TMP_LOG_PATH.exists() - and not os.listdir(TMP_LOG_PATH) - ): - shutil.rmtree(TMP_LOG_PATH) + print(f"Error writing to {incomplete_file}: {err}", flush=True) + return log_counts - if end: - if tol_file.exists(): - try: - df = pd.read_csv(tol_file, on_bad_lines="warn") - # df = df.drop_duplicates(subset=["config", "mode"], keep="last") - df = df.sort_values(by=["API", "dtype", "config", "mode"], ignore_index=True) - df.to_csv(tol_file, index=False, na_rep="nan") - except Exception as err: - print(f"Error arranging {tol_file}: {err}", flush=True) - if stable_file.exists(): - try: - df = pd.read_csv(stable_file, on_bad_lines="warn") - # df = df.drop_duplicates(subset=["config", "comp"], keep="last") - df = df.sort_values(by=["API", "dtype", "config", "comp"], ignore_index=True) - df.to_csv(stable_file, index=False, na_rep="nan") - except Exception as err: - print(f"Error arranging {stable_file}: {err}", flush=True) +def _agg_comp(cleanup, tmp_exists): + comp_tmp_dir = TMP_LOG_PATH / "comp" if tmp_exists else None + comp_out_dir = TEST_LOG_PATH / "comp" + has_comp = (comp_tmp_dir and comp_tmp_dir.exists()) or comp_out_dir.exists() + if not comp_tmp_dir or not comp_tmp_dir.exists(): + return has_comp - log_counts = {} - checkpoint_file = TEST_LOG_PATH / "checkpoint.txt" - api_configs = set() + for dim_dir in sorted(comp_tmp_dir.iterdir()): + if not dim_dir.is_dir(): + continue + out_dim_dir = comp_out_dir / dim_dir.name + out_dim_dir.mkdir(parents=True, exist_ok=True) + for prefix in LOG_PREFIXES.values(): + log_files = list(dim_dir.glob(f"{prefix}_*.txt")) + _agg_text(log_files, out_dim_dir / f"{prefix}.txt", cleanup) + if cleanup and dim_dir.exists() and not any(dim_dir.iterdir()): + dim_dir.rmdir() + if cleanup and comp_tmp_dir.exists() and not any(comp_tmp_dir.iterdir()): + comp_tmp_dir.rmdir() + return has_comp + + +def _scan_dups(log_dir): + config_to_types: dict[str, list[str]] = {} + for log_type, prefix in LOG_PREFIXES.items(): + if log_type == "checkpoint": + continue + log_file = log_dir / f"{prefix}.txt" + if not log_file.exists(): + continue try: - with checkpoint_file.open("r") as f: - api_configs = {line.strip() for line in f if line.strip()} - log_counts["checkpoint"] = len(api_configs) - except Exception as err: - print(f"Error reading {checkpoint_file}: {err}", flush=True) + with log_file.open("r") as f: + for raw_line in f: + line = raw_line.strip() + if line: + config_to_types.setdefault(line, []).append(log_type) + except Exception: + pass + return {config: types for config, types in config_to_types.items() if len(types) > 1} + + +def _add_dups(log_counts, scope, duplicates): + if duplicates: + log_counts.setdefault("_integrity_errors", []).append( + {"scope": scope, "duplicates": duplicates} + ) + +def _merge_comp_main(): + comp_out_dir = TEST_LOG_PATH / "comp" + if not comp_out_dir.exists(): + return + for dim_dir in sorted(comp_out_dir.iterdir()): + if not dim_dir.is_dir(): + continue for log_type, prefix in LOG_PREFIXES.items(): if log_type == "checkpoint": continue - log_file = TEST_LOG_PATH / f"{prefix}.txt" - if not log_file.exists(): + dim_log_file = dim_dir / f"{prefix}.txt" + if not dim_log_file.exists(): continue + main_log_file = TEST_LOG_PATH / f"{prefix}.txt" try: - with log_file.open("r") as f: - lines = {line.strip() for line in f if line.strip()} - api_configs -= lines - log_counts[log_type] = len(lines) + with dim_log_file.open("r") as inf: + lines = {line.strip() for line in inf if line.strip()} + if lines: + with main_log_file.open("a") as outf: + outf.writelines(f"{line}\n" for line in sorted(lines)) except Exception as err: - print(f"Error reading {log_file}: {err}", flush=True) + print(f"Error merging {dim_log_file} -> {main_log_file}: {err}", flush=True) - if api_configs: - log_counts["incomplete"] = len(api_configs) - incomplete_file = TEST_LOG_PATH / "api_config_incomplete.txt" - try: - with incomplete_file.open("w") as f: - f.writelines(f"{line}\n" for line in sorted(api_configs)) - except Exception as err: - print(f"Error writing to {incomplete_file}: {err}", flush=True) - # === 终态日志互斥性检查 === - config_to_types: dict[str, list[str]] = {} - for log_type, prefix in LOG_PREFIXES.items(): - if log_type == "checkpoint": - continue - log_file = TEST_LOG_PATH / f"{prefix}.txt" - if not log_file.exists(): - continue - try: - with log_file.open("r") as f: - for raw_line in f: - line = raw_line.strip() - if line: - config_to_types.setdefault(line, []).append(log_type) - except Exception: - pass - - duplicates = {config: types for config, types in config_to_types.items() if len(types) > 1} - if duplicates: - print("\n" + "!" * 50) - print("INTEGRITY ERROR: configs found in multiple log types:") - for config, types in sorted(duplicates.items())[:20]: - print(f" {config}") - print(f" -> {', '.join(types)}") - if len(duplicates) > 20: - print(f" ... and {len(duplicates) - 20} more") - print("!" * 50 + "\n") - assert not duplicates, ( - f"Log integrity violation: {len(duplicates)} config(s) appear in " - f"multiple terminal log types. This indicates a classification bug." - ) - - return log_counts +def _check_logs(log_counts, has_comp): + comp_out_dir = TEST_LOG_PATH / "comp" + if has_comp: + for dim_dir in sorted(comp_out_dir.iterdir()) if comp_out_dir.exists() else []: + if dim_dir.is_dir(): + _add_dups(log_counts, f"comp/{dim_dir.name}", _scan_dups(dim_dir)) + _merge_comp_main() + return + _add_dups(log_counts, "main log directory", _scan_dups(TEST_LOG_PATH)) + + +def _clean_tmp(cleanup, all_success): + if cleanup and all_success and TMP_LOG_PATH.exists() and not any(TMP_LOG_PATH.iterdir()): + shutil.rmtree(TMP_LOG_PATH) + + +def aggregate_logs(end=False, cleanup=False): + """聚合所有相同类型的日志文件""" + cleanup_tmp = end or cleanup + tmp_exists = TMP_LOG_PATH.exists() + if not tmp_exists and not cleanup_tmp: + TMP_LOG_PATH.mkdir(exist_ok=True) + return + + tol_file = TEST_LOG_PATH / "tol.csv" + stable_file = TEST_LOG_PATH / "stable.csv" + all_success = _agg_results(cleanup_tmp, tmp_exists) + all_success = _agg_inorder(cleanup_tmp, tmp_exists) and all_success + all_success = ( + _agg_csv( + sorted(TMP_LOG_PATH.glob("tol_*.csv")) if tmp_exists else [], + tol_file, + TOL_HEADER, + cleanup_tmp, + ) + and all_success + ) + all_success = ( + _agg_csv( + sorted(TMP_LOG_PATH.glob("stable_*.csv")) if tmp_exists else [], + stable_file, + STABLE_HEADER, + cleanup_tmp, + ) + and all_success + ) + _clean_tmp(cleanup_tmp, all_success) + + if not end: + return + + _sort_csv(tol_file, ["API", "dtype", "config", "mode"]) + _sort_csv(stable_file, ["API", "dtype", "config", "comp"]) + log_counts = _count_logs() + has_comp = _agg_comp(cleanup_tmp, tmp_exists) + _check_logs(log_counts, has_comp) + return log_counts + + +def _visible_counts(log_counts): + return {key: value for key, value in log_counts.items() if not key.startswith("_")} + + +def _sum_counts(log_counts, log_types): + return sum(log_counts.get(log_type, 0) for log_type in log_types) + + +def _print_dups(integrity_errors): + for issue in integrity_errors: + scope = issue["scope"] + duplicates = issue["duplicates"] + print("\n" + "!" * 50) + print(f"WARNING: configs found in multiple log types ({scope}):") + for config, types in sorted(duplicates.items())[:20]: + print(f" {config}") + print(f" -> {', '.join(types)}") + if len(duplicates) > 20: + print(f" ... and {len(duplicates) - 20} more") + print( + f"Found {len(duplicates)} duplicated config(s). " + "Please check log classification, but the test statistics above are still available." + ) + print("!" * 50 + "\n") def print_log_info(all_case, log_counts=None): """打印日志统计信息""" if log_counts is None: log_counts = {} - test_case = log_counts.get("checkpoint", 0) - pass_case = log_counts.get("pass", 0) - skip_case = log_counts.get("skip", 0) - paddle_issue_case = sum( - log_counts.get(log_type, 0) - for log_type in [ - "paddle_error", - "paddle_accuracy", - "paddle_bitwise", - "paddle_cuda", - "paddle_crash", - ] - ) - test_issue_case = sum( - log_counts.get(log_type, 0) - for log_type in [ - "torch_error", - "config_input", - "config_parse", - "config_convert", - ] - ) - retest_case = sum(log_counts.get(log_type, 0) for log_type in ["oom", "timeout"]) + integrity_errors = log_counts.get("_integrity_errors", []) + counts = _visible_counts(log_counts) + paddle_types = [ + "paddle_error", + "paddle_accuracy", + "paddle_bitwise", + "paddle_cuda", + "paddle_crash", + ] + test_types = ["torch_error", "config_input", "config_parse", "config_convert"] - # 打印统计信息 print("\n" + "=" * 50) print("Test Case Statistics".center(50)) print("=" * 50) print(f"{'Pending cases':<30}: {all_case:>8}") - print(f"{'Tested cases':<30}: {test_case:>8}") - print(f"{'Pass cases':<30}: {pass_case:>8}") - print(f"{'Skip cases':<30}: {skip_case:>8}") - print(f"{'Paddle issue cases':<30}: {paddle_issue_case:>8}") - print(f"{'Test issue cases':<30}: {test_issue_case:>8}") - print(f"{'Retest cases':<30}: {retest_case:>8}") - if log_counts: + print(f"{'Tested cases':<30}: {counts.get('checkpoint', 0):>8}") + print(f"{'Pass cases':<30}: {counts.get('pass', 0):>8}") + print(f"{'Skip cases':<30}: {counts.get('skip', 0):>8}") + print(f"{'Paddle issue cases':<30}: {_sum_counts(counts, paddle_types):>8}") + print(f"{'Test issue cases':<30}: {_sum_counts(counts, test_types):>8}") + print(f"{'Retest cases':<30}: {_sum_counts(counts, ['oom', 'timeout']):>8}") + if counts: print("-" * 50) print("Log Type Breakdown:") - for log_type, count in log_counts.items(): + for log_type, count in counts.items(): print(f" {log_type:<28}: {count:>8}") print("=" * 50 + "\n") + _print_dups(integrity_errors) stdout_fd = None @@ -620,97 +743,61 @@ def restore_stdio(): orig_stderr_fd = None -def log_accuracy_tolerance(error_msg, api, config, dtype, is_backward=False): - """从 torch.testing.assert_close 的异常消息中提取最大绝对误差和相对误差 - 将误差数据记录到 CSV 文件 - """ - output_file = TMP_LOG_PATH / f"tol_{os.getpid()}.csv" - mode = "backward" if is_backward else "forward" - print(f"mode={mode} {config}\n{error_msg}", flush=True) - +def _get_diff(error_msg, abs_pattern, rel_pattern): if error_msg == "Identical": - max_abs_diff = 0.0 - max_rel_diff = 0.0 - else: - max_abs_diff = None - max_rel_diff = None + return 0.0, 0.0 - # 使用正则表达式提取误差值 - abs_pattern = ( - r"(?:Absolute|Greatest absolute) difference: (\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" - ) - rel_pattern = ( - r"(?:Relative|Greatest relative) difference: (\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" - ) - abs_match = re.search(abs_pattern, error_msg) - rel_match = re.search(rel_pattern, error_msg) + abs_match = re.search(abs_pattern, error_msg) + rel_match = re.search(rel_pattern, error_msg) + if not abs_match or not rel_match: + return None, None + try: + return float(abs_match.group(1)), float(rel_match.group(1)) + except ValueError: + return None, None - if abs_match and rel_match: - try: - max_abs_diff = float(abs_match.group(1)) - max_rel_diff = float(rel_match.group(1)) - except ValueError: - pass - row = [api, config, dtype, mode, str(max_abs_diff), str(max_rel_diff)] +def _append_csv(output_file, header, row): try: - with open(output_file, mode="a", newline="") as f: + is_new = not output_file.exists() or output_file.stat().st_size == 0 + with output_file.open("a", newline="") as f: writer = csv.writer(f) - if not output_file.exists() or output_file.stat().st_size == 0: - writer.writerow( - [ - "API", - "config", - "dtype", - "mode", - "max_abs_diff", - "max_rel_diff", - ] - ) + if is_new: + writer.writerow(header) writer.writerow(row) except Exception as err: print(f"Error writing to {output_file}: {err}", flush=True) -def log_accuracy_stable(error_msg, api, config, dtype, comp): - output_file = TMP_LOG_PATH / f"stable_{os.getpid()}.csv" - print(f"comp={comp} {config}\n{error_msg}", flush=True) - - if error_msg == "Identical": - max_abs_diff = 0.0 - max_rel_diff = 0.0 - else: - max_abs_diff = None - max_rel_diff = None - - # 使用正则表达式提取误差值 - abs_pattern = r"(?:Absolute|Greatest absolute|Max absolute) difference(?: among violations)?: (\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" - rel_pattern = r"(?:Relative|Greatest relative|Max relative) difference(?: among violations)?: (\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" - abs_match = re.search(abs_pattern, error_msg) - rel_match = re.search(rel_pattern, error_msg) +def log_accuracy_tolerance(error_msg, api, config, dtype, is_backward=False): + """从 torch.testing.assert_close 的异常消息中提取最大绝对误差和相对误差 + 将误差数据记录到 CSV 文件 + """ + mode = "backward" if is_backward else "forward" + print(f"mode={mode} {config}\n{error_msg}", flush=True) + abs_pattern = ( + r"(?:Absolute|Greatest absolute) difference: " + r"(\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" + ) + rel_pattern = ( + r"(?:Relative|Greatest relative) difference: " + r"(\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" + ) + max_abs_diff, max_rel_diff = _get_diff(error_msg, abs_pattern, rel_pattern) + row = [api, config, dtype, mode, str(max_abs_diff), str(max_rel_diff)] + _append_csv(TMP_LOG_PATH / f"tol_{os.getpid()}.csv", TOL_HEADER, row) - if abs_match and rel_match: - try: - max_abs_diff = float(abs_match.group(1)) - max_rel_diff = float(rel_match.group(1)) - except ValueError: - pass +def log_accuracy_stable(error_msg, api, config, dtype, comp): + print(f"comp={comp} {config}\n{error_msg}", flush=True) + abs_pattern = ( + r"(?:Absolute|Greatest absolute|Max absolute) difference(?: among violations)?: " + r"(\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" + ) + rel_pattern = ( + r"(?:Relative|Greatest relative|Max relative) difference(?: among violations)?: " + r"(\d+\.?\d*(?:[eE][+-]?\d+)?|nan|inf)\b" + ) + max_abs_diff, max_rel_diff = _get_diff(error_msg, abs_pattern, rel_pattern) row = [api, config, dtype, comp, str(max_abs_diff), str(max_rel_diff)] - try: - with open(output_file, mode="a", newline="") as f: - writer = csv.writer(f) - if not output_file.exists() or output_file.stat().st_size == 0: - writer.writerow( - [ - "API", - "config", - "dtype", - "comp", - "max_abs_diff", - "max_rel_diff", - ] - ) - writer.writerow(row) - except Exception as err: - print(f"Error writing to {output_file}: {err}", flush=True) + _append_csv(TMP_LOG_PATH / f"stable_{os.getpid()}.csv", STABLE_HEADER, row) diff --git a/tools/error_stat/error_stat.py b/tools/error_stat/error_stat.py index 86d2ef0f..cfe5dc97 100644 --- a/tools/error_stat/error_stat.py +++ b/tools/error_stat/error_stat.py @@ -61,16 +61,16 @@ def check_count_consistency(parsed_keys, config_keys, prefix): # 校验从 log_inorder.log 解析到的 case 数量与结果文件一致，不一致说明日志不完整或被截断 parsed_len = len(parsed_keys) config_len = len(config_keys) - if parsed_len != config_len: - missing_keys = config_keys - parsed_keys - extra_keys = parsed_keys - config_keys - msg = ( - f"[ASSERT ERROR] {prefix} 数量不一致: " - f"config={config_len}, parsed={parsed_len}, " - f"缺失={len(missing_keys)} {sorted(missing_keys)[:3]}, " - f"多余={len(extra_keys)} {sorted(extra_keys)[:3]}" - ) - raise AssertionError(msg) + if parsed_len == config_len: + return None + missing_keys = config_keys - parsed_keys + extra_keys = parsed_keys - config_keys + return ( + f"[WARNING] {prefix} 数量不一致: " + f"config={config_len}, parsed={parsed_len}, " + f"缺失={len(missing_keys)} {sorted(missing_keys)[:3]}, " + f"多余={len(extra_keys)} {sorted(extra_keys)[:3]}" + ) def read_configs(file_path): @@ -155,6 +155,20 @@ def merge_classified_logs(classified_logs, log_types): return merged_logs +def print_consistency_warnings(warnings): + if not warnings: + return + print("\n" + "!" * 50) + print("WARNING: log count consistency issues were found:") + for warning in warnings: + print(f" {warning}") + print( + "Result files have still been generated. " + "Please check whether logs are incomplete or duplicated." + ) + print("!" * 50 + "\n") + + def write_logs_and_meta(output_path, logs_dict, prefix): # 为指定分类写出三个文件：完整日志块、API 名列表、config 字符串列表 output_path = Path(output_path) @@ -192,13 +206,18 @@ def error_state(input_path, output_path, split_errors=False): return classified_logs = classify_by_config(logs, config_sets) + consistency_warnings = [] pass_logs = classified_logs.get("pass", {}) - check_count_consistency(set(pass_logs), config_sets["pass"], "pass") + warning = check_count_consistency(set(pass_logs), config_sets["pass"], "pass") + if warning: + consistency_warnings.append(warning) write_logs_and_meta(output_path, pass_logs, "pass") skip_logs = classified_logs.get("skip", {}) - check_count_consistency(set(skip_logs), config_sets["skip"], "skip") + warning = check_count_consistency(set(skip_logs), config_sets["skip"], "skip") + if warning: + consistency_warnings.append(warning) if skip_logs: write_logs_and_meta(output_path, skip_logs, "skip") @@ -208,19 +227,26 @@ def error_state(input_path, output_path, split_errors=False): if log_type in ("checkpoint", "pass", "skip"): continue category_logs = classified_logs.get(log_type, {}) - check_count_consistency(set(category_logs), configs, log_type) + warning = check_count_consistency(set(category_logs), configs, log_type) + if warning: + consistency_warnings.append(warning) if category_logs: write_logs_and_meta(output_path, category_logs, log_type) + print_consistency_warnings(consistency_warnings) return # 默认汇总模式：按 SUMMARY_GROUPS 合并同组分类后输出 for group_name, log_types in SUMMARY_GROUPS.items(): group_logs = merge_classified_logs(classified_logs, log_types) group_configs = set().union(*(config_sets[log_type] for log_type in log_types)) - check_count_consistency(set(group_logs), group_configs, group_name) + warning = check_count_consistency(set(group_logs), group_configs, group_name) + if warning: + consistency_warnings.append(warning) if group_logs: write_logs_and_meta(output_path, group_logs, group_name) + print_consistency_warnings(consistency_warnings) + def parse_args(argv=None): parser = argparse.ArgumentParser(description="test_log 分类整理工具（可按终态类型拆分）") diff --git a/tools/prof/paddleapitest_matmul_heatmap.py b/tools/prof/paddleapitest_matmul_heatmap.py index 01bf658f..8d33caf5 100644 --- a/tools/prof/paddleapitest_matmul_heatmap.py +++ b/tools/prof/paddleapitest_matmul_heatmap.py @@ -168,8 +168,7 @@ def init_log_writer(output_dir: Path) -> None: log_dir.mkdir(parents=True, exist_ok=True) try: - log_writer.set_test_log_path(str(log_dir)) - log_writer.set_engineV2() + log_writer.init_log(str(log_dir), worker_tmp_logs=True) except Exception as err: print(f"[warn] failed to initialize log_writer: {err}") From 8f5b7759203f30203fa5131b6facf076f57f2cc5 Mon Sep 17 00:00:00 2001 From: cangtianhuang <1903374751@qq.com> Date: Thu, 16 Jul 2026 15:21:41 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=9A=B8=20Update=20log=20writer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tester/api_config/log_writer.py | 93 ++++++++++++++++++++++++++------- 1 file changed, 73 insertions(+), 20 deletions(-) diff --git a/tester/api_config/log_writer.py b/tester/api_config/log_writer.py index 45855c1b..752fabbc 100644 --- a/tester/api_config/log_writer.py +++ b/tester/api_config/log_writer.py @@ -491,14 +491,16 @@ def _count_logs(): except Exception as err: print(f"Error reading {log_file}: {err}", flush=True) + incomplete_file = TEST_LOG_PATH / "api_config_incomplete.txt" if api_configs: log_counts["incomplete"] = len(api_configs) - incomplete_file = TEST_LOG_PATH / "api_config_incomplete.txt" try: with incomplete_file.open("w") as f: f.writelines(f"{line}\n" for line in sorted(api_configs)) except Exception as err: print(f"Error writing to {incomplete_file}: {err}", flush=True) + else: + incomplete_file.unlink(missing_ok=True) return log_counts @@ -550,37 +552,59 @@ def _add_dups(log_counts, scope, duplicates): ) -def _merge_comp_main(): +def _read_log_lines(log_file): + if not log_file.exists(): + return set() + try: + with log_file.open("r") as f: + return {line.strip() for line in f if line.strip()} + except Exception as err: + print(f"Error reading {log_file}: {err}", flush=True) + return set() + + +def _sync_comp_main_summary(): comp_out_dir = TEST_LOG_PATH / "comp" if not comp_out_dir.exists(): return + + main_lines_by_type = { + log_type: _read_log_lines(TEST_LOG_PATH / f"{prefix}.txt") + for log_type, prefix in LOG_PREFIXES.items() + if log_type != "checkpoint" + } for dim_dir in sorted(comp_out_dir.iterdir()): if not dim_dir.is_dir(): continue for log_type, prefix in LOG_PREFIXES.items(): if log_type == "checkpoint": continue - dim_log_file = dim_dir / f"{prefix}.txt" - if not dim_log_file.exists(): - continue - main_log_file = TEST_LOG_PATH / f"{prefix}.txt" - try: - with dim_log_file.open("r") as inf: - lines = {line.strip() for line in inf if line.strip()} - if lines: - with main_log_file.open("a") as outf: - outf.writelines(f"{line}\n" for line in sorted(lines)) - except Exception as err: - print(f"Error merging {dim_log_file} -> {main_log_file}: {err}", flush=True) + main_lines_by_type[log_type].update(_read_log_lines(dim_dir / f"{prefix}.txt")) + + for log_type, lines in main_lines_by_type.items(): + log_file = TEST_LOG_PATH / f"{LOG_PREFIXES[log_type]}.txt" + try: + if lines: + with log_file.open("w") as f: + f.writelines(f"{line}\n" for line in sorted(lines)) + else: + log_file.unlink(missing_ok=True) + except Exception as err: + print(f"Error writing to {log_file}: {err}", flush=True) def _check_logs(log_counts, has_comp): comp_out_dir = TEST_LOG_PATH / "comp" if has_comp: + log_counts["_multi_classification"] = True for dim_dir in sorted(comp_out_dir.iterdir()) if comp_out_dir.exists() else []: - if dim_dir.is_dir(): - _add_dups(log_counts, f"comp/{dim_dir.name}", _scan_dups(dim_dir)) - _merge_comp_main() + if not dim_dir.is_dir(): + continue + duplicates = _scan_dups(dim_dir) + if duplicates: + log_counts.setdefault("_comp_integrity_errors", []).append( + {"scope": f"comp/{dim_dir.name}", "duplicates": duplicates} + ) return _add_dups(log_counts, "main log directory", _scan_dups(TEST_LOG_PATH)) @@ -627,8 +651,10 @@ def aggregate_logs(end=False, cleanup=False): _sort_csv(tol_file, ["API", "dtype", "config", "mode"]) _sort_csv(stable_file, ["API", "dtype", "config", "comp"]) - log_counts = _count_logs() has_comp = _agg_comp(cleanup_tmp, tmp_exists) + if has_comp: + _sync_comp_main_summary() + log_counts = _count_logs() _check_logs(log_counts, has_comp) return log_counts @@ -659,11 +685,31 @@ def _print_dups(integrity_errors): print("!" * 50 + "\n") +def _print_comp_dups(comp_integrity_errors): + for issue in comp_integrity_errors: + scope = issue["scope"] + duplicates = issue["duplicates"] + print("\n" + "!" * 50) + print(f"WARNING: configs found in multiple log types within {scope}:") + for config, types in sorted(duplicates.items())[:20]: + print(f" {config}") + print(f" -> {', '.join(types)}") + if len(duplicates) > 20: + print(f" ... and {len(duplicates) - 20} more") + print( + f"Found {len(duplicates)} duplicated config(s) inside {scope}. " + "Each comp dimension should still be mutually exclusive." + ) + print("!" * 50 + "\n") + + def print_log_info(all_case, log_counts=None): """打印日志统计信息""" if log_counts is None: log_counts = {} integrity_errors = log_counts.get("_integrity_errors", []) + comp_integrity_errors = log_counts.get("_comp_integrity_errors", []) + is_multi_classification = log_counts.get("_multi_classification") counts = _visible_counts(log_counts) paddle_types = [ "paddle_error", @@ -677,7 +723,7 @@ def print_log_info(all_case, log_counts=None): print("\n" + "=" * 50) print("Test Case Statistics".center(50)) print("=" * 50) - print(f"{'Pending cases':<30}: {all_case:>8}") + print(f"{'Remaining cases':<30}: {all_case:>8}") print(f"{'Tested cases':<30}: {counts.get('checkpoint', 0):>8}") print(f"{'Pass cases':<30}: {counts.get('pass', 0):>8}") print(f"{'Skip cases':<30}: {counts.get('skip', 0):>8}") @@ -687,10 +733,17 @@ def print_log_info(all_case, log_counts=None): if counts: print("-" * 50) print("Log Type Breakdown:") + if is_multi_classification: + print( + " Note: accuracy_stable comp-dimension breakdown; one config may appear in multiple result sets." + ) for log_type, count in counts.items(): print(f" {log_type:<28}: {count:>8}") print("=" * 50 + "\n") - _print_dups(integrity_errors) + if is_multi_classification: + _print_comp_dups(comp_integrity_errors) + else: + _print_dups(integrity_errors) stdout_fd = None From a8027264dc2ef664057a38cbca767f609b119688 Mon Sep 17 00:00:00 2001 From: cangtianhuang <1903374751@qq.com> Date: Thu, 16 Jul 2026 16:45:54 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=90=9B=20Fix=20.tmp=20remain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- engineV2.py | 4 ++-- engineV4.py | 4 ++-- tester/api_config/log_writer.py | 16 +++++++++------- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/engineV2.py b/engineV2.py index 63a10b96..6e9a813f 100644 --- a/engineV2.py +++ b/engineV2.py @@ -1239,7 +1239,7 @@ def cleanup_handler(*args): elif err.exitcode in (-signal.SIGKILL, -signal.SIGTERM): checkpoint_ready = False print( - f"[warn] Worker was externally killed " + f"[warn] Worker was externally killed for {config} " f"(exit={err.exitcode}); case will be retried on next run.", flush=True, ) @@ -1274,7 +1274,7 @@ def cleanup_handler(*args): finally: print(f"{tested_case} cases have been tested.", flush=True) log_counts = aggregate_logs(end=True) - print_log_info(all_case, log_counts) + print_log_info(max(all_case - tested_case, 0), log_counts) end_time = time.time() total_time = end_time - start_time print(f"Test time: {round(total_time / 60, 3)} minutes.", flush=True) diff --git a/engineV4.py b/engineV4.py index e5097016..a831a12f 100644 --- a/engineV4.py +++ b/engineV4.py @@ -1954,7 +1954,7 @@ def cleanup_handler(*args): if external_kill: print( - f"[warn] Worker was externally killed " + f"[warn] Worker was externally killed for {config} " f"(exit={exitcode}); case will be retried on next run.", flush=True, ) @@ -2061,7 +2061,7 @@ def cleanup_handler(*args): cleanup_sanitizer_tmp_dir() print(f"{tested_case} cases have been tested.", flush=True) log_counts = aggregate_logs(end=True) - print_log_info(all_case, log_counts) + print_log_info(max(all_case - tested_case, 0), log_counts) end_time = time.time() total_time = end_time - start_time print(f"Test time: {round(total_time / 60, 3)} minutes.", flush=True) diff --git a/tester/api_config/log_writer.py b/tester/api_config/log_writer.py index 752fabbc..93ac4f06 100644 --- a/tester/api_config/log_writer.py +++ b/tester/api_config/log_writer.py @@ -597,6 +597,8 @@ def _check_logs(log_counts, has_comp): comp_out_dir = TEST_LOG_PATH / "comp" if has_comp: log_counts["_multi_classification"] = True + if _scan_dups(TEST_LOG_PATH): + log_counts["_has_multi_result_overlap"] = True for dim_dir in sorted(comp_out_dir.iterdir()) if comp_out_dir.exists() else []: if not dim_dir.is_dir(): continue @@ -644,14 +646,15 @@ def aggregate_logs(end=False, cleanup=False): ) and all_success ) - _clean_tmp(cleanup_tmp, all_success) if not end: + _clean_tmp(cleanup_tmp, all_success) return _sort_csv(tol_file, ["API", "dtype", "config", "mode"]) _sort_csv(stable_file, ["API", "dtype", "config", "comp"]) has_comp = _agg_comp(cleanup_tmp, tmp_exists) + _clean_tmp(cleanup_tmp, all_success) if has_comp: _sync_comp_main_summary() log_counts = _count_logs() @@ -703,13 +706,14 @@ def _print_comp_dups(comp_integrity_errors): print("!" * 50 + "\n") -def print_log_info(all_case, log_counts=None): +def print_log_info(remaining_case, log_counts=None): """打印日志统计信息""" if log_counts is None: log_counts = {} integrity_errors = log_counts.get("_integrity_errors", []) comp_integrity_errors = log_counts.get("_comp_integrity_errors", []) is_multi_classification = log_counts.get("_multi_classification") + has_multi_result_overlap = log_counts.get("_has_multi_result_overlap") counts = _visible_counts(log_counts) paddle_types = [ "paddle_error", @@ -723,7 +727,7 @@ def print_log_info(all_case, log_counts=None): print("\n" + "=" * 50) print("Test Case Statistics".center(50)) print("=" * 50) - print(f"{'Remaining cases':<30}: {all_case:>8}") + print(f"{'Remaining cases':<30}: {remaining_case:>8}") print(f"{'Tested cases':<30}: {counts.get('checkpoint', 0):>8}") print(f"{'Pass cases':<30}: {counts.get('pass', 0):>8}") print(f"{'Skip cases':<30}: {counts.get('skip', 0):>8}") @@ -733,10 +737,8 @@ def print_log_info(all_case, log_counts=None): if counts: print("-" * 50) print("Log Type Breakdown:") - if is_multi_classification: - print( - " Note: accuracy_stable comp-dimension breakdown; one config may appear in multiple result sets." - ) + if is_multi_classification and has_multi_result_overlap: + print(" Note: In accuracy_stable mode, one config may appear in multiple result sets.") for log_type, count in counts.items(): print(f" {log_type:<28}: {count:>8}") print("=" * 50 + "\n")