diff --git a/test_pipeline/config_preprocess/config_lines.py b/test_pipeline/config_preprocess/config_lines.py new file mode 100644 index 00000000..7241aa00 --- /dev/null +++ b/test_pipeline/config_preprocess/config_lines.py @@ -0,0 +1,91 @@ +"""Paddle API 配置行的共享解析工具。""" + +# 每个逻辑调用必须以 paddle. 开头并具有完整外层括号。 +# 同一物理行允许连续记录多个调用，调用之间可以没有空白分隔符。 +# 字符串和嵌套参数中的 paddle. 只属于当前调用，不能成为切分边界。 +# 不能无损归属的残余文本必须报错，由调用方决定拒绝或终止。 + +from __future__ import annotations + +import string + + +# 采集配置中的 API 名只接受 ASCII 标识符，避免 Unicode 或运算符混入边界。 +API_NAME_CHARS = frozenset(string.ascii_letters + string.digits + "_.") + + +def matching_close(text, start, opening, closing): + """返回与 start 处括号匹配的结束位置。""" + depth = 0 + quote = None + escaped = False + for index in range(start, len(text)): + char = text[index] + if quote is not None: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + continue + if char in {'"', "'"}: + quote = char + elif char == opening: + depth += 1 + elif char == closing: + depth -= 1 + if depth == 0: + return index + return None + + +def _api_name_end(text, start): + """返回 API 名结束位置，并阻止粘连的第二个 paddle 前缀混入名称。""" + index = start + while index < len(text) and text[index] in API_NAME_CHARS: + # 名称内部再次出现 paddle. 表示前一个调用缺少左括号，而非更长 API 名。 + if index > start and text.startswith("paddle.", index): + break + index += 1 + return index + + +def _is_valid_api_name(api_name): + """验证 paddle 后的每级名称均为 Python 标识符。""" + parts = api_name.split(".") + return parts[0] == "paddle" and len(parts) > 1 and all( + part + and (part[0] in string.ascii_letters or part[0] == "_") + and all(char in string.ascii_letters + string.digits + "_" for char in part[1:]) + for part in parts[1:] + ) + + +def split_top_level_calls(text): + """将一行中连续的顶层 paddle.* 调用无损拆分。""" + calls = [] + cursor = 0 + while cursor < len(text): + while cursor < len(text) and text[cursor].isspace(): + cursor += 1 + if cursor >= len(text): + break + if not text.startswith("paddle.", cursor): + # 非 paddle 残余文本不能静默丢弃，否则会造成测试覆盖缺失。 + raise ValueError( + "顶层调用结束后存在无法识别的内容，期望下一个 paddle.* 调用" + ) + + name_end = _api_name_end(text, cursor) + if not _is_valid_api_name(text[cursor:name_end]): + raise ValueError("顶层 paddle.* API 名不合法") + if name_end >= len(text) or text[name_end] != "(": + raise ValueError("顶层 paddle.* 调用缺少左括号") + closing = matching_close(text, name_end, "(", ")") + if closing is None: + raise ValueError("顶层 paddle.* 调用括号不匹配") + # 调用文本原样保留，规范化阶段不重排参数或改变空白。 + calls.append(text[cursor : closing + 1]) + cursor = closing + 1 + return calls diff --git a/test_pipeline/config_preprocess/normalize_config_lines.py b/test_pipeline/config_preprocess/normalize_config_lines.py new file mode 100644 index 00000000..cc9cb98c --- /dev/null +++ b/test_pipeline/config_preprocess/normalize_config_lines.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""将原始配置规范化为每行一个完整的 Paddle API 调用。""" + +# 规范化只修复物理换行丢失，不改变调用内容、顺序或重复次数。 +# 成功拆出的调用立即写入输出，避免 GB 级输入常驻内存。 +# 失败的原始行连同行号写入拒绝文件，供修复后重新处理。 +# 严格模式在完整扫描结束后失败，已完成工作和错误证据都会保留。 + +from __future__ import annotations + +import argparse +from pathlib import Path + +from test_pipeline.config_preprocess.config_lines import split_top_level_calls + + +def normalize_file(input_path, output_path, reject_path): + """流式拆分文件并记录拒绝行。""" + # 返回计数只描述结构规范化，不做 APIConfig 语义解析或文本去重。 + input_path = Path(input_path) + output_path = Path(output_path) + reject_path = Path(reject_path) + if input_path.resolve() == output_path.resolve(): + raise ValueError("输入和输出文件不能相同") + + output_path.parent.mkdir(parents=True, exist_ok=True) + reject_path.parent.mkdir(parents=True, exist_ok=True) + counts = {"lines": 0, "calls": 0, "split_lines": 0, "rejected": 0} + with ( + input_path.open(encoding="utf-8") as source_file, + output_path.open("w", encoding="utf-8") as normalized_file, + reject_path.open("w", encoding="utf-8") as reject_file, + ): + for line_number, raw_line in enumerate(source_file, start=1): + counts["lines"] += 1 + config = raw_line.strip() + if not config: + continue + try: + calls = split_top_level_calls(config) + except ValueError as error: + # 错误行完整落盘，后续可以修复后单独重放，不会丢失原始数据。 + counts["rejected"] += 1 + reject_file.write(f"{input_path}:{line_number}\t{error}\t{config}\n") + continue + if len(calls) > 1: + counts["split_lines"] += 1 + for call in calls: + normalized_file.write(call + "\n") + counts["calls"] += 1 + + if counts["rejected"] == 0: + reject_path.unlink() + return counts + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("-i", "--input", required=True, help="原始配置文件") + parser.add_argument("-o", "--output", required=True, help="规范化配置文件") + parser.add_argument( + "--rejects", + default=None, + help="拒绝文件路径，默认是 .unparsed.txt", + ) + parser.add_argument( + "--strict", + action="store_true", + help="完成扫描并写出拒绝文件后，对任何坏行返回非零状态", + ) + return parser.parse_args(argv) + + +def main(argv=None): + args = parse_args(argv) + reject_path = args.rejects or f"{args.output}.unparsed.txt" + counts = normalize_file(args.input, args.output, reject_path) + print( + f"规范化: {args.input} -> {args.output}，" + f"输入 {counts['lines']} 行，输出 {counts['calls']} 个调用，" + f"拆分 {counts['split_lines']} 行，拒绝 {counts['rejected']} 行" + ) + if counts["rejected"]: + print(f"警告: 无法拆分的配置已写入 {reject_path}") + if args.strict: + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test_pipeline/config_preprocess/test_to_0_size_config.py b/test_pipeline/config_preprocess/test_to_0_size_config.py new file mode 100644 index 00000000..3c152504 --- /dev/null +++ b/test_pipeline/config_preprocess/test_to_0_size_config.py @@ -0,0 +1,104 @@ +"""验证配置行的顶层调用拆分协议。""" + +# 测试同时覆盖共享结构规范化和 0-size 入口的 APIConfig 语义解析。 +# 重点保证脏数据可定位、可落盘，并且不会阻止后续合法调用被处理。 + +import tempfile +import unittest +from pathlib import Path + +from test_pipeline.config_preprocess.config_lines import split_top_level_calls +from test_pipeline.config_preprocess.normalize_config_lines import normalize_file +from test_pipeline.config_preprocess.to_0_size_config import iter_api_configs + + +class SplitTopLevelCallsTest(unittest.TestCase): + """确保拆分不会改变独立调用的内容和顺序。""" + + def test_split_two_calls_without_separator(self): + text = "paddle.zeros([1])paddle.randn([2])" + self.assertEqual( + split_top_level_calls(text), + ["paddle.zeros([1])", "paddle.randn([2])"], + ) + + def test_nested_parentheses_and_string_are_not_split(self): + # 字符串中的伪调用和右括号不能提前结束外层调用。 + text = 'paddle.foo((1, 2), "paddle.bar())")paddle.zeros([1])' + self.assertEqual( + split_top_level_calls(text), + ['paddle.foo((1, 2), "paddle.bar())")', "paddle.zeros([1])"], + ) + + def test_unbalanced_or_trailing_text_is_rejected(self): + # 无法无损拆分的数据必须失败，不能只保留可识别的前缀。 + with self.assertRaisesRegex(ValueError, "括号不匹配"): + split_top_level_calls("paddle.zeros([1]") + with self.assertRaisesRegex(ValueError, "无法识别"): + split_top_level_calls("paddle.zeros([1]) + 1") + + def test_api_name_scan_does_not_cross_into_next_prefix(self): + with self.assertRaisesRegex(ValueError, "缺少左括号"): + split_top_level_calls("paddle.foopaddle.bar(1)") + + def test_iter_api_configs_parses_each_call(self): + with tempfile.NamedTemporaryFile("w", encoding="utf-8") as config_file: + config_file.write("paddle.zeros([1])paddle.randn([2])\n") + config_file.flush() + configs = list(iter_api_configs(config_file.name)) + self.assertEqual( + [config.api_name for config in configs], + ["paddle.zeros", "paddle.randn"], + ) + + def test_api_config_error_contains_source_line(self): + with tempfile.NamedTemporaryFile("w", encoding="utf-8") as config_file: + config_file.write("paddle.add(Tensor([1]), Tensor([1]))\n") + config_file.flush() + with self.assertRaisesRegex( + ValueError, rf"{config_file.name}:1: 无法解析拆分后的调用" + ): + list(iter_api_configs(config_file.name)) + + def test_api_config_reject_callback_continues_with_later_calls(self): + rejected = [] + with tempfile.NamedTemporaryFile("w", encoding="utf-8") as config_file: + config_file.write( + "paddle.add(Tensor([1]), Tensor([1]))\n" + "paddle.zeros([1])\n" + ) + config_file.flush() + configs = list( + iter_api_configs( + config_file.name, + on_reject=lambda *details: rejected.append(details), + ) + ) + + self.assertEqual([config.api_name for config in configs], ["paddle.zeros"]) + self.assertEqual(rejected[0][1], 1) + self.assertIn("无法解析拆分后的调用", str(rejected[0][3])) + + def test_normalize_file_keeps_valid_calls_and_records_rejects(self): + with tempfile.TemporaryDirectory() as temp_dir: + source = Path(temp_dir) / "input.txt" + output = Path(temp_dir) / "normalized.txt" + rejects = Path(temp_dir) / "normalized.txt.unparsed.txt" + source.write_text( + "paddle.zeros([1])paddle.randn([2])\n" + "paddle.foopaddle.bar(1)\n", + encoding="utf-8", + ) + counts = normalize_file(source, output, rejects) + + self.assertEqual( + output.read_text(encoding="utf-8").splitlines(), + ["paddle.zeros([1])", "paddle.randn([2])"], + ) + self.assertEqual(counts["split_lines"], 1) + self.assertEqual(counts["rejected"], 1) + self.assertIn(":2\t", rejects.read_text(encoding="utf-8")) + + +if __name__ == "__main__": + unittest.main() diff --git a/test_pipeline/config_preprocess/to_0_size_config.py b/test_pipeline/config_preprocess/to_0_size_config.py index c83009fc..c4d62e4c 100644 --- a/test_pipeline/config_preprocess/to_0_size_config.py +++ b/test_pipeline/config_preprocess/to_0_size_config.py @@ -9,6 +9,12 @@ import numpy import paddle +from test_pipeline.config_preprocess.config_lines import ( + matching_close as _matching_close, +) +from test_pipeline.config_preprocess.config_lines import ( + split_top_level_calls as _split_top_level_calls, +) from tester.api_config.parser import APIConfig from tester.input_generation.tensor_config import TensorConfig from tqdm import tqdm @@ -43,40 +49,46 @@ def get_tensor_configs(api_config): return tensor_configs -def iter_api_configs(config_path): - """逐行解析配置，避免把全部 APIConfig 对象同时保存在内存。""" +def iter_api_configs(config_path, on_reject=None): + """解析配置，支持一行中连续的多个顶层 API 调用。""" + # 未提供回调时保持严格失败语义，并附加输入文件与行号。 + # 提供回调时记录坏配置后继续，yield 始终位于异常处理范围之外。 with open(config_path, encoding="utf-8") as config_file: - for raw_line in config_file: + for line_number, raw_line in enumerate(config_file, start=1): config = raw_line.strip() - if config and config.startswith("paddle."): - yield APIConfig(config) - - -def _matching_close(text, start, opening, closing): - """返回嵌套括号的结束位置，字符串内容中的括号不参与配对。""" - # 需要跳过字符串中的括号，否则 callable 或字符串参数会破坏定位。 - depth = 0 - quote = None - escaped = False - for index in range(start, len(text)): - char = text[index] - if quote is not None: - if escaped: - escaped = False - elif char == "\\": - escaped = True - elif char == quote: - quote = None - continue - if char in ('"', "'"): - quote = char - elif char == opening: - depth += 1 - elif char == closing: - depth -= 1 - if depth == 0: - return index - return None + if not config: + continue + if not config.startswith("paddle."): + error = ValueError("配置不是 paddle.* 顶层调用") + if on_reject is None: + raise ValueError( + f"{config_path}:{line_number}: {error}" + ) from error + on_reject(config_path, line_number, config, error) + continue + try: + calls = _split_top_level_calls(config) + except ValueError as error: + if on_reject is None: + raise ValueError(f"{config_path}:{line_number}: {error}") from error + on_reject(config_path, line_number, config, error) + continue + for call in calls: + # APIConfig 失败也必须保留文件和行号，便于定位截断配置。 + try: + api_config = APIConfig(call) + except Exception as error: + parse_error = ValueError( + f"无法解析拆分后的调用 {call[:80]!r}: " + f"{type(error).__name__}: {error}" + ) + if on_reject is None: + raise ValueError( + f"{config_path}:{line_number}: {parse_error}" + ) from error + on_reject(config_path, line_number, call, parse_error) + continue + yield api_config def _find_tensor_shape_spans(config): @@ -441,6 +453,11 @@ def parse_args(): default="api_config_0_size.txt", help="输出文件路径（默认：当前目录下 api_config_0_size.txt）", ) + parser.add_argument( + "--strict", + action="store_true", + help="扫描完成后如有无法解析的配置则返回非零状态", + ) return parser.parse_args() @@ -452,30 +469,48 @@ def parse_args(): os.makedirs(output_dir, exist_ok=True) output_dir = os.path.dirname(args.output) or "." - with tempfile.TemporaryDirectory(prefix=".0size_chunks.", dir=output_dir) as chunk_dir: - # 临时块位于输出目录，保证大文件处理不依赖系统 /tmp 空间。 - chunk_paths = [] - chunk_lines = set() - chunk_bytes = 0 - chunk_index = 0 - for input_file in args.inputs: - print(f"处理: {input_file}") - for api_config in tqdm(iter_api_configs(input_file)): - # 逐个变体进入有限大小的块集合，避免累计完整输出。 - for variant in to_0_size_config(api_config): - if variant in chunk_lines: - continue - chunk_lines.add(variant) - chunk_bytes += len(variant) + 1 - if chunk_bytes >= CHUNK_BYTES: - path = _flush_chunk(chunk_lines, chunk_dir, chunk_index) - chunk_paths.append(path) - chunk_index += 1 - chunk_lines.clear() - chunk_bytes = 0 - path = _flush_chunk(chunk_lines, chunk_dir, chunk_index) - if path is not None: - chunk_paths.append(path) - unique_count = _merge_chunks(chunk_paths, args.output) + reject_path = f"{args.output}.unparsed.txt" + rejected_count = [0] + with open(reject_path, "w", encoding="utf-8") as reject_file: + + def record_reject(config_path, line_number, config, error): + rejected_count[0] += 1 + reject_file.write( + f"{config_path}:{line_number}\t{error}\t{config}\n" + ) + + with tempfile.TemporaryDirectory(prefix=".0size_chunks.", dir=output_dir) as chunk_dir: + # 临时块位于输出目录，保证大文件处理不依赖系统 /tmp 空间。 + chunk_paths = [] + chunk_lines = set() + chunk_bytes = 0 + chunk_index = 0 + for input_file in args.inputs: + print(f"处理: {input_file}") + for api_config in tqdm( + iter_api_configs(input_file, on_reject=record_reject) + ): + # 逐个变体进入有限大小的块集合，避免累计完整输出。 + for variant in to_0_size_config(api_config): + if variant in chunk_lines: + continue + chunk_lines.add(variant) + chunk_bytes += len(variant) + 1 + if chunk_bytes >= CHUNK_BYTES: + path = _flush_chunk(chunk_lines, chunk_dir, chunk_index) + chunk_paths.append(path) + chunk_index += 1 + chunk_lines.clear() + chunk_bytes = 0 + path = _flush_chunk(chunk_lines, chunk_dir, chunk_index) + if path is not None: + chunk_paths.append(path) + unique_count = _merge_chunks(chunk_paths, args.output) print(f"输出: {args.output}，共 {unique_count} 行") + if rejected_count[0]: + print(f"警告: {rejected_count[0]} 条配置无法解析，详见 {reject_path}") + if args.strict: + raise SystemExit(1) + else: + os.remove(reject_path) diff --git a/test_pipeline/run_pipeline.sh b/test_pipeline/run_pipeline.sh index 722f66ed..d9875b04 100755 --- a/test_pipeline/run_pipeline.sh +++ b/test_pipeline/run_pipeline.sh @@ -63,7 +63,9 @@ OUTPUT_DIR="$(cd "$OUTPUT_DIR" && pwd)" PADDLEONLY_1M_DIR="$OUTPUT_DIR/paddleonly_1M" PADDLEONLY_0SIZE_DIR="$OUTPUT_DIR/paddleonly_0size" PADDLEONLY_4096_DIR="$OUTPUT_DIR/paddleonly_4096" -mkdir -p "$PADDLEONLY_1M_DIR" "$PADDLEONLY_0SIZE_DIR" "$PADDLEONLY_4096_DIR" +NORMALIZED_INPUT_DIR="$OUTPUT_DIR/.normalized" +mkdir -p "$PADDLEONLY_1M_DIR" "$PADDLEONLY_0SIZE_DIR" "$PADDLEONLY_4096_DIR" \ + "$NORMALIZED_INPUT_DIR" echo "======================================================================" echo "API Config 全流程处理" @@ -86,6 +88,22 @@ if [ ! -f "$INPUT_DIR/api_config_1024.txt" ] || [ ! -f "$INPUT_DIR/api_config_20 exit 1 fi +# ============================================================================ +# Step 0: 规范化输入，保证后续所有处理器每行只接收一个顶层调用 +# ============================================================================ +echo "" +echo "[Step 0] 规范化原始配置行..." + +for config_name in \ + api_config_1024.txt api_config_2048.txt api_config_4096.txt api_config_8192.txt; do + if [ -f "$INPUT_DIR/$config_name" ]; then + python "$PROCESSOR_DIR/normalize_config_lines.py" \ + -i "$INPUT_DIR/$config_name" \ + -o "$NORMALIZED_INPUT_DIR/$config_name" \ + --strict + fi +done + # ============================================================================ # Step 1: 推导虚假 4096 并验证（如果有真实 4096） # ============================================================================ @@ -97,15 +115,15 @@ DERIVED_4096="$OUTPUT_DIR/.derived_4096.txt" DERIVED_1M="$OUTPUT_DIR/.derived_1M.txt" python "$PROCESSOR_DIR/derive_api_seq.py" 4096 \ - --small "$INPUT_DIR/api_config_1024.txt" \ - --large "$INPUT_DIR/api_config_2048.txt" \ + --small "$NORMALIZED_INPUT_DIR/api_config_1024.txt" \ + --large "$NORMALIZED_INPUT_DIR/api_config_2048.txt" \ -o "$DERIVED_4096" if [ -f "$INPUT_DIR/api_config_4096.txt" ]; then echo "" python "$PROCESSOR_DIR/verify_api_seq.py" \ -d "$DERIVED_4096" \ - -r "$INPUT_DIR/api_config_4096.txt" + -r "$NORMALIZED_INPUT_DIR/api_config_4096.txt" else echo " [跳过验证] 未找到真实 api_config_4096.txt" fi @@ -117,8 +135,8 @@ echo "" echo "[Step 2] 推导 1M (seq=1048576)..." python "$PROCESSOR_DIR/derive_api_seq.py" 1048576 \ - --small "$INPUT_DIR/api_config_1024.txt" \ - --large "$INPUT_DIR/api_config_2048.txt" \ + --small "$NORMALIZED_INPUT_DIR/api_config_1024.txt" \ + --large "$NORMALIZED_INPUT_DIR/api_config_2048.txt" \ -o "$DERIVED_1M" # ============================================================================ @@ -140,7 +158,7 @@ echo "[Step 4] 筛选目标 seq + 去重，并生成 0-size..." ORIG_MERGED_NAME="4096.txt" # 1024/2048 仅作为推导锚点，8192 已废弃；三者不得进入任何最终配置集。 if [ -f "$INPUT_DIR/api_config_4096.txt" ]; then - FINAL_4096_SOURCE="$INPUT_DIR/api_config_4096.txt" + FINAL_4096_SOURCE="$NORMALIZED_INPUT_DIR/api_config_4096.txt" echo " 使用真实 api_config_4096.txt" else FINAL_4096_SOURCE="$DERIVED_4096" @@ -159,7 +177,8 @@ python "$PROCESSOR_DIR/dedup_config.py" \ # 避免对可能达到数 GB 的中间文件再做一次全量去重。 python "$PROCESSOR_DIR/to_0_size_config.py" \ -i "$PADDLEONLY_4096_DIR/$ORIG_MERGED_NAME" \ - -o "$PADDLEONLY_0SIZE_DIR/0size.txt" + -o "$PADDLEONLY_0SIZE_DIR/0size.txt" \ + --strict # 指定比例时只替换保留集；未指定时完全跳过缩减，保持默认流程结果不变。 if [ -n "$SLIM_RATIO" ]; then