diff --git a/Lib/test/test_peg_generator/test_c_generator.py b/Lib/test/test_peg_generator/test_c_generator.py new file mode 100644 index 00000000000000..5b48738c2620a9 --- /dev/null +++ b/Lib/test/test_peg_generator/test_c_generator.py @@ -0,0 +1,281 @@ +import io +import unittest +from unittest import mock + +from test import test_tools + +test_tools.skip_if_missing("peg_generator") +with test_tools.imports_under_tool("peg_generator"): + from pegen import grammar as grammar_module + from pegen.c_generator import CParserGenerator + from pegen.c_generator_file import CParserEmitter + from pegen.grammar import NamedItem, RuleKind + from pegen.grammar_parser import GeneratedParser as GrammarParser + from pegen.testutil import ALL_TOKENS, EXACT_TOKENS, NON_EXACT_TOKENS, parse_string + + +class TestCGenerator(unittest.TestCase): + def make_generator(self, source): + grammar = parse_string(source, GrammarParser) + return CParserGenerator( + grammar, ALL_TOKENS, EXACT_TOKENS, NON_EXACT_TOKENS, io.StringIO() + ) + + def emit_parser(self, parser): + output = io.StringIO() + CParserEmitter(parser, output).emit() + return output.getvalue() + + def test_rule_types_distinguish_implicit_and_explicit_void_pointer(self): + generator = self.make_generator(""" + start: &implicit implicit explicit + implicit: NAME + explicit[void*]: NAME + """) + generator.rules["explicit"].type = "void *" + start, implicit, explicit = generator.prepare("example.gram").rules + + self.assertIsNone(implicit.signature.return_type) + self.assertEqual(explicit.signature.return_type, "void *") + self.assertEqual(implicit.signature.c_return_type, "void *") + self.assertEqual(explicit.signature.c_return_type, "void *") + self.assertEqual( + implicit.signature.declaration(), "static void *implicit_rule(Parser *p);" + ) + self.assertEqual( + explicit.signature.declaration(), "static void * explicit_rule(Parser *p);" + ) + call = start.alternatives[0].calls[0] + self.assertEqual(call.function, "_PyPegen_lookahead") + generator = self.make_generator("start: &explicit\nexplicit[void*]: NAME\n") + with self.assertRaisesRegex(RuntimeError, "return type is incompatible"): + generator.prepare("example.gram") + + def test_parser_plan_does_not_depend_on_compilation_state(self): + generator = self.make_generator(""" + @header 'CUSTOM HEADER' + @subheader 'CUSTOM SUBHEADER' + @trailer 'CUSTOM TRAILER %(modulename)s %(mode)d' + @modulename 'sample' + @bytecode '1' + start[mod_ty]: expr_without_invalid 'pass' "zsoft" "asoft" ('bb' | 'aa')* ENDMARKER + expr_without_invalid[expr_ty] (memo): name=expr [NUMBER] { name } + expr[expr_ty]: expr '+' NAME | NAME + """) + generator.debug = True + parser = generator.prepare("some/path/example.gram") + expected = self.emit_parser(parser) + + self.assertEqual(generator.file.getvalue(), "") + self.assertEqual(parser.source_name, "example.gram") + self.assertEqual(parser.headers, ("CUSTOM HEADER", "CUSTOM SUBHEADER")) + self.assertEqual(parser.trailer, "CUSTOM TRAILER sample 2") + self.assertEqual(parser.soft_keywords, ("asoft", "zsoft")) + self.assertEqual([word for word, _ in parser.keyword_groups[2]], ["bb", "aa"]) + self.assertTrue(any(rule.signature.kind is RuleKind.LOOP0 for rule in parser.rules)) + with self.assertRaises(AttributeError): + parser.rules[0].alternatives[0].calls[0].assigned_variable = "changed" + + generator.grammar.metas.clear() + generator.grammar.metas["trailer"] = "%(missing)s" + for rule in generator.all_rules.values(): + rule.name = "changed" + rule.type = "changed_type" + rule.flags = frozenset() + rule.rhs.alts[0].action = "changed_action" + rule.rhs.alts.clear() + generator.rules.clear() + generator.all_rules.clear() + generator.keywords.clear() + generator.soft_keywords.clear() + generator.debug = False + generator.skip_actions = True + self.assertEqual(self.emit_parser(parser), expected) + + def test_repeated_preparation_keeps_variable_names_local(self): + source = """ + start: 'run' expr term bindings other ENDMARKER + expr: expr '+' NAME | NAME + term: term '*' NUMBER | NUMBER + bindings: (name_var=NUMBER) name_var[expr_ty]=(NAME) [NUMBER] (NAME | NUMBER) { name_var_1 } + other: name_var=NUMBER name_var=NAME { name_var_1 } + """ + generator = self.make_generator(source) + parser = generator.prepare("example.gram") + expected = self.emit_parser(parser) + + self.assertEqual(self.emit_parser(parser), expected) + self.assertEqual(generator.prepare("example.gram"), parser) + other = self.make_generator(source).prepare("example.gram") + self.assertEqual(other, parser) + self.assertEqual(self.emit_parser(other), expected) + self.assertEqual(expected.count("expr_ty name_var_1;"), 2) + self.assertEqual(expected.count("_res = name_var_1;"), 2) + self.assertNotIn("name_var_2", expected) + + def test_prepared_prefixes_preserve_reuse_and_repeatability(self): + generator = self.make_generator(""" + start: prefix ':' NAME | prefix ':' NUMBER | NAME | prefix '=' NAME + prefix[expr_ty] (memo): NAME + """) + parser = generator.prepare("example.gram") + start = parser.rules[0] + prefix, = start.prefixes + self.assertEqual(prefix.type, "expr_ty") + for alt in start.alternatives[:2]: + self.assertIn("!p->call_invalid_rules", alt.calls[0].expression()) + self.assertIn(prefix.result, alt.calls[0].expression()) + self.assertEqual(start.alternatives[3].calls[0].expression(), "prefix_rule(p)") + self.assertEqual(generator.prepare("example.gram"), parser) + expected = self.emit_parser(parser) + generator.rules.clear() + generator.all_rules.clear() + self.assertEqual(self.emit_parser(parser), expected) + + def test_nullable_prefix_is_not_reused(self): + generator = self.make_generator(""" + start: prefix ':' NAME | prefix ':' NUMBER + prefix (memo): NAME? + """) + start = generator.prepare("example.gram").rules[0] + self.assertEqual(start.prefixes, ()) + for alt in start.alternatives: + self.assertEqual(alt.calls[0].expression(), "prefix_rule(p)") + + def test_invalid_trailer_fails_before_output(self): + generator = self.make_generator(""" + @trailer '%(missing)s' + start: NAME ENDMARKER + """) + with self.assertRaisesRegex(KeyError, "missing"): + generator.generate("example.gram") + self.assertEqual(generator.file.getvalue(), "") + + def test_empty_keyword_tables(self): + parser = self.make_generator("start: NAME ENDMARKER\n").prepare("example.gram") + source = self.emit_parser(parser) + + self.assertEqual(parser.keyword_groups, ()) + self.assertEqual(parser.soft_keywords, ()) + self.assertIn("static const int n_keyword_lists = 0;", source) + self.assertIn( + "static KeywordToken *reserved_keywords[] = {\n" + " (KeywordToken[]) {{NULL, -1}},\n" + "};", + source, + ) + self.assertIn("static char *soft_keywords[] = {\n NULL,\n};", source) + + def test_lowering_rejects_undiscovered_items(self): + for replacement in ("missing", "(NAME NUMBER)", None): + with self.subTest(replacement=replacement): + generator = self.make_generator("start: NAME ENDMARKER\n") + generator.collect_rules() + lowerer = generator.callmakervisitor.make_lowerer() + inventory = tuple(generator.all_rules) + counter = generator.counter + rule = generator.rules["start"] + items = rule.rhs.alts[0].items + if replacement is None: + items[0] = NamedItem(None, items[0].item) + else: + grammar = parse_string(f"start: {replacement}\n", GrammarParser) + items[0].item = grammar.rules["start"].rhs.alts[0].items[0].item + with self.assertRaisesRegex(RuntimeError, "not discovered"): + lowerer.prepare_rule(rule) + self.assertEqual(tuple(generator.all_rules), inventory) + self.assertEqual(generator.counter, counter) + + def test_helper_resolution_does_not_depend_on_display_settings(self): + source = """ + start: NAME (a=NAME { a }) NAME* NAME+ ','.NAME+ ENDMARKER + """ + for simple in (True, False): + with self.subTest(simple=simple), mock.patch.object( + grammar_module, "SIMPLE_STR", simple + ): + generator = self.make_generator(source) + expected = generator.prepare("example.gram").rules + with mock.patch.object(grammar_module, "SIMPLE_STR", not simple): + actual = generator.prepare("example.gram").rules + self.assertEqual(len(actual), len(expected)) + for old, new in zip(expected, actual): + self.assertEqual(old.signature, new.signature) + self.assertEqual( + [alt.calls for alt in old.alternatives], + [alt.calls for alt in new.alternatives], + ) + + def test_invalid_rule_gating_uses_references(self): + cases = ( + ("invalid_example", True), + ("value=invalid_example", True), + ("[invalid_example]", True), + ("invalid_example?", True), + ("invalid_example*", True), + ("invalid_example+", True), + ("invalid_example.NAME+", True), + ("[invalid_example.NAME+]", True), + ("[invalid_example.(NAME NAME)+]", False), + ("[[invalid_example.(NAME NAME)+]]", False), + ("[invalid_example.(NAME | NUMBER)+]", False), + ("&invalid_example", False), + ("[invalid_example | NAME]", False), + ("invalid_name=NAME", False), + ) + for item, requires_invalid_rules in cases: + for simple in (True, False): + with self.subTest(item=item, simple=simple), mock.patch.object( + grammar_module, "SIMPLE_STR", simple + ): + generator = self.make_generator(f""" + start: {item} {{ _PyPegen_dummy_name(p) }} + invalid_example: NAME + """) + start = generator.prepare("example.gram").rules[0] + self.assertEqual( + start.alternatives[0].requires_invalid_rules, + requires_invalid_rules, + ) + + def test_lowering_preserves_legacy_named_call_types(self): + generator = self.make_generator(""" + start: Mixed LPAR ENDMARKER + Mixed[expr_ty]: NAME + """) + start, mixed = generator.prepare("example.gram").rules + self.assertEqual(mixed.signature.return_type, "expr_ty") + for call, name in zip(start.alternatives[0].calls, ("Mixed", "LPAR")): + with self.subTest(name=name): + self.assertEqual(call.function, f"{name}_rule") + self.assertIsNone(call.return_type) + + def test_lowering_snapshots_symbols_and_tokens(self): + grammar = parse_string(""" + start: 'pass' '+' atom ENDMARKER + atom[expr_ty]: NAME + """, GrammarParser) + exact_tokens = dict(EXACT_TOKENS) + non_exact_tokens = set(NON_EXACT_TOKENS) + generator = CParserGenerator( + grammar, ALL_TOKENS, exact_tokens, non_exact_tokens, io.StringIO() + ) + generator.collect_rules() + lowerer = generator.callmakervisitor.make_lowerer() + start = generator.rules["start"] + atom = generator.rules["atom"] + expected = lowerer.prepare_rule(start) + + atom.type = "stmt_ty" + generator.all_rules.clear() + generator.tokens.clear() + generator.keywords.clear() + exact_tokens.clear() + non_exact_tokens.clear() + + self.assertEqual(lowerer.prepare_rule(atom).signature.return_type, "expr_ty") + self.assertEqual(lowerer.prepare_rule(start), expected) + + +if __name__ == "__main__": + unittest.main() diff --git a/Lib/test/test_peg_generator/test_c_parser.py b/Lib/test/test_peg_generator/test_c_parser.py index cd0b907667e4a2..c430f648bc1563 100644 --- a/Lib/test/test_peg_generator/test_c_parser.py +++ b/Lib/test/test_peg_generator/test_c_parser.py @@ -239,6 +239,21 @@ def test_negative_lookahead(self) -> None: """ self.run_test(grammar_source, test_source) + def test_optional_gather_with_invalid_separator(self) -> None: + grammar_source = """ + start: 'prefix' guard_without_invalid NAME NEWLINE ENDMARKER + guard_without_invalid: + | [invalid_separator.(NAME NAME)+] { _PyPegen_dummy_name(p) } + invalid_separator: '+' + """ + test_source = """ + self.check_input_strings_for_grammar( + valid_cases=["prefix hello", "prefix a b hello", "prefix a b + c d hello"], + invalid_cases=["prefix", "prefix a b"], + ) + """ + self.run_test(grammar_source, test_source) + def test_cut(self) -> None: grammar_source = """ start: X ~ Y Z | X Q S @@ -404,6 +419,132 @@ def test_same_name_different_types(self) -> None: """ self.run_test(grammar_source, test_source) + def test_alternative_variable_bindings(self) -> None: + grammar_source = """ + start[mod_ty]: a=stmt NEWLINE ENDMARKER { + _PyAST_Module((asdl_stmt_seq *)_PyPegen_singleton_seq(p, a), NULL, p->arena) } + stmt[stmt_ty]: + | &NAME NAME name_var[expr_ty]=NAME NUMBER? { + _PyAST_Expr(name_var_1, EXTRA) } + | &NUMBER name_var=NUMBER name_var[expr_ty]=NAME { + _PyAST_Expr(name_var_1, EXTRA) } + """ + test_source = """ + for source in ("first second", "first second 42", "42 second"): + actual = parse.parse_string(source, mode=1) + self.assertEqual(len(actual.body), 1) + self.assertIsInstance(actual.body[0], ast.Expr) + self.assertIsInstance(actual.body[0].value, ast.Name) + self.assertEqual(actual.body[0].value.id, "second") + """ + self.run_test(grammar_source, test_source) + + def test_rule_cleanup(self) -> None: + grammar_source = """ + @subheader ''' + #define CHECK_INVALID(expected) \\ + (assert(p->call_invalid_rules == (expected)), _PyPegen_dummy_name(p)) + ''' + start: enable (checked_without_invalid '+' | checked_without_invalid after | after) NEWLINE ENDMARKER + enable: 'enable' { (p->call_invalid_rules = 1, _PyPegen_dummy_name(p)) } + checked_without_invalid (memo): "value" ~ NAME { CHECK_INVALID(0) } + after: NAME { CHECK_INVALID(1) } + """ + test_source = """ + self.check_input_strings_for_grammar([ + "enable value name +", # Successful rule return. + "enable value name tail", # Memoized return after backtracking. + "enable fallback", # Failed rule return. + "enable value", # Early return through a cut. + ]) + """ + self.run_test(grammar_source, test_source) + + def test_left_recursive_rule_cleanup(self) -> None: + grammar_source = """ + @subheader ''' + #define CHECK_INVALID(expected) \\ + (assert(p->call_invalid_rules == (expected)), _PyPegen_dummy_name(p)) + ''' + start: enable (expr_without_invalid after | after) NEWLINE ENDMARKER + enable: 'enable' { (p->call_invalid_rules = 1, _PyPegen_dummy_name(p)) } + expr_without_invalid: + | expr_without_invalid '+' NAME { CHECK_INVALID(0) } + | NAME { CHECK_INVALID(0) } + after: NAME { CHECK_INVALID(1) } | NUMBER { CHECK_INVALID(1) } + """ + test_source = """ + self.check_input_strings_for_grammar([ + "enable name tail", + "enable name + other + last tail", + "enable fallback", # Backtrack past a successful recursive rule. + "enable 42", # The recursive rule has no successful alternative. + ]) + """ + self.run_test(grammar_source, test_source) + + def test_nested_rule_cleanup(self) -> None: + grammar_source = """ + @subheader ''' + #define CHECK_INVALID(expected) \\ + (assert(p->call_invalid_rules == (expected)), _PyPegen_dummy_name(p)) + ''' + start: enable outer_without_invalid after NEWLINE ENDMARKER + enable: 'enable' { (p->call_invalid_rules = 1, _PyPegen_dummy_name(p)) } + outer_without_invalid: + | inner_without_invalid '+' { CHECK_INVALID(0) } + | inner_without_invalid inside { CHECK_INVALID(0) } + | inside { CHECK_INVALID(0) } + inner_without_invalid (memo): 'value' NAME { CHECK_INVALID(0) } + inside: NAME { CHECK_INVALID(0) } + after: NAME { CHECK_INVALID(1) } + """ + test_source = """ + self.check_input_strings_for_grammar([ + "enable value name + tail", # Restore the enclosing disabled state. + "enable value name middle tail", # Restore it on a memoized return. + "enable fallback tail", # Restore it when the inner rule fails. + ]) + """ + self.run_test(grammar_source, test_source) + + def test_repetition_result_order(self) -> None: + grammar_source = """ + start[mod_ty]: a=statements NEWLINE ENDMARKER { + _PyAST_Module(a, NULL, p->arena) } + statements[asdl_stmt_seq*]: + | 'repeat0' a=stmt* { (asdl_stmt_seq*)a } + | 'repeat1' a=stmt+ { (asdl_stmt_seq*)a } + | 'gather' a=','.stmt+ { (asdl_stmt_seq*)a } + stmt[stmt_ty]: a=NAME { _PyAST_Expr(a, EXTRA) } + """ + test_source = """ + for mode, separator in (("repeat0", " "), ("repeat1", " "), ("gather", ",")): + for count in (1, 2, 5, 17): + with self.subTest(mode=mode, count=count): + names = ["name" + str(index) for index in range(count)] + result = parse.parse_string(mode + " " + separator.join(names), mode=1) + self.assertEqual([stmt.value.id for stmt in result.body], names) + result = parse.parse_string("repeat0", mode=1) + self.assertEqual(result.body, []) + """ + self.run_test(grammar_source, test_source) + + def test_repetition_action_errors(self) -> None: + grammar_source = """ + start: ('repeat0' item* | 'repeat1' item+ | 'gather' ','.item+) NEWLINE ENDMARKER + item: NAME | 'fail' { PyTuple_New(-1) } + """ + test_source = """ + for mode, separator in (("repeat0", " "), ("repeat1", " "), ("gather", ",")): + for items in (("fail",), ("first", "second", "fail")): + with self.subTest(mode=mode, items=items): + with self.assertRaises(SystemError): + parse.parse_string(mode + " " + separator.join(items), mode=0) + parse.parse_string(mode + " first", mode=0) + """ + self.run_test(grammar_source, test_source) + def test_with_stmt_with_paren(self) -> None: grammar_source = """ start[mod_ty]: a=[statements] ENDMARKER { _PyAST_Module(a, NULL, p->arena) } diff --git a/Lib/test/test_peg_generator/test_pegen.py b/Lib/test/test_peg_generator/test_pegen.py index f39fcc2e0d8daf..64c875367cef27 100644 --- a/Lib/test/test_peg_generator/test_pegen.py +++ b/Lib/test/test_peg_generator/test_pegen.py @@ -12,7 +12,7 @@ with test_tools.imports_under_tool("peg_generator"): from pegen.grammar_parser import GeneratedParser as GrammarParser from pegen.testutil import parse_string, generate_parser, make_parser - from pegen.grammar import GrammarVisitor, GrammarError, Grammar + from pegen.grammar import GrammarVisitor, GrammarError, Grammar, RuleKind from pegen.grammar_visualizer import ASTGrammarPrinter from pegen.parser import Parser from pegen.parser_generator import compute_nullables, compute_left_recursives @@ -537,6 +537,29 @@ def test_advanced_left_recursive(self) -> None: self.assertTrue(rules["start"].left_recursive) self.assertFalse(rules["sign"].left_recursive) + def test_helper_rule_kinds_do_not_depend_on_names(self) -> None: + grammar = parse_string(""" + start: NAME* NUMBER+ ','.NAME+ + """, GrammarParser) + generator = PythonParserGenerator(grammar, io.StringIO()) + generator.collect_rules() + helpers = [ + rule for rule in generator.all_rules.values() + if rule is not grammar.rules["start"] + ] + self.assertCountEqual( + [rule.kind for rule in helpers], + [RuleKind.LOOP0, RuleKind.LOOP1, RuleKind.LOOP0, RuleKind.GATHER], + ) + for rule in helpers: + is_loop, is_gather = rule.is_loop(), rule.is_gather() + rule.name = "renamed" + self.assertEqual(rule.is_loop(), is_loop) + self.assertEqual(rule.is_gather(), is_gather) + grammar.rules["start"].name = "_loop1_name_only" + self.assertFalse(grammar.rules["start"].is_loop()) + self.assertFalse(grammar.rules["start"].is_gather()) + def test_mutually_left_recursive(self) -> None: grammar_source = """ start: foo 'E' diff --git a/Tools/peg_generator/pegen/c_generator.py b/Tools/peg_generator/pegen/c_generator.py index 044366c3aac140..b2978d3a0baa75 100644 --- a/Tools/peg_generator/pegen/c_generator.py +++ b/Tools/peg_generator/pegen/c_generator.py @@ -1,32 +1,23 @@ -import ast +"""Prepare a complete C parser description and emit it.""" + import os.path -import re -from collections.abc import Callable -from dataclasses import dataclass, field -from enum import Enum -from typing import IO, Any +from typing import IO from pegen import grammar -from pegen.grammar import ( - Alt, - Cut, - Forced, - Gather, - GrammarVisitor, - Group, - Leaf, - Lookahead, - NamedItem, - NameLeaf, - NegativeLookahead, - Opt, - PositiveLookahead, - Repeat0, - Repeat1, - Rhs, - Rule, - StringLeaf, +from pegen.c_generator_calls import ( + CCallMakerVisitor as CCallMakerVisitor, +) +from pegen.c_generator_calls import ( + FunctionCall as FunctionCall, +) +from pegen.c_generator_calls import ( + NodeTypes as NodeTypes, +) +from pegen.c_generator_calls import ( + consuming_rules as consuming_rules, ) +from pegen.c_generator_file import CParserEmitter +from pegen.c_generator_model import CParser, CRule, CRuleSignature from pegen.parser_generator import ParserGenerator EXTENSION_PREFIX = """\ @@ -66,345 +57,7 @@ """ -class NodeTypes(Enum): - NAME_TOKEN = 0 - NUMBER_TOKEN = 1 - STRING_TOKEN = 2 - GENERIC_TOKEN = 3 - KEYWORD = 4 - SOFT_KEYWORD = 5 - CUT_OPERATOR = 6 - F_STRING_CHUNK = 7 - - -BASE_NODETYPES = { - "NAME": NodeTypes.NAME_TOKEN, - "NUMBER": NodeTypes.NUMBER_TOKEN, - "STRING": NodeTypes.STRING_TOKEN, - "SOFT_KEYWORD": NodeTypes.SOFT_KEYWORD, -} - - -@dataclass -class FunctionCall: - function: str - arguments: list[Any] = field(default_factory=list) - assigned_variable: str | None = None - assigned_variable_type: str | None = None - return_type: str | None = None - nodetype: NodeTypes | None = None - force_true: bool = False - comment: str | None = None - - def __str__(self) -> str: - parts = [] - parts.append(self.function) - if self.arguments: - parts.append(f"({', '.join(map(str, self.arguments))})") - if self.force_true: - parts.append(", !p->error_indicator") - if self.assigned_variable: - if self.assigned_variable_type: - parts = [ - "(", - self.assigned_variable, - " = ", - "(", - self.assigned_variable_type, - ")", - *parts, - ")", - ] - else: - parts = ["(", self.assigned_variable, " = ", *parts, ")"] - if self.comment: - parts.append(f" // {self.comment}") - return "".join(parts) - - -class CCallMakerVisitor(GrammarVisitor): - def __init__( - self, - parser_generator: ParserGenerator, - exact_tokens: dict[str, int], - non_exact_tokens: set[str], - ): - self.gen = parser_generator - self.exact_tokens = exact_tokens - self.non_exact_tokens = non_exact_tokens - self.cache: dict[str, str] = {} - self.cleanup_statements: list[str] = [] - - def keyword_helper(self, keyword: str) -> FunctionCall: - return FunctionCall( - assigned_variable="_keyword", - function="_PyPegen_expect_token", - arguments=["p", self.gen.keywords[keyword]], - return_type="Token *", - nodetype=NodeTypes.KEYWORD, - comment=f"token='{keyword}'", - ) - - def soft_keyword_helper(self, value: str) -> FunctionCall: - return FunctionCall( - assigned_variable="_keyword", - function="_PyPegen_expect_soft_keyword", - arguments=["p", value], - return_type="expr_ty", - nodetype=NodeTypes.SOFT_KEYWORD, - comment=f"soft_keyword='{value}'", - ) - - def visit_NameLeaf(self, node: NameLeaf) -> FunctionCall: - name = node.value - if name in self.non_exact_tokens: - if name in BASE_NODETYPES: - return FunctionCall( - assigned_variable=f"{name.lower()}_var", - function=f"_PyPegen_{name.lower()}_token", - arguments=["p"], - nodetype=BASE_NODETYPES[name], - return_type="expr_ty", - comment=name, - ) - return FunctionCall( - assigned_variable=f"{name.lower()}_var", - function="_PyPegen_expect_token", - arguments=["p", name], - nodetype=NodeTypes.GENERIC_TOKEN, - return_type="Token *", - comment=f"token='{name}'", - ) - - type = None - rule = self.gen.all_rules.get(name.lower()) - if rule is not None: - type = "asdl_seq *" if rule.is_loop() or rule.is_gather() else rule.type - - return FunctionCall( - assigned_variable=f"{name}_var", - function=f"{name}_rule", - arguments=["p"], - return_type=type, - comment=f"{node}", - ) - - def visit_StringLeaf(self, node: StringLeaf) -> FunctionCall: - val = ast.literal_eval(node.value) - if re.match(r"[a-zA-Z_]\w*\Z", val): # This is a keyword - if node.value.endswith("'"): - return self.keyword_helper(val) - else: - return self.soft_keyword_helper(node.value) - else: - assert val in self.exact_tokens, f"{node.value} is not a known literal" - type = self.exact_tokens[val] - return FunctionCall( - assigned_variable="_literal", - function="_PyPegen_expect_token", - arguments=["p", type], - nodetype=NodeTypes.GENERIC_TOKEN, - return_type="Token *", - comment=f"token='{val}'", - ) - - def visit_NamedItem(self, node: NamedItem) -> FunctionCall: - call = self.generate_call(node.item) - if node.name: - call.assigned_variable = node.name - if node.type: - call.assigned_variable_type = node.type - return call - - def assert_no_undefined_behavior( - self, call: FunctionCall, wrapper: str, expected_rtype: str | None, - ) -> None: - if call.return_type != expected_rtype: - raise RuntimeError( - f"{call.function} return type is incompatible with {wrapper}: " - f"expect: {expected_rtype}, actual: {call.return_type}" - ) - - def lookahead_call_helper(self, node: Lookahead, positive: int) -> FunctionCall: - call = self.generate_call(node.node) - comment = None - if call.nodetype is NodeTypes.NAME_TOKEN: - function = "_PyPegen_lookahead_for_expr" - self.assert_no_undefined_behavior(call, function, "expr_ty") - elif call.nodetype is NodeTypes.STRING_TOKEN: - # _PyPegen_string_token() returns 'void *' instead of 'Token *'; - # in addition, the overall function call would return 'expr_ty'. - assert call.function == "_PyPegen_string_token" - function = "_PyPegen_lookahead" - self.assert_no_undefined_behavior(call, function, "expr_ty") - elif call.nodetype == NodeTypes.SOFT_KEYWORD: - function = "_PyPegen_lookahead_with_string" - self.assert_no_undefined_behavior(call, function, "expr_ty") - elif call.nodetype in {NodeTypes.GENERIC_TOKEN, NodeTypes.KEYWORD}: - function = "_PyPegen_lookahead_with_int" - self.assert_no_undefined_behavior(call, function, "Token *") - comment = f"token={node.node}" - elif call.return_type == "expr_ty": - function = "_PyPegen_lookahead_for_expr" - elif call.return_type == "stmt_ty": - function = "_PyPegen_lookahead_for_stmt" - else: - function = "_PyPegen_lookahead" - self.assert_no_undefined_behavior(call, function, None) - return FunctionCall( - function=function, - arguments=[positive, call.function, *call.arguments], - return_type="int", - comment=comment, - ) - - def visit_PositiveLookahead(self, node: PositiveLookahead) -> FunctionCall: - return self.lookahead_call_helper(node, 1) - - def visit_NegativeLookahead(self, node: NegativeLookahead) -> FunctionCall: - return self.lookahead_call_helper(node, 0) - - def visit_Forced(self, node: Forced) -> FunctionCall: - call = self.generate_call(node.node) - if isinstance(node.node, Leaf): - assert isinstance(node.node, Leaf) - val = ast.literal_eval(node.node.value) - assert val in self.exact_tokens, f"{node.node.value} is not a known literal" - type = self.exact_tokens[val] - return FunctionCall( - assigned_variable="_literal", - function="_PyPegen_expect_forced_token", - arguments=["p", type, f'"{val}"'], - nodetype=NodeTypes.GENERIC_TOKEN, - return_type="Token *", - comment=f"forced_token='{val}'", - ) - if isinstance(node.node, Group): - call = self.visit(node.node.rhs) - call.assigned_variable = None - call.comment = None - return FunctionCall( - assigned_variable="_literal", - function="_PyPegen_expect_forced_result", - arguments=["p", str(call), f'"{node.node.rhs!s}"'], - return_type="void *", - comment=f"forced_token=({node.node.rhs!s})", - ) - else: - raise NotImplementedError(f"Forced tokens don't work with {node.node} nodes") - - def visit_Opt(self, node: Opt) -> FunctionCall: - call = self.generate_call(node.node) - return FunctionCall( - assigned_variable="_opt_var", - function=call.function, - arguments=call.arguments, - force_true=True, - comment=f"{node}", - ) - - def _generate_artificial_rule_call( - self, - node: Any, - prefix: str, - rule_generation_func: Callable[[], str], - return_type: str | None = None, - ) -> FunctionCall: - node_str = f"{node}" - key = f"{prefix}_{node_str}" - if key in self.cache: - name = self.cache[key] - else: - name = rule_generation_func() - self.cache[key] = name - - return FunctionCall( - assigned_variable=f"{name}_var", - function=f"{name}_rule", - arguments=["p"], - return_type=return_type, - comment=node_str, - ) - - def visit_Rhs(self, node: Rhs) -> FunctionCall: - if node.can_be_inlined: - return self.generate_call(node.alts[0].items[0]) - - return self._generate_artificial_rule_call( - node, - "rhs", - lambda: self.gen.artificial_rule_from_rhs(node), - ) - - def visit_Repeat0(self, node: Repeat0) -> FunctionCall: - return self._generate_artificial_rule_call( - node, - "repeat0", - lambda: self.gen.artificial_rule_from_repeat(node.node, is_repeat1=False), - "asdl_seq *", - ) - - def visit_Repeat1(self, node: Repeat1) -> FunctionCall: - return self._generate_artificial_rule_call( - node, - "repeat1", - lambda: self.gen.artificial_rule_from_repeat(node.node, is_repeat1=True), - "asdl_seq *", - ) - - def visit_Gather(self, node: Gather) -> FunctionCall: - return self._generate_artificial_rule_call( - node, - "gather", - lambda: self.gen.artificial_rule_from_gather(node), - "asdl_seq *", - ) - - def visit_Group(self, node: Group) -> FunctionCall: - return self.generate_call(node.rhs) - - def visit_Cut(self, node: Cut) -> FunctionCall: - return FunctionCall( - assigned_variable="_cut_var", - return_type="int", - function="1", - nodetype=NodeTypes.CUT_OPERATOR, - ) - - def generate_call(self, node: Any) -> FunctionCall: - return super().visit(node) - - -def consuming_rules(rules: dict[str, Rule]) -> set[str]: - """Conservatively prove which rules consume a token whenever they succeed.""" - consuming: set[str] = set() - - def consumes(node: Any) -> bool: - if isinstance(node, NamedItem): - return consumes(node.item) - if isinstance(node, NameLeaf): - return node.value not in rules or node.value in consuming - if isinstance(node, StringLeaf): - return True - if isinstance(node, Group): - return consumes(node.rhs) - if isinstance(node, Rhs): - return bool(node.alts) and all(any(consumes(i) for i in alt.items) for alt in node.alts) - if isinstance(node, (Forced, Repeat1, Gather)): - return consumes(node.node) - # Predicates, cuts, optional items, and zero-or-more items can succeed - # without consuming. Actions are assumed not to rewrite parser marks. - return False - - while True: - added = {name for name, rule in rules.items() - if name not in consuming and consumes(rule.rhs)} - if not added: - return consuming - consuming.update(added) - - -class CParserGenerator(ParserGenerator, GrammarVisitor): +class CParserGenerator(ParserGenerator): def __init__( self, grammar: grammar.Grammar, @@ -419,551 +72,63 @@ def __init__( self.callmakervisitor: CCallMakerVisitor = CCallMakerVisitor( self, exact_tokens, non_exact_tokens ) - self._varname_counter = 0 + self._collected = False self.debug = debug self.skip_actions = skip_actions - self.cleanup_statements: list[str] = [] - self.consuming = consuming_rules(self.rules) - self.prefix_calls: dict[int, tuple[str, str, str]] = {} - - def add_level(self) -> None: - self.print("if (p->level++ == MAXSTACK || _PyPegen_stack_exhausted(p)) {") - with self.indent(): - self.print("_Pypegen_stack_overflow(p);") - self.print("}") - - def remove_level(self) -> None: - self.print("p->level--;") - - def add_return(self, ret_val: str) -> None: - for stmt in self.cleanup_statements: - self.print(stmt) - self.remove_level() - self.print(f"return {ret_val};") - - def unique_varname(self, name: str = "tmpvar") -> str: - new_var = name + "_" + str(self._varname_counter) - self._varname_counter += 1 - return new_var - - def call_with_errorcheck_return(self, call_text: str, returnval: str) -> None: - error_var = self.unique_varname() - self.print(f"int {error_var} = {call_text};") - self.print(f"if ({error_var}) {{") - with self.indent(): - self.add_return(returnval) - self.print("}") - - def call_with_errorcheck_goto(self, call_text: str, goto_target: str) -> None: - error_var = self.unique_varname() - self.print(f"int {error_var} = {call_text};") - self.print(f"if ({error_var}) {{") - with self.indent(): - self.print(f"goto {goto_target};") - self.print("}") - - def out_of_memory_return( - self, - expr: str, - cleanup_code: str | None = None, - ) -> None: - self.print(f"if ({expr}) {{") - with self.indent(): - if cleanup_code is not None: - self.print(cleanup_code) - self.print("p->error_indicator = 1;") - self.print("PyErr_NoMemory();") - self.add_return("NULL") - self.print("}") - - def out_of_memory_goto(self, expr: str, goto_target: str) -> None: - self.print(f"if ({expr}) {{") - with self.indent(): - self.print("PyErr_NoMemory();") - self.print(f"goto {goto_target};") - self.print("}") def generate(self, filename: str) -> None: + parser = self.prepare(filename) + CParserEmitter(parser, self.file).emit() + + def prepare(self, filename: str) -> CParser: self.collect_rules() - basename = os.path.basename(filename) - self.print(f"// @generated by pegen from {basename}") - header = self.grammar.metas.get("header", EXTENSION_PREFIX) - if header: - self.print(header.rstrip("\n")) - subheader = self.grammar.metas.get("subheader", "") - if subheader: - self.print(subheader) - self._setup_keywords() - self._setup_soft_keywords() - for i, (rulename, rule) in enumerate(self.all_rules.items(), 1000): - comment = " // Left-recursive" if rule.left_recursive else "" - self.print(f"#define {rulename}_type {i}{comment}") - self.print() - for rulename, rule in self.all_rules.items(): - if rule.is_loop() or rule.is_gather(): - type = "asdl_seq *" - elif rule.type: - type = rule.type + " " - else: - type = "void *" - self.print(f"static {type}{rulename}_rule(Parser *p);") - self.print() - for rulename, rule in list(self.all_rules.items()): - self.print() - if rule.left_recursive: - self.print("// Left-recursive") - self.visit(rule) + lowerer = self.callmakervisitor.make_lowerer() + rules = tuple( + lowerer.prepare_rule(rule, skip_actions=self.skip_actions) + for rule in self.all_rules.values() + ) + headers = [] + if header := self.grammar.metas.get("header", EXTENSION_PREFIX): + headers.append(header.rstrip("\n")) + if subheader := self.grammar.metas.get("subheader", ""): + headers.append(subheader) + return CParser( + source_name=os.path.basename(filename), + headers=tuple(headers), + keyword_groups=self._prepare_keywords(), + soft_keywords=tuple(sorted(self.soft_keywords)), + rules=rules, + trailer=self._prepare_trailer(rules), + debug=self.debug, + ) + + def collect_rules(self) -> None: + # Keyword generation also uses this entry point without emitting C. + if not self._collected: + super().collect_rules() + self._collected = True + + def _prepare_keywords(self) -> tuple[tuple[tuple[str, int], ...], ...]: + if not self.keywords: + return () + groups: list[list[tuple[str, int]]] = [ + [] for _ in range(max(map(len, self.keywords)) + 1) + ] + for keyword, token_type in self.keywords.items(): + groups[len(keyword)].append((keyword, token_type)) + return tuple(tuple(group) for group in groups) + + def _prepare_trailer(self, rules: tuple[CRule, ...]) -> str | None: if self.skip_actions: mode = 0 else: - mode = int(self.rules["start"].type == "mod_ty") if "start" in self.rules else 1 - if mode == 1 and self.grammar.metas.get("bytecode"): - mode += 1 + start = next((rule.signature for rule in rules if rule.signature.name == "start"), None) + match start: + case None | CRuleSignature(return_type="mod_ty"): + mode = 2 if self.grammar.metas.get("bytecode") else 1 + case _: + mode = 0 modulename = self.grammar.metas.get("modulename", "parse") - trailer = self.grammar.metas.get("trailer", EXTENSION_SUFFIX) - if trailer: - self.print(trailer.rstrip("\n") % dict(mode=mode, modulename=modulename)) - - def _group_keywords_by_length(self) -> dict[int, list[tuple[str, int]]]: - groups: dict[int, list[tuple[str, int]]] = {} - for keyword_str, keyword_type in self.keywords.items(): - length = len(keyword_str) - if length in groups: - groups[length].append((keyword_str, keyword_type)) - else: - groups[length] = [(keyword_str, keyword_type)] - return groups - - def _setup_keywords(self) -> None: - n_keyword_lists = ( - len(max(self.keywords.keys(), key=len)) + 1 if len(self.keywords) > 0 else 0 - ) - self.print(f"static const int n_keyword_lists = {n_keyword_lists};") - groups = self._group_keywords_by_length() - self.print("static KeywordToken *reserved_keywords[] = {") - with self.indent(): - num_groups = max(groups) + 1 if groups else 1 - for keywords_length in range(num_groups): - if keywords_length not in groups.keys(): - self.print("(KeywordToken[]) {{NULL, -1}},") - else: - self.print("(KeywordToken[]) {") - with self.indent(): - for keyword_str, keyword_type in groups[keywords_length]: - self.print(f'{{"{keyword_str}", {keyword_type}}},') - self.print("{NULL, -1},") - self.print("},") - self.print("};") - - def _setup_soft_keywords(self) -> None: - soft_keywords = sorted(self.soft_keywords) - self.print("static char *soft_keywords[] = {") - with self.indent(): - for keyword in soft_keywords: - self.print(f'"{keyword}",') - self.print("NULL,") - self.print("};") - - def _set_up_token_start_metadata_extraction(self) -> None: - self.print("if (p->mark == p->fill && _PyPegen_fill_token(p) < 0) {") - with self.indent(): - self.print("p->error_indicator = 1;") - self.add_return("NULL") - self.print("}") - self.print("int _start_lineno = p->tokens[_mark]->lineno;") - self.print("UNUSED(_start_lineno); // Only used by EXTRA macro") - self.print("int _start_col_offset = p->tokens[_mark]->col_offset;") - self.print("UNUSED(_start_col_offset); // Only used by EXTRA macro") - - def _set_up_token_end_metadata_extraction(self) -> None: - self.print("Token *_token = _PyPegen_get_last_nonnwhitespace_token(p);") - self.print("if (_token == NULL) {") - with self.indent(): - self.add_return("NULL") - self.print("}") - self.print("int _end_lineno = _token->end_lineno;") - self.print("UNUSED(_end_lineno); // Only used by EXTRA macro") - self.print("int _end_col_offset = _token->end_col_offset;") - self.print("UNUSED(_end_col_offset); // Only used by EXTRA macro") - - def _check_for_errors(self) -> None: - self.print("if (p->error_indicator) {") - with self.indent(): - self.add_return("NULL") - self.print("}") - - def _set_up_rule_memoization(self, node: Rule, result_type: str) -> None: - self.print("{") - with self.indent(): - self.add_level() - self.print(f"{result_type} _res = NULL;") - self.print(f"if (_PyPegen_is_memoized(p, {node.name}_type, &_res)) {{") - with self.indent(): - self.add_return("_res") - self.print("}") - self.print("int _mark = p->mark;") - self.print("int _resmark = p->mark;") - self.print(f"Memo *_memo = _PyPegen_insert_memo_direct(p, _mark, {node.name}_type);") - self.print("if (_memo == NULL) {") - with self.indent(): - self.add_return("NULL") - self.print("}") - self.print("while (1) {") - with self.indent(): - self.print("_memo->node = _res;") - self.print("_memo->mark = p->mark;") - self.print("p->mark = _mark;") - self.print(f"void *_raw = {node.name}_raw(p);") - self.print("if (p->error_indicator) {") - with self.indent(): - self.add_return("NULL") - self.print("}") - self.print("if (_raw == NULL || p->mark <= _resmark)") - with self.indent(): - self.print("break;") - self.print("_resmark = p->mark;") - self.print("_res = _raw;") - self.print("}") - self.print("p->mark = _resmark;") - self.add_return("_res") - self.print("}") - self.print(f"static {result_type}") - self.print(f"{node.name}_raw(Parser *p)") - - def _should_memoize(self, node: Rule) -> bool: - return "memo" in node.flags and not node.left_recursive - - def _handle_default_rule_body(self, node: Rule, rhs: Rhs, result_type: str) -> None: - memoize = self._should_memoize(node) - - with self.indent(): - self.add_level() - self._check_for_errors() - self.print(f"{result_type} _res = NULL;") - if memoize: - self.print(f"if (_PyPegen_is_memoized(p, {node.name}_type, &_res)) {{") - with self.indent(): - self.add_return("_res") - self.print("}") - self.print("int _mark = p->mark;") - self.prepare_prefix_calls(rhs) - if any(alt.action and "EXTRA" in alt.action for alt in rhs.alts): - self._set_up_token_start_metadata_extraction() - self.visit( - rhs, - is_loop=False, - is_gather=node.is_gather(), - rulename=node.name, - ) - if self.debug: - self.print(f'D(fprintf(stderr, "Fail at %d: {node.name}\\n", p->mark));') - self.print("_res = NULL;") - self.print(" done:") - with self.indent(): - if memoize: - self.print(f"_PyPegen_insert_memo(p, _mark, {node.name}_type, _res);") - self.add_return("_res") - - def _handle_loop_rule_body(self, node: Rule, rhs: Rhs) -> None: - memoize = self._should_memoize(node) - is_repeat1 = node.name.startswith("_loop1") - - with self.indent(): - self.add_level() - self._check_for_errors() - self.print("void *_res = NULL;") - if memoize: - self.print(f"if (_PyPegen_is_memoized(p, {node.name}_type, &_res)) {{") - with self.indent(): - self.add_return("_res") - self.print("}") - self.print("int _mark = p->mark;") - if memoize: - self.print("int _start_mark = p->mark;") - self.print("void **_children = PyMem_Malloc(sizeof(void *));") - self.out_of_memory_return("!_children") - self.print("Py_ssize_t _children_capacity = 1;") - self.print("Py_ssize_t _n = 0;") - if any(alt.action and "EXTRA" in alt.action for alt in rhs.alts): - self._set_up_token_start_metadata_extraction() - self.visit( - rhs, - is_loop=True, - is_gather=node.is_gather(), - rulename=node.name, - ) - if is_repeat1: - self.print("if (_n == 0 || p->error_indicator) {") - with self.indent(): - self.print("PyMem_Free(_children);") - self.add_return("NULL") - self.print("}") - self.print("asdl_seq *_seq = (asdl_seq*)_Py_asdl_generic_seq_new(_n, p->arena);") - self.out_of_memory_return("!_seq", cleanup_code="PyMem_Free(_children);") - self.print("for (Py_ssize_t i = 0; i < _n; i++) asdl_seq_SET_UNTYPED(_seq, i, _children[i]);") - self.print("PyMem_Free(_children);") - if memoize and node.name: - self.print(f"_PyPegen_insert_memo(p, _start_mark, {node.name}_type, _seq);") - self.add_return("_seq") - - def prepare_prefix_calls(self, rhs: Rhs) -> None: - # Reuse a memoized, consuming prefix only within a consecutive group. - # Suffix parsing starts after the prefix and cannot revisit its start - # through ordinary grammar backtracking. Diagnostic calls are unchanged. - def candidate(alt: Alt) -> Rule | None: - if not alt.items or not isinstance(alt.items[0].item, NameLeaf): - return None - rule = self.rules.get(alt.items[0].item.value) - if rule is None or rule.name not in self.consuming: - return None - if self._should_memoize(rule) or (rule.left_recursive and rule.leader): - return rule - return None - - i = 0 - while i < len(rhs.alts): - rule = candidate(rhs.alts[i]) - j = i + 1 - while rule is not None and j < len(rhs.alts) and candidate(rhs.alts[j]) is rule: - j += 1 - if rule is not None and j - i > 1: - name = self.unique_varname("_prefix") - result, end, valid = name + "_result", name + "_end", name + "_valid" - self.print(f"{rule.type or 'void *'} {result} = NULL;") - self.print(f"int {end} = 0, {valid} = 0;") - for alt in rhs.alts[i:j]: - self.prefix_calls[id(alt.items[0])] = result, end, valid - i = j - - def visit_Rule(self, node: Rule) -> None: - self.prefix_calls = {} - is_loop = node.is_loop() - is_gather = node.is_gather() - rhs = node.flatten() - if is_loop or is_gather: - result_type = "asdl_seq *" - elif node.type: - result_type = node.type - else: - result_type = "void *" - - for line in str(node).splitlines(): - self.print(f"// {line}") - if node.left_recursive and node.leader: - self.print(f"static {result_type} {node.name}_raw(Parser *);") - - self.print(f"static {result_type}") - self.print(f"{node.name}_rule(Parser *p)") - - if node.left_recursive and node.leader: - self._set_up_rule_memoization(node, result_type) - - self.print("{") - - if node.name.endswith("without_invalid"): - with self.indent(): - self.print("int _prev_call_invalid = p->call_invalid_rules;") - self.print("p->call_invalid_rules = 0;") - self.cleanup_statements.append("p->call_invalid_rules = _prev_call_invalid;") - - if is_loop: - self._handle_loop_rule_body(node, rhs) - else: - self._handle_default_rule_body(node, rhs, result_type) - - if node.name.endswith("without_invalid"): - self.cleanup_statements.pop() - - self.print("}") - - def visit_NamedItem(self, node: NamedItem) -> None: - call = self.callmakervisitor.generate_call(node) - if id(node) in self.prefix_calls: - result, end, valid = self.prefix_calls[id(node)] - original = f"{call.function}({', '.join(map(str, call.arguments))})" - call.function = ( - f"((!p->call_invalid_rules && {valid}) ? " - f"(p->mark = {end}, {result}) : " - f"({result} = {original}, {end} = p->mark, {valid} = 1, {result}))" - ) - call.arguments = [] - if call.assigned_variable: - call.assigned_variable = self.dedupe(call.assigned_variable) - self.print(call) - - def visit_Rhs( - self, node: Rhs, is_loop: bool, is_gather: bool, rulename: str | None - ) -> None: - if is_loop: - assert len(node.alts) == 1 - for alt in node.alts: - self.visit(alt, is_loop=is_loop, is_gather=is_gather, rulename=rulename) - - def join_conditions(self, keyword: str, node: Any) -> None: - self.print(f"{keyword} (") - with self.indent(): - first = True - for item in node.items: - if first: - first = False - else: - self.print("&&") - self.visit(item) - self.print(")") - - def emit_action(self, node: Alt, cleanup_code: str | None = None) -> None: - self.print(f"_res = {node.action};") - - self.print("if ((_res == NULL || p->error_indicator) && PyErr_Occurred()) {") - with self.indent(): - self.print("p->error_indicator = 1;") - if cleanup_code: - self.print(cleanup_code) - self.add_return("NULL") - self.print("}") - - if self.debug: - self.print( - f'D(fprintf(stderr, "Hit with action [%d-%d]: %s\\n", _mark, p->mark, "{node}"));' - ) - - def emit_default_action(self, is_gather: bool, node: Alt) -> None: - if len(self.local_variable_names) > 1: - if is_gather: - assert len(self.local_variable_names) == 2 - self.print( - f"_res = _PyPegen_seq_insert_in_front(p, " - f"{self.local_variable_names[0]}, {self.local_variable_names[1]});" - ) - else: - if self.debug: - self.print( - f'D(fprintf(stderr, "Hit without action [%d:%d]: %s\\n", _mark, p->mark, "{node}"));' - ) - self.print( - f"_res = _PyPegen_dummy_name(p, {', '.join(self.local_variable_names)});" - ) - else: - if self.debug: - self.print( - f'D(fprintf(stderr, "Hit with default action [%d:%d]: %s\\n", _mark, p->mark, "{node}"));' - ) - self.print(f"_res = {self.local_variable_names[0]};") - - def emit_dummy_action(self) -> None: - self.print("_res = _PyPegen_dummy_name(p);") - - def handle_alt_normal(self, node: Alt, is_gather: bool, rulename: str | None) -> None: - self.join_conditions(keyword="if", node=node) - self.print("{") - # We have parsed successfully all the conditions for the option. - with self.indent(): - node_str = str(node).replace('"', '\\"') - self.print( - f'D(fprintf(stderr, "%*c+ {rulename}[%d-%d]: %s succeeded!\\n", p->level, \' \', _mark, p->mark, "{node_str}"));' - ) - # Prepare to emit the rule action and do so - if node.action and "EXTRA" in node.action: - self._set_up_token_end_metadata_extraction() - if self.skip_actions: - self.emit_dummy_action() - elif node.action: - self.emit_action(node) - else: - self.emit_default_action(is_gather, node) - - # As the current option has parsed correctly, do not continue with the rest. - self.print("goto done;") - self.print("}") - - def handle_alt_loop(self, node: Alt, is_gather: bool, rulename: str | None) -> None: - # Condition of the main body of the alternative - self.join_conditions(keyword="while", node=node) - self.print("{") - # We have parsed successfully one item! - with self.indent(): - # Prepare to emit the rule action and do so - if node.action and "EXTRA" in node.action: - self._set_up_token_end_metadata_extraction() - if self.skip_actions: - self.emit_dummy_action() - elif node.action: - self.emit_action(node, cleanup_code="PyMem_Free(_children);") - else: - self.emit_default_action(is_gather, node) - - # Add the result of rule to the temporary buffer of children. This buffer - # will populate later an asdl_seq with all elements to return. - self.print("if (_n == _children_capacity) {") - with self.indent(): - self.print("_children_capacity *= 2;") - self.print( - "void **_new_children = PyMem_Realloc(_children, _children_capacity*sizeof(void *));" - ) - self.out_of_memory_return("!_new_children", cleanup_code="PyMem_Free(_children);") - self.print("_children = _new_children;") - self.print("}") - self.print("_children[_n++] = _res;") - self.print("_mark = p->mark;") - self.print("}") - - def visit_Alt( - self, node: Alt, is_loop: bool, is_gather: bool, rulename: str | None - ) -> None: - if len(node.items) == 1 and str(node.items[0]).startswith("invalid_"): - self.print(f"if (p->call_invalid_rules) {{ // {node}") - else: - self.print(f"{{ // {node}") - with self.indent(): - self._check_for_errors() - node_str = str(node).replace('"', '\\"') - self.print( - f'D(fprintf(stderr, "%*c> {rulename}[%d-%d]: %s\\n", p->level, \' \', _mark, p->mark, "{node_str}"));' - ) - # Prepare variable declarations for the alternative - vars = self.collect_vars(node) - for v, var_type in sorted(item for item in vars.items() if item[0] is not None): - if not var_type: - var_type = "void *" - else: - var_type += " " - if v == "_cut_var": - v += " = 0" # cut_var must be initialized - self.print(f"{var_type}{v};") - if v and v.startswith("_opt_var"): - self.print(f"UNUSED({v}); // Silence compiler warnings") - - with self.local_variable_context(): - if is_loop: - self.handle_alt_loop(node, is_gather, rulename) - else: - self.handle_alt_normal(node, is_gather, rulename) - - self.print("p->mark = _mark;") - node_str = str(node).replace('"', '\\"') - self.print( - f"D(fprintf(stderr, \"%*c%s {rulename}[%d-%d]: %s failed!\\n\", p->level, ' ',\n" - f' p->error_indicator ? "ERROR!" : "-", _mark, p->mark, "{node_str}"));' - ) - if "_cut_var" in vars: - self.print("if (_cut_var) {") - with self.indent(): - self.add_return("NULL") - self.print("}") - self.print("}") - - def collect_vars(self, node: Alt) -> dict[str | None, str | None]: - types = {} - with self.local_variable_context(): - for item in node.items: - name, type = self.add_var(item) - types[name] = type - return types - - def add_var(self, node: NamedItem) -> tuple[str | None, str | None]: - call = self.callmakervisitor.generate_call(node.item) - name = node.name if node.name else call.assigned_variable - if name is not None: - name = self.dedupe(name) - return_type = call.return_type if node.type is None else node.type - return name, return_type + if trailer := self.grammar.metas.get("trailer", EXTENSION_SUFFIX): + return trailer.rstrip("\n") % dict(mode=mode, modulename=modulename) + return None diff --git a/Tools/peg_generator/pegen/c_generator_calls.py b/Tools/peg_generator/pegen/c_generator_calls.py new file mode 100644 index 00000000000000..dc3c0cca130412 --- /dev/null +++ b/Tools/peg_generator/pegen/c_generator_calls.py @@ -0,0 +1,553 @@ +"""Discover calls and helper rules, then prepare rules from resolved calls.""" + +import ast +import re +from collections.abc import Mapping +from dataclasses import replace +from types import MappingProxyType +from typing import TYPE_CHECKING, Any + +from pegen.c_generator_model import ( + CAction, + CAlternative, + CBindingKind, + CPrefix, + CRule, + CRuleSignature, + CVariable, +) +from pegen.c_generator_model import ( + FunctionCall as FunctionCall, +) +from pegen.c_generator_model import ( + NodeTypes as NodeTypes, +) +from pegen.grammar import ( + Alt, + Cut, + Forced, + Gather, + GrammarVisitor, + Group, + Item, + Leaf, + Lookahead, + NamedItem, + NameLeaf, + NegativeLookahead, + Opt, + PositiveLookahead, + Repeat0, + Repeat1, + Rhs, + Rule, + RuleKind, + StringLeaf, +) + +if TYPE_CHECKING: + from pegen.parser_generator import ParserGenerator + + +BASE_NODETYPES = { + "NAME": NodeTypes.NAME_TOKEN, + "NUMBER": NodeTypes.NUMBER_TOKEN, + "STRING": NodeTypes.STRING_TOKEN, + "SOFT_KEYWORD": NodeTypes.SOFT_KEYWORD, +} + +_HelperNode = Rhs | Repeat0 | Repeat1 | Gather + + +def rule_signature(rule: Rule) -> CRuleSignature: + return_type = rule.type if rule.kind is RuleKind.NORMAL else "asdl_seq *" + return CRuleSignature(rule.name, rule.kind, return_type) + + +def bind_call(node: NamedItem, call: FunctionCall) -> FunctionCall: + if not node.name and not node.type: + return call + return replace( + call, + assigned_variable=node.name or call.assigned_variable, + assigned_variable_type=node.type or call.assigned_variable_type, + binding_kind=CBindingKind.NORMAL if node.name else call.binding_kind, + ) + + +def consuming_rules(rules: dict[str, Rule]) -> set[str]: + """Conservatively prove which rules consume a token whenever they succeed.""" + consuming: set[str] = set() + + def consumes(node: Any) -> bool: + if isinstance(node, NamedItem): + return consumes(node.item) + if isinstance(node, NameLeaf): + return node.value not in rules or node.value in consuming + if isinstance(node, StringLeaf): + return True + if isinstance(node, Group): + return consumes(node.rhs) + if isinstance(node, Rhs): + return bool(node.alts) and all(any(consumes(i) for i in alt.items) for alt in node.alts) + if isinstance(node, (Forced, Repeat1, Gather)): + return consumes(node.node) + # Predicates, cuts, optional items, and zero-or-more items can succeed + # without consuming. Actions are assumed not to rewrite parser marks. + return False + + while True: + added = {name for name, rule in rules.items() + if name not in consuming and consumes(rule.rhs)} + if not added: + return consuming + consuming.update(added) + + +class CCallMakerVisitor(GrammarVisitor): + def __init__( + self, + parser_generator: "ParserGenerator", + exact_tokens: dict[str, int], + non_exact_tokens: set[str], + ): + self._registry = parser_generator + self._keywords = parser_generator.keywords + self._exact_tokens = exact_tokens + self._non_exact_tokens = non_exact_tokens + self._helper_cache: dict[tuple[type, str], str] = {} + self._calls: dict[NamedItem, tuple[Item, FunctionCall]] = {} + + def visit(self, node: Any, *args: Any, **kwargs: Any) -> FunctionCall: + match node: + case NamedItem(item=item): + call = self.visit(item) + self._calls[node] = (item, call) + return bind_call(node, call) + case NameLeaf(): + return self._name_call(node) + case StringLeaf(): + return self._string_call(node) + case PositiveLookahead(): + return self._lookahead_call(node, 1) + case NegativeLookahead(): + return self._lookahead_call(node, 0) + case Forced(): + return self._forced_call(node) + case Opt(): + return self._optional_call(node) + case Rhs(can_be_inlined=True): + return self.visit(node.alts[0].items[0]) + case Rhs() | Repeat0() | Repeat1() | Gather(): + return self._helper_call(node) + case Group(rhs=rhs): + return self.visit(rhs) + case Cut(): + return FunctionCall( + assigned_variable="_cut_var", + return_type="int", + function="1", + nodetype=NodeTypes.CUT_OPERATOR, + binding_kind=CBindingKind.CUT, + ) + case _: + return self.generic_visit(node, *args, **kwargs) + + def _keyword_call(self, keyword: str) -> FunctionCall: + return FunctionCall( + assigned_variable="_keyword", + function="_PyPegen_expect_token", + arguments=("p", self._keywords[keyword]), + return_type="Token *", + nodetype=NodeTypes.KEYWORD, + comment=f"token='{keyword}'", + ) + + def _soft_keyword_call(self, value: str) -> FunctionCall: + return FunctionCall( + assigned_variable="_keyword", + function="_PyPegen_expect_soft_keyword", + arguments=("p", value), + return_type="expr_ty", + nodetype=NodeTypes.SOFT_KEYWORD, + comment=f"soft_keyword='{value}'", + ) + + def _name_call(self, node: NameLeaf) -> FunctionCall: + name = node.value + if name in self._non_exact_tokens: + if name in BASE_NODETYPES: + return FunctionCall( + assigned_variable=f"{name.lower()}_var", + function=f"_PyPegen_{name.lower()}_token", + arguments=("p",), + nodetype=BASE_NODETYPES[name], + return_type="expr_ty", + comment=name, + ) + return FunctionCall( + assigned_variable=f"{name.lower()}_var", + function="_PyPegen_expect_token", + arguments=("p", name), + nodetype=NodeTypes.GENERIC_TOKEN, + return_type="Token *", + comment=f"token='{name}'", + ) + + type = None + if (signature := self._lookup_rule(name)) is not None: + type = signature.return_type + + return FunctionCall( + assigned_variable=f"{name}_var", + function=f"{name}_rule", + arguments=("p",), + return_type=type, + comment=f"{node}", + ) + + def _string_call(self, node: StringLeaf) -> FunctionCall: + val = ast.literal_eval(node.value) + if re.match(r"[a-zA-Z_]\w*\Z", val): # This is a keyword + if node.value.endswith("'"): + return self._keyword_call(val) + else: + return self._soft_keyword_call(node.value) + else: + assert val in self._exact_tokens, f"{node.value} is not a known literal" + type = self._exact_tokens[val] + return FunctionCall( + assigned_variable="_literal", + function="_PyPegen_expect_token", + arguments=("p", type), + nodetype=NodeTypes.GENERIC_TOKEN, + return_type="Token *", + comment=f"token='{val}'", + ) + + def _assert_compatible_return_type( + self, call: FunctionCall, wrapper: str, expected_rtype: str | None, + ) -> None: + if call.return_type != expected_rtype: + raise RuntimeError( + f"{call.function} return type is incompatible with {wrapper}: " + f"expect: {expected_rtype}, actual: {call.return_type}" + ) + + def _lookahead_call(self, node: Lookahead, positive: int) -> FunctionCall: + call = self.visit(node.node) + comment = None + match call: + case FunctionCall(nodetype=NodeTypes.NAME_TOKEN): + function = "_PyPegen_lookahead_for_expr" + self._assert_compatible_return_type(call, function, "expr_ty") + case FunctionCall(nodetype=NodeTypes.STRING_TOKEN): + # _PyPegen_string_token() returns 'void *' instead of 'Token *'; + # in addition, the overall function call would return 'expr_ty'. + assert call.function == "_PyPegen_string_token" + function = "_PyPegen_lookahead" + self._assert_compatible_return_type(call, function, "expr_ty") + case FunctionCall(nodetype=NodeTypes.SOFT_KEYWORD): + function = "_PyPegen_lookahead_with_string" + self._assert_compatible_return_type(call, function, "expr_ty") + case FunctionCall(nodetype=NodeTypes.GENERIC_TOKEN | NodeTypes.KEYWORD): + function = "_PyPegen_lookahead_with_int" + self._assert_compatible_return_type(call, function, "Token *") + comment = f"token={node.node}" + case FunctionCall(return_type="expr_ty"): + function = "_PyPegen_lookahead_for_expr" + case FunctionCall(return_type="stmt_ty"): + function = "_PyPegen_lookahead_for_stmt" + case _: + function = "_PyPegen_lookahead" + self._assert_compatible_return_type(call, function, None) + return FunctionCall( + function=function, + arguments=(positive, call.function, *call.arguments), + return_type="int", + comment=comment, + ) + + def _forced_call(self, node: Forced) -> FunctionCall: + call = self.visit(node.node) + match node.node: + case Leaf(value=value): + val = ast.literal_eval(value) + assert val in self._exact_tokens, f"{value} is not a known literal" + return FunctionCall( + assigned_variable="_literal", + function="_PyPegen_expect_forced_token", + arguments=("p", self._exact_tokens[val], f'"{val}"'), + nodetype=NodeTypes.GENERIC_TOKEN, + return_type="Token *", + comment=f"forced_token='{val}'", + ) + case Group(rhs=rhs): + return FunctionCall( + assigned_variable="_literal", + function="_PyPegen_expect_forced_result", + arguments=("p", call.expression(), f'"{rhs!s}"'), + return_type="void *", + comment=f"forced_token=({rhs!s})", + ) + case _: + raise NotImplementedError(f"Forced tokens don't work with {node.node} nodes") + + def _optional_call(self, node: Opt) -> FunctionCall: + call = self.visit(node.node) + return FunctionCall( + assigned_variable="_opt_var", + function=call.function, + arguments=call.arguments, + force_true=True, + comment=f"{node}", + binding_kind=CBindingKind.OPTIONAL, + ) + + def _helper_call( + self, + node: _HelperNode, + ) -> FunctionCall: + node_str = f"{node}" + signature = self._resolve_artificial_rule(node) + name = signature.name + return FunctionCall( + assigned_variable=f"{name}_var", + function=f"{name}_rule", + arguments=("p",), + return_type=signature.return_type, + comment=node_str, + ) + + def _lookup_rule(self, name: str) -> CRuleSignature | None: + if (rule := self._registry.all_rules.get(name.lower())) is not None: + return rule_signature(rule) + return None + + def _resolve_artificial_rule(self, node: _HelperNode) -> CRuleSignature: + # Preserve helper reuse and numbering from the fixed-point traversal. + key = (type(node), str(node)) + if (name := self._helper_cache.get(key)) is None: + match node: + case Rhs(): + name = self._registry.artificial_rule_from_rhs(node) + case Repeat0(node=child): + name = self._registry.artificial_rule_from_repeat(child, is_repeat1=False) + case Repeat1(node=child): + name = self._registry.artificial_rule_from_repeat(child, is_repeat1=True) + case Gather(): + name = self._registry.artificial_rule_from_gather(node) + self._helper_cache[key] = name + return rule_signature(self._registry.all_rules[name]) + + def make_lowerer(self) -> "CCallLowerer": + return CCallLowerer( + calls=self._calls, + rules=self._registry.all_rules, + original_rules=self._registry.rules, + signatures={ + name: rule_signature(rule) for name, rule in self._registry.all_rules.items() + }, + ) + + +class CCallLowerer: + """Resolve bindings, actions and control flow without registering rules. + + Discovery and lowering operate on the same, unchanged grammar. + """ + + def __init__( + self, + *, + calls: Mapping[NamedItem, tuple[Item, FunctionCall]], + rules: Mapping[str, Rule], + original_rules: dict[str, Rule], + signatures: Mapping[str, CRuleSignature], + ): + self._calls = MappingProxyType(dict(calls)) + self._signatures = MappingProxyType(dict(signatures)) + self._prefixes: dict[str, tuple[CPrefix, ...]] = {} + self._prefix_calls: dict[NamedItem, CPrefix] = {} + consuming = consuming_rules(original_rules) + counter = 0 + + def candidate(alt: Alt) -> Rule | None: + if not alt.items or not isinstance(alt.items[0].item, NameLeaf): + return None + rule = original_rules.get(alt.items[0].item.value) + if rule is None or rule.name not in consuming: + return None + if ("memo" in rule.flags and not rule.left_recursive) or ( + rule.left_recursive and rule.leader + ): + return rule + return None + + for rule in rules.values(): + if rule.kind in {RuleKind.LOOP0, RuleKind.LOOP1}: + continue + # Reuse a consuming prefix only within a consecutive group. + # Diagnostic calls still invoke the original rule. + prefixes = [] + alts = rule.flatten().alts + i = 0 + while i < len(alts): + prefix_rule = candidate(alts[i]) + j = i + 1 + while prefix_rule is not None and j < len(alts) and candidate(alts[j]) is prefix_rule: + j += 1 + if prefix_rule is not None and j - i > 1: + prefix = CPrefix(f"_prefix_{counter}", prefix_rule.type or "void *") + counter += 1 + prefixes.append(prefix) + for alt in alts[i:j]: + self._prefix_calls[alt.items[0]] = prefix + i = j + self._prefixes[rule.name] = tuple(prefixes) + + def prepare_rule(self, rule: Rule, *, skip_actions: bool = False) -> CRule: + if (signature := self._signatures.get(rule.name)) is None: + raise RuntimeError(f"Rule {rule.name!r} was not discovered") + rhs = rule.flatten() + if signature.kind in {RuleKind.LOOP0, RuleKind.LOOP1}: + assert len(rhs.alts) == 1 + return CRule( + signature=signature, + text=str(rule), + alternatives=tuple( + self.prepare_alt(alt, kind=signature.kind, skip_actions=skip_actions) + for alt in rhs.alts + ), + left_recursive=rule.left_recursive, + leader=rule.leader, + memoize="memo" in rule.flags and not rule.left_recursive, + disable_invalid_rules=rule.name.endswith("without_invalid"), + prefixes=self._prefixes.get(rule.name, ()), + ) + + def prepare_alt( + self, + node: Alt, + *, + kind: RuleKind = RuleKind.NORMAL, + skip_actions: bool = False, + ) -> CAlternative: + calls: list[FunctionCall] = [] + variables: dict[str, CVariable] = {} + cut_variable = None + for item in node.items: + recorded = self._calls.get(item) + if recorded is None or recorded[0] is not item.item: + raise RuntimeError(f"Item {item} was not discovered") + call = bind_call(item, recorded[1]) + if (prefix := self._prefix_calls.get(item)) is not None: + result, end, valid = prefix.result, prefix.end, prefix.valid + original = call.expression() + call = replace( + call, + function=( + f"((!p->call_invalid_rules && {valid}) ? " + f"(p->mark = {end}, {result}) : " + f"({result} = {original}, {end} = p->mark, {valid} = 1, {result}))" + ), + arguments=(), + ) + if original_name := call.assigned_variable: + name = original_name + counter = 0 + while name in variables: + counter += 1 + name = f"{original_name}_{counter}" + if name != original_name: + call = replace(call, assigned_variable=name) + initializer = ( + "0" if call.binding_kind is CBindingKind.CUT and cut_variable is None + else None + ) + if initializer is not None: + cut_variable = name + variables[name] = CVariable( + name=name, + type=call.return_type if item.type is None else item.type, + initializer=initializer, + unused=call.binding_kind is CBindingKind.OPTIONAL, + ) + calls.append(call) + return CAlternative( + text=str(node), + action=self._prepare_action(node, list(variables), kind, skip_actions), + calls=tuple(calls), + variables=tuple(variables.values()), + cut_variable=cut_variable, + requires_invalid_rules=self._requires_invalid_rules(node), + uses_locations=bool(node.action and "EXTRA" in node.action), + ) + + def _requires_invalid_rules(self, node: Alt) -> bool: + match node.items: + case [NamedItem(item=item)]: + pass + case _: + return False + # Preserve the source convention for bare, optional and repeated + # invalid references, including an invalid gather separator. + while True: + match item: + case Rhs(alts=[Alt(items=[NamedItem(item=child)])]): + item = child + case Gather(separator=separator): + item = separator + case Opt(node=child) | Repeat0(node=child) | Repeat1(node=child): + # A compound optional may match empty, so an invalid + # reference inside it must not gate the whole alternative. + if self._is_compound(child): + return False + item = child + case NameLeaf(value=name): + return name.startswith("invalid_") + case _: + return False + + def _is_compound(self, item: Item) -> bool: + match item: + case Rhs(alts=alts): + return len(alts) > 1 or any( + len(alt.items) > 1 or any(self._is_compound(part.item) for part in alt.items) + for alt in alts + ) + case Group(rhs=rhs): + return self._is_compound(rhs) + case Gather(separator=separator, node=child): + return self._is_compound(separator) or self._is_compound(child) + case ( + Opt(node=child) | Repeat0(node=child) | Repeat1(node=child) + | Forced(node=child) | Lookahead(node=child) + ): + return self._is_compound(child) + case StringLeaf(value=value): + return " " in value + case _: + return False + + @staticmethod + def _prepare_action( + node: Alt, names: list[str], kind: RuleKind, skip_actions: bool, + ) -> CAction: + if skip_actions: + return CAction("_PyPegen_dummy_name(p)") + if action := node.action: + return CAction( + action, checked=True, debug_message="Hit with action [%d-%d]: %s", + ) + match names: + case [first, rest] if kind is RuleKind.GATHER: + return CAction(f"_PyPegen_seq_insert_in_front(p, {first}, {rest})") + case [_, _, *_]: + assert kind is not RuleKind.GATHER + return CAction( + f"_PyPegen_dummy_name(p, {', '.join(names)})", + debug_message="Hit without action [%d:%d]: %s", + ) + case _: + return CAction(names[0], debug_message="Hit with default action [%d:%d]: %s") diff --git a/Tools/peg_generator/pegen/c_generator_file.py b/Tools/peg_generator/pegen/c_generator_file.py new file mode 100644 index 00000000000000..4ce9345d5f5c82 --- /dev/null +++ b/Tools/peg_generator/pegen/c_generator_file.py @@ -0,0 +1,82 @@ +"""Emit a complete C parser from an immutable plan and local output state.""" + +from collections.abc import Iterator +from contextlib import contextmanager +from typing import IO + +from pegen.c_generator_model import CParser +from pegen.c_generator_rules import CRuleEmitter + + +class CParserEmitter: + def __init__(self, parser: CParser, file: IO[str] | None): + self._parser = parser + self._file = file + self._level = 0 + + def emit(self) -> None: + self._emit_headers() + self._emit_keywords() + self._emit_declarations() + self._emit_rules() + if self._parser.trailer is not None: + self.print(self._parser.trailer) + + def _emit_headers(self) -> None: + self.print(f"// @generated by pegen from {self._parser.source_name}") + for header in self._parser.headers: + self.print(header) + + def _emit_keywords(self) -> None: + groups = self._parser.keyword_groups + self.print(f"static const int n_keyword_lists = {len(groups)};") + self.print("static KeywordToken *reserved_keywords[] = {") + with self.indent(): + for group in groups or ((),): + if not group: + self.print("(KeywordToken[]) {{NULL, -1}},") + else: + self.print("(KeywordToken[]) {") + with self.indent(): + for keyword, token_type in group: + self.print(f'{{"{keyword}", {token_type}}},') + self.print("{NULL, -1},") + self.print("},") + self.print("};") + self.print("static char *soft_keywords[] = {") + with self.indent(): + for keyword in self._parser.soft_keywords: + self.print(f'"{keyword}",') + self.print("NULL,") + self.print("};") + + def _emit_declarations(self) -> None: + for rule_id, rule in enumerate(self._parser.rules, 1000): + comment = " // Left-recursive" if rule.left_recursive else "" + self.print(f"#define {rule.signature.name}_type {rule_id}{comment}") + self.print() + for rule in self._parser.rules: + self.print(rule.signature.declaration()) + self.print() + + def _emit_rules(self) -> None: + for rule in self._parser.rules: + self.print() + if rule.left_recursive: + self.print("// Left-recursive") + CRuleEmitter(self, rule, debug=self._parser.debug).emit() + + def print(self, *args: object) -> None: + if not args: + print(file=self._file) + else: + print(" " * self._level, end="", file=self._file) + print(*args, file=self._file) + + @contextmanager + def indent(self) -> Iterator[None]: + self._level += 1 + try: + yield + finally: + self._level -= 1 diff --git a/Tools/peg_generator/pegen/c_generator_model.py b/Tools/peg_generator/pegen/c_generator_model.py new file mode 100644 index 00000000000000..7b25977d70efb5 --- /dev/null +++ b/Tools/peg_generator/pegen/c_generator_model.py @@ -0,0 +1,142 @@ +"""Immutable C parser descriptions shared by lowering and emission.""" + +from dataclasses import dataclass +from enum import Enum, auto + +from pegen.grammar import RuleKind + + +class NodeTypes(Enum): + NAME_TOKEN = 0 + NUMBER_TOKEN = 1 + STRING_TOKEN = 2 + GENERIC_TOKEN = 3 + KEYWORD = 4 + SOFT_KEYWORD = 5 + CUT_OPERATOR = 6 + F_STRING_CHUNK = 7 + + +class CBindingKind(Enum): + NORMAL = auto() + OPTIONAL = auto() + CUT = auto() + + +@dataclass(frozen=True, slots=True) +class FunctionCall: + function: str + arguments: tuple[str | int, ...] = () + assigned_variable: str | None = None + assigned_variable_type: str | None = None + return_type: str | None = None + nodetype: NodeTypes | None = None + force_true: bool = False + comment: str | None = None + binding_kind: CBindingKind = CBindingKind.NORMAL + + def expression(self) -> str: + """Render the invocation without its alternative-local binding or comment.""" + expression = self.function + if arguments := self.arguments: + expression += f"({', '.join(map(str, arguments))})" + if self.force_true: + expression += ", !p->error_indicator" + return expression + + def __str__(self) -> str: + expression = self.expression() + if variable := self.assigned_variable: + cast = f"({self.assigned_variable_type})" if self.assigned_variable_type else "" + expression = f"({variable} = {cast}{expression})" + if comment := self.comment: + expression += f" // {comment}" + return expression + + +@dataclass(frozen=True, slots=True) +class CRuleSignature: + name: str + kind: RuleKind + return_type: str | None + + @property + def c_return_type(self) -> str: + return self.return_type or "void *" + + def declaration(self) -> str: + separator = " " if self.kind is RuleKind.NORMAL and self.return_type else "" + return f"static {self.c_return_type}{separator}{self.name}_rule(Parser *p);" + + +@dataclass(frozen=True, slots=True) +class CVariable: + name: str + type: str | None + initializer: str | None = None + unused: bool = False + + +@dataclass(frozen=True, slots=True) +class CAction: + expression: str + checked: bool = False + debug_message: str | None = None + + +@dataclass(frozen=True, slots=True) +class CAlternative: + text: str + action: CAction + calls: tuple[FunctionCall, ...] + variables: tuple[CVariable, ...] + cut_variable: str | None + requires_invalid_rules: bool + uses_locations: bool + + +@dataclass(frozen=True, slots=True) +class CPrefix: + name: str + type: str + + @property + def result(self) -> str: + return self.name + "_result" + + @property + def end(self) -> str: + return self.name + "_end" + + @property + def valid(self) -> str: + return self.name + "_valid" + + +@dataclass(frozen=True, slots=True) +class CRule: + signature: CRuleSignature + text: str + alternatives: tuple[CAlternative, ...] + left_recursive: bool + leader: bool + memoize: bool + disable_invalid_rules: bool + prefixes: tuple[CPrefix, ...] = () + + @property + def uses_locations(self) -> bool: + return any(alt.uses_locations for alt in self.alternatives) + + +@dataclass(frozen=True, slots=True) +class CParser: + """Complete file-emission input, independent of compilation state.""" + + source_name: str + headers: tuple[str, ...] + keyword_groups: tuple[tuple[tuple[str, int], ...], ...] + soft_keywords: tuple[str, ...] + rules: tuple[CRule, ...] + trailer: str | None + debug: bool diff --git a/Tools/peg_generator/pegen/c_generator_rules.py b/Tools/peg_generator/pegen/c_generator_rules.py new file mode 100644 index 00000000000000..6dd598beb1b39c --- /dev/null +++ b/Tools/peg_generator/pegen/c_generator_rules.py @@ -0,0 +1,363 @@ +"""Render prepared C rules with rule-local output and cleanup state.""" + +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager +from dataclasses import dataclass +from typing import Protocol + +from pegen.c_generator_model import CAlternative, CRule +from pegen.grammar import RuleKind + + +class CWriter(Protocol): + def print(self, *args: object) -> None: + ... + + def indent(self) -> AbstractContextManager[None]: + ... + + +@dataclass(frozen=True, slots=True) +class _CReturnEmitter: + writer: CWriter + cleanups: tuple[str, ...] = () + + def with_cleanup(self, cleanup: str) -> "_CReturnEmitter": + return _CReturnEmitter(self.writer, (cleanup, *self.cleanups)) + + def emit(self, value: str) -> None: + for cleanup in self.cleanups: + self.writer.print(cleanup) + self.writer.print("p->level--;") + self.writer.print(f"return {value};") + + def check_memory(self, expr: str) -> None: + self.writer.print(f"if ({expr}) {{") + with self.writer.indent(): + self.no_memory() + self.writer.print("}") + + def no_memory(self) -> None: + self.writer.print("p->error_indicator = 1;") + self.writer.print("PyErr_NoMemory();") + self.emit("NULL") + + +class _LoopBuffer: + """Temporary repetition storage and the exits that release it.""" + + _release = "PyMem_Free(_children);" + + def __init__(self, writer: CWriter, returns: _CReturnEmitter): + self._print = writer.print + self._indent = writer.indent + self._returns = returns + self.error_returns = returns.with_cleanup(self._release) + + def initialize(self) -> None: + self._print("void **_children = PyMem_Malloc(sizeof(void *));") + self._returns.check_memory("!_children") + self._print("Py_ssize_t _children_capacity = 1;") + self._print("Py_ssize_t _n = 0;") + + def append(self, value: str) -> None: + self._print("if (_n == _children_capacity) {") + with self._indent(): + self._print("_children_capacity *= 2;") + self._print( + "void **_new_children = PyMem_Realloc(_children, _children_capacity*sizeof(void *));" + ) + self._check_memory("!_new_children") + self._print("_children = _new_children;") + self._print("}") + self._print(f"_children[_n++] = {value};") + + def finish(self, *, require_one: bool) -> str: + if require_one: + self._print("if (_n == 0 || p->error_indicator) {") + with self._indent(): + self.error_returns.emit("NULL") + self._print("}") + self._print("asdl_seq *_seq = (asdl_seq*)_Py_asdl_generic_seq_new(_n, p->arena);") + self._check_memory("!_seq") + self._print("for (Py_ssize_t i = 0; i < _n; i++) asdl_seq_SET_UNTYPED(_seq, i, _children[i]);") + self._print(self._release) + return "_seq" + + def _check_memory(self, expr: str) -> None: + self._print(f"if ({expr}) {{") + with self._indent(): + self._print(self._release) + self._returns.no_memory() + self._print("}") + + +class CRuleEmitter: + def __init__(self, writer: CWriter, rule: CRule, *, debug: bool = False): + self._writer = writer + self._print = writer.print + self._indent = writer.indent + self._rule = rule + self._debug = debug + self._returns = _CReturnEmitter(writer) + + def emit(self) -> None: + rule = self._rule + signature = rule.signature + result_type = signature.c_return_type + for line in rule.text.splitlines(): + self._print(f"// {line}") + if rule.left_recursive and rule.leader: + self._print(f"static {result_type} {signature.name}_raw(Parser *);") + self._print(f"static {result_type}") + self._print(f"{signature.name}_rule(Parser *p)") + if rule.left_recursive and rule.leader: + self._emit_left_recursive_wrapper() + self._print("{") + with self._invalid_rule_context(): + match signature.kind: + case RuleKind.LOOP0 | RuleKind.LOOP1: + self._emit_loop_body() + case RuleKind.NORMAL | RuleKind.GATHER: + self._emit_rule_body() + self._print("}") + + @contextmanager + def _invalid_rule_context(self) -> Iterator[None]: + if not self._rule.disable_invalid_rules: + yield + return + with self._indent(): + self._print("int _prev_call_invalid = p->call_invalid_rules;") + self._print("p->call_invalid_rules = 0;") + previous_returns = self._returns + self._returns = previous_returns.with_cleanup( + "p->call_invalid_rules = _prev_call_invalid;" + ) + try: + yield + finally: + self._returns = previous_returns + + def _emit_left_recursive_wrapper(self) -> None: + signature = self._rule.signature + result_type = signature.c_return_type + self._print("{") + with self._indent(): + self._emit_recursion_check() + self._print(f"{result_type} _res = NULL;") + self._print(f"if (_PyPegen_is_memoized(p, {signature.name}_type, &_res)) {{") + with self._indent(): + self._returns.emit("_res") + self._print("}") + self._print("int _mark = p->mark;") + self._print("int _resmark = p->mark;") + self._print(f"Memo *_memo = _PyPegen_insert_memo_direct(p, _mark, {signature.name}_type);") + self._print("if (_memo == NULL) {") + with self._indent(): + self._returns.emit("NULL") + self._print("}") + self._print("while (1) {") + with self._indent(): + self._print("_memo->node = _res;") + self._print("_memo->mark = p->mark;") + self._print("p->mark = _mark;") + self._print(f"void *_raw = {signature.name}_raw(p);") + self._print("if (p->error_indicator) {") + with self._indent(): + self._returns.emit("NULL") + self._print("}") + self._print("if (_raw == NULL || p->mark <= _resmark)") + with self._indent(): + self._print("break;") + self._print("_resmark = p->mark;") + self._print("_res = _raw;") + self._print("}") + self._print("p->mark = _resmark;") + self._returns.emit("_res") + self._print("}") + self._print(f"static {result_type}") + self._print(f"{signature.name}_raw(Parser *p)") + + def _emit_rule_body(self) -> None: + signature = self._rule.signature + memoize = self._rule.memoize + result_type = signature.c_return_type + + with self._indent(): + self._emit_recursion_check() + self._emit_error_check() + self._print(f"{result_type} _res = NULL;") + if memoize: + self._print(f"if (_PyPegen_is_memoized(p, {signature.name}_type, &_res)) {{") + with self._indent(): + self._returns.emit("_res") + self._print("}") + self._print("int _mark = p->mark;") + for prefix in self._rule.prefixes: + self._print(f"{prefix.type} {prefix.result} = NULL;") + self._print(f"int {prefix.end} = 0, {prefix.valid} = 0;") + if self._rule.uses_locations: + self._emit_token_start_metadata() + for alt in self._rule.alternatives: + with self._alternative(alt): + self._emit_normal_alt(alt) + if self._debug: + self._print(f'D(fprintf(stderr, "Fail at %d: {signature.name}\\n", p->mark));') + self._print("_res = NULL;") + self._print(" done:") + with self._indent(): + if memoize: + self._print(f"_PyPegen_insert_memo(p, _mark, {signature.name}_type, _res);") + self._returns.emit("_res") + + def _emit_loop_body(self) -> None: + rule = self._rule + signature = rule.signature + buffer = _LoopBuffer(self._writer, self._returns) + with self._indent(): + self._emit_recursion_check() + self._emit_error_check() + self._print("void *_res = NULL;") + if rule.memoize: + self._print(f"if (_PyPegen_is_memoized(p, {signature.name}_type, &_res)) {{") + with self._indent(): + self._returns.emit("_res") + self._print("}") + self._print("int _mark = p->mark;") + if rule.memoize: + self._print("int _start_mark = p->mark;") + buffer.initialize() + if rule.uses_locations: + self._emit_token_start_metadata() + alt, = rule.alternatives + with self._alternative(alt): + self._emit_loop_alt(alt, buffer) + result = buffer.finish(require_one=signature.kind is RuleKind.LOOP1) + if rule.memoize: + self._print(f"_PyPegen_insert_memo(p, _start_mark, {signature.name}_type, {result});") + self._returns.emit(result) + + @contextmanager + def _alternative(self, alt: CAlternative) -> Iterator[None]: + rulename = self._rule.signature.name + if alt.requires_invalid_rules: + self._print(f"if (p->call_invalid_rules) {{ // {alt.text}") + else: + self._print(f"{{ // {alt.text}") + with self._indent(): + self._emit_error_check() + node_str = alt.text.replace('"', '\\"') + self._print( + f'D(fprintf(stderr, "%*c> {rulename}[%d-%d]: %s\\n", p->level, \' \', _mark, p->mark, "{node_str}"));' + ) + for variable in sorted(alt.variables, key=lambda var: var.name): + ctype = variable.type + " " if variable.type else "void *" + initializer = ( + f" = {variable.initializer}" if variable.initializer is not None else "" + ) + self._print(f"{ctype}{variable.name}{initializer};") + if variable.unused: + self._print(f"UNUSED({variable.name}); // Silence compiler warnings") + + yield + + self._print("p->mark = _mark;") + self._print( + f"D(fprintf(stderr, \"%*c%s {rulename}[%d-%d]: %s failed!\\n\", p->level, ' ',\n" + f' p->error_indicator ? "ERROR!" : "-", _mark, p->mark, "{node_str}"));' + ) + if alt.cut_variable is not None: + self._print(f"if ({alt.cut_variable}) {{") + with self._indent(): + self._returns.emit("NULL") + self._print("}") + self._print("}") + + def _emit_conditions(self, keyword: str, alt: CAlternative) -> None: + self._print(f"{keyword} (") + with self._indent(): + for index, call in enumerate(alt.calls): + if index: + self._print("&&") + self._print(call) + self._print(")") + + def _emit_normal_alt(self, alt: CAlternative) -> None: + rulename = self._rule.signature.name + self._emit_conditions(keyword="if", alt=alt) + self._print("{") + with self._indent(): + node_str = alt.text.replace('"', '\\"') + self._print( + f'D(fprintf(stderr, "%*c+ {rulename}[%d-%d]: %s succeeded!\\n", p->level, \' \', _mark, p->mark, "{node_str}"));' + ) + self._emit_alt_action(alt, self._returns) + + self._print("goto done;") + self._print("}") + + def _emit_loop_alt(self, alt: CAlternative, buffer: _LoopBuffer) -> None: + self._emit_conditions(keyword="while", alt=alt) + self._print("{") + with self._indent(): + self._emit_alt_action(alt, buffer.error_returns) + buffer.append("_res") + self._print("_mark = p->mark;") + self._print("}") + + def _emit_alt_action(self, alt: CAlternative, error_returns: _CReturnEmitter) -> None: + # Location failures and explicit-action failures have distinct cleanup + # paths in the generated parser. Keep their return contexts separate. + if alt.uses_locations: + self._emit_token_end_metadata() + if not alt.action.checked: + self._emit_action_debug(alt) + self._print(f"_res = {alt.action.expression};") + if alt.action.checked: + self._print("if ((_res == NULL || p->error_indicator) && PyErr_Occurred()) {") + with self._indent(): + self._print("p->error_indicator = 1;") + error_returns.emit("NULL") + self._print("}") + self._emit_action_debug(alt) + + def _emit_action_debug(self, alt: CAlternative) -> None: + if self._debug and alt.action.debug_message: + self._print( + f'D(fprintf(stderr, "{alt.action.debug_message}\\n", _mark, p->mark, "{alt.text}"));' + ) + + def _emit_token_start_metadata(self) -> None: + self._print("if (p->mark == p->fill && _PyPegen_fill_token(p) < 0) {") + with self._indent(): + self._print("p->error_indicator = 1;") + self._returns.emit("NULL") + self._print("}") + self._print("int _start_lineno = p->tokens[_mark]->lineno;") + self._print("UNUSED(_start_lineno); // Only used by EXTRA macro") + self._print("int _start_col_offset = p->tokens[_mark]->col_offset;") + self._print("UNUSED(_start_col_offset); // Only used by EXTRA macro") + + def _emit_token_end_metadata(self) -> None: + self._print("Token *_token = _PyPegen_get_last_nonnwhitespace_token(p);") + self._print("if (_token == NULL) {") + with self._indent(): + self._returns.emit("NULL") + self._print("}") + self._print("int _end_lineno = _token->end_lineno;") + self._print("UNUSED(_end_lineno); // Only used by EXTRA macro") + self._print("int _end_col_offset = _token->end_col_offset;") + self._print("UNUSED(_end_col_offset); // Only used by EXTRA macro") + + def _emit_error_check(self) -> None: + self._print("if (p->error_indicator) {") + with self._indent(): + self._returns.emit("NULL") + self._print("}") + + def _emit_recursion_check(self) -> None: + self._print("if (p->level++ == MAXSTACK || _PyPegen_stack_exhausted(p)) {") + with self._indent(): + self._print("_Pypegen_stack_overflow(p);") + self._print("}") diff --git a/Tools/peg_generator/pegen/grammar.py b/Tools/peg_generator/pegen/grammar.py index d3c2eca6615a9f..f81e64aa583778 100644 --- a/Tools/peg_generator/pegen/grammar.py +++ b/Tools/peg_generator/pegen/grammar.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections.abc import Iterable, Iterator, Set +from enum import Enum, auto from typing import Any @@ -57,20 +58,36 @@ def __iter__(self) -> Iterator[Rule]: SIMPLE_STR = True +class RuleKind(Enum): + NORMAL = auto() + LOOP0 = auto() + LOOP1 = auto() + GATHER = auto() + + class Rule: - def __init__(self, name: str, type: str | None, rhs: Rhs, flags: frozenset[str] | None = None): + def __init__( + self, + name: str, + type: str | None, + rhs: Rhs, + flags: frozenset[str] | None = None, + *, + kind: RuleKind = RuleKind.NORMAL, + ): self.name = name self.type = type self.rhs = rhs self.flags = flags or frozenset() + self.kind = kind self.left_recursive = False self.leader = False def is_loop(self) -> bool: - return self.name.startswith("_loop") + return self.kind in (RuleKind.LOOP0, RuleKind.LOOP1) def is_gather(self) -> bool: - return self.name.startswith("_gather") + return self.kind is RuleKind.GATHER def __str__(self) -> str: if SIMPLE_STR or self.type is None: @@ -84,22 +101,18 @@ def __str__(self) -> str: return "\n".join(lines) def __repr__(self) -> str: - return f"Rule({self.name!r}, {self.type!r}, {self.rhs!r})" + kind = f", kind=RuleKind.{self.kind.name}" if self.kind is not RuleKind.NORMAL else "" + return f"Rule({self.name!r}, {self.type!r}, {self.rhs!r}{kind})" def __iter__(self) -> Iterator[Rhs]: yield self.rhs def flatten(self) -> Rhs: # If it's a single parenthesized group, flatten it. - rhs = self.rhs - if ( - not self.is_loop() - and len(rhs.alts) == 1 - and len(rhs.alts[0].items) == 1 - and isinstance(rhs.alts[0].items[0].item, Group) - ): - rhs = rhs.alts[0].items[0].item.rhs - return rhs + match self.rhs: + case Rhs(alts=[Alt(items=[NamedItem(item=Group(rhs=rhs))])]) if not self.is_loop(): + return rhs + return self.rhs class Leaf: @@ -147,12 +160,11 @@ def __iter__(self) -> Iterator[list[Alt]]: @property def can_be_inlined(self) -> bool: - if len(self.alts) != 1 or len(self.alts[0].items) != 1: - return False - # If the alternative has an action we cannot inline - if getattr(self.alts[0], "action", None) is not None: - return False - return True + match self.alts: + case [Alt(items=[_], action=None)]: + return True + case _: + return False class Alt: diff --git a/Tools/peg_generator/pegen/grammar_analysis.py b/Tools/peg_generator/pegen/grammar_analysis.py new file mode 100644 index 00000000000000..5b716cf209e000 --- /dev/null +++ b/Tools/peg_generator/pegen/grammar_analysis.py @@ -0,0 +1,155 @@ +"""Nullable and left-recursion analysis for source grammar rules.""" + +from collections.abc import Iterable, Set +from typing import Any + +from pegen import sccutils +from pegen.grammar import ( + Alt, + Cut, + Forced, + Gather, + GrammarVisitor, + Group, + NamedItem, + NameLeaf, + Opt, + Repeat0, + Repeat1, + Rhs, + Rule, + StringLeaf, +) + + +class NullableVisitor(GrammarVisitor): + def __init__(self, rules: dict[str, Rule]) -> None: + self.rules = rules + self.visited: set[Any] = set() + self.nullables: set[Rule | NamedItem] = set() + + def visit(self, node: Any, *args: Any, **kwargs: Any) -> bool | None: + match node: + case Rule(rhs=rhs): + if node in self.visited: + return False + self.visited.add(node) + if self.visit(rhs): + self.nullables.add(node) + return node in self.nullables + case NamedItem(item=item): + if self.visit(item): + self.nullables.add(node) + return node in self.nullables + case Rhs(alts=alts): + return any(self.visit(alt) for alt in alts) + case Alt(items=items): + return all(self.visit(item) for item in items) + case Forced() | Opt() | Repeat0(): + return True + case Repeat1() | Gather() | Cut(): + return False + case Group(rhs=rhs): + return self.visit(rhs) + case NameLeaf(value=name): + if (rule := self.rules.get(name)) is not None: + return self.visit(rule) + # Token or unknown; never empty. + return False + case StringLeaf(value=value): + # The string token '' is considered empty. + return not value + case _: + return self.generic_visit(node, *args, **kwargs) + + +def compute_nullables(rules: dict[str, Rule]) -> set[Any]: + """Compute which rules in a grammar are nullable. + + Thanks to TatSu (tatsu/leftrec.py) for inspiration. + """ + nullable_visitor = NullableVisitor(rules) + for rule in rules.values(): + nullable_visitor.visit(rule) + return nullable_visitor.nullables + + +class InitialNamesVisitor(GrammarVisitor): + def __init__(self, rules: dict[str, Rule]) -> None: + self.rules = rules + self.nullables = compute_nullables(rules) + + def generic_visit(self, node: Iterable[Any], *args: Any, **kwargs: Any) -> set[Any]: + names: set[str] = set() + for value in node: + if isinstance(value, list): + for item in value: + names |= self.visit(item, *args, **kwargs) + else: + names |= self.visit(value, *args, **kwargs) + return names + + def visit(self, node: Any, *args: Any, **kwargs: Any) -> set[Any]: + match node: + case Alt(items=items): + names: set[str] = set() + for item in items: + names |= self.visit(item) + if item not in self.nullables: + break + return names + case Forced() | Cut() | StringLeaf(): + return set() + case NameLeaf(value=name): + return {name} + case _: + return self.generic_visit(node, *args, **kwargs) + + +def compute_left_recursives( + rules: dict[str, Rule] +) -> tuple[dict[str, Set[str]], list[Set[str]]]: + graph = make_first_graph(rules) + sccs = list(sccutils.strongly_connected_components(graph.keys(), graph)) + for scc in sccs: + if len(scc) > 1: + for name in scc: + rules[name].left_recursive = True + # Try to find a leader such that all cycles go through it. + leaders = set(scc) + for start in scc: + for cycle in sccutils.find_cycles_in_scc(graph, scc, start): + # print("Cycle:", " -> ".join(cycle)) + leaders -= scc - set(cycle) + if not leaders: + raise ValueError( + f"SCC {scc} has no leadership candidate (no element is included in all cycles)" + ) + # print("Leaders:", leaders) + leader = min(leaders) # Pick an arbitrary leader from the candidates. + rules[leader].leader = True + else: + name = min(scc) # The only element. + if name in graph[name]: + rules[name].left_recursive = True + rules[name].leader = True + return graph, sccs + + +def make_first_graph(rules: dict[str, Rule]) -> dict[str, Set[str]]: + """Compute the graph of left-invocations. + + There's an edge from A to B if A may invoke B at its initial + position. + + Note that this requires the nullable flags to have been computed. + """ + initial_name_visitor = InitialNamesVisitor(rules) + graph: dict[str, Set[str]] = {} + vertices: set[str] = set() + for rulename, rhs in rules.items(): + graph[rulename] = names = initial_name_visitor.visit(rhs) + vertices |= names + for vertex in vertices: + graph.setdefault(vertex, set()) + return graph diff --git a/Tools/peg_generator/pegen/parser_generator.py b/Tools/peg_generator/pegen/parser_generator.py index 81314b0cc073f9..f7783f1e7810b0 100644 --- a/Tools/peg_generator/pegen/parser_generator.py +++ b/Tools/peg_generator/pegen/parser_generator.py @@ -3,30 +3,38 @@ import re import sys from abc import abstractmethod -from collections.abc import Iterable, Iterator, Set -from typing import IO, Any +from collections.abc import Iterator +from typing import IO -from pegen import sccutils from pegen.grammar import ( Alt, - Cut, - Forced, Gather, Grammar, GrammarError, GrammarVisitor, - Group, - Lookahead, NamedItem, NameLeaf, - Opt, Plain, - Repeat0, - Repeat1, Rhs, Rule, + RuleKind, StringLeaf, ) +from pegen.grammar_analysis import ( + InitialNamesVisitor as InitialNamesVisitor, +) +from pegen.grammar_analysis import ( + NullableVisitor as NullableVisitor, +) +from pegen.grammar_analysis import ( + compute_left_recursives as compute_left_recursives, +) +from pegen.grammar_analysis import ( + compute_nullables as compute_nullables, +) +from pegen.grammar_analysis import ( + make_first_graph as make_first_graph, +) class RuleCollectorVisitor(GrammarVisitor): @@ -93,12 +101,7 @@ def __init__(self, grammar: Grammar, tokens: set[str], file: IO[str] | None): self.keywords: dict[str, int] = {} self.soft_keywords: set[str] = set() self.rules = grammar.rules - self.validate_rule_names() - if "trailer" not in grammar.metas and "start" not in self.rules: - raise GrammarError("Grammar without a trailer must have a 'start' rule") - checker = RuleCheckingVisitor(self.rules, self.tokens) - for rule in self.rules.values(): - checker.visit(rule) + self._validate_grammar() self.file = file self.level = 0 self.first_graph, self.first_sccs = compute_left_recursives(self.rules) @@ -107,6 +110,14 @@ def __init__(self, grammar: Grammar, tokens: set[str], file: IO[str] | None): self.all_rules: dict[str, Rule] = self.rules.copy() # Rules + temporal rules self._local_variable_stack: list[list[str]] = [] + def _validate_grammar(self) -> None: + self.validate_rule_names() + if "trailer" not in self.grammar.metas and "start" not in self.rules: + raise GrammarError("Grammar without a trailer must have a 'start' rule") + checker = RuleCheckingVisitor(self.rules, self.tokens) + for rule in self.rules.values(): + checker.visit(rule) + def validate_rule_names(self) -> None: for rule in self.rules: if rule.startswith("_"): @@ -175,10 +186,14 @@ def artificial_rule_from_repeat(self, node: Plain, is_repeat1: bool) -> str: self.counter += 1 if is_repeat1: prefix = "_loop1_" + kind = RuleKind.LOOP1 else: prefix = "_loop0_" + kind = RuleKind.LOOP0 name = f"{prefix}{self.counter}" - self.all_rules[name] = Rule(name, None, Rhs([Alt([NamedItem(None, node)])])) + self.all_rules[name] = Rule( + name, None, Rhs([Alt([NamedItem(None, node)])]), kind=kind + ) return name def artificial_rule_from_gather(self, node: Gather) -> str: @@ -192,6 +207,7 @@ def artificial_rule_from_gather(self, node: Gather) -> str: extra_function_name, None, Rhs([extra_function_alt]), + kind=RuleKind.LOOP0, ) self.counter += 1 name = f"_gather_{self.counter}" @@ -202,6 +218,7 @@ def artificial_rule_from_gather(self, node: Gather) -> str: name, None, Rhs([alt]), + kind=RuleKind.GATHER, ) return name @@ -213,168 +230,3 @@ def dedupe(self, name: str) -> str: name = f"{origname}_{counter}" self.local_variable_names.append(name) return name - - -class NullableVisitor(GrammarVisitor): - def __init__(self, rules: dict[str, Rule]) -> None: - self.rules = rules - self.visited: set[Any] = set() - self.nullables: set[Rule | NamedItem] = set() - - def visit_Rule(self, rule: Rule) -> bool: - if rule in self.visited: - return False - self.visited.add(rule) - if self.visit(rule.rhs): - self.nullables.add(rule) - return rule in self.nullables - - def visit_Rhs(self, rhs: Rhs) -> bool: - for alt in rhs.alts: - if self.visit(alt): - return True - return False - - def visit_Alt(self, alt: Alt) -> bool: - for item in alt.items: - if not self.visit(item): - return False - return True - - def visit_Forced(self, force: Forced) -> bool: - return True - - def visit_LookAhead(self, lookahead: Lookahead) -> bool: - return True - - def visit_Opt(self, opt: Opt) -> bool: - return True - - def visit_Repeat0(self, repeat: Repeat0) -> bool: - return True - - def visit_Repeat1(self, repeat: Repeat1) -> bool: - return False - - def visit_Gather(self, gather: Gather) -> bool: - return False - - def visit_Cut(self, cut: Cut) -> bool: - return False - - def visit_Group(self, group: Group) -> bool: - return self.visit(group.rhs) - - def visit_NamedItem(self, item: NamedItem) -> bool: - if self.visit(item.item): - self.nullables.add(item) - return item in self.nullables - - def visit_NameLeaf(self, node: NameLeaf) -> bool: - if node.value in self.rules: - return self.visit(self.rules[node.value]) - # Token or unknown; never empty. - return False - - def visit_StringLeaf(self, node: StringLeaf) -> bool: - # The string token '' is considered empty. - return not node.value - - -def compute_nullables(rules: dict[str, Rule]) -> set[Any]: - """Compute which rules in a grammar are nullable. - - Thanks to TatSu (tatsu/leftrec.py) for inspiration. - """ - nullable_visitor = NullableVisitor(rules) - for rule in rules.values(): - nullable_visitor.visit(rule) - return nullable_visitor.nullables - - -class InitialNamesVisitor(GrammarVisitor): - def __init__(self, rules: dict[str, Rule]) -> None: - self.rules = rules - self.nullables = compute_nullables(rules) - - def generic_visit(self, node: Iterable[Any], *args: Any, **kwargs: Any) -> set[Any]: - names: set[str] = set() - for value in node: - if isinstance(value, list): - for item in value: - names |= self.visit(item, *args, **kwargs) - else: - names |= self.visit(value, *args, **kwargs) - return names - - def visit_Alt(self, alt: Alt) -> set[Any]: - names: set[str] = set() - for item in alt.items: - names |= self.visit(item) - if item not in self.nullables: - break - return names - - def visit_Forced(self, force: Forced) -> set[Any]: - return set() - - def visit_LookAhead(self, lookahead: Lookahead) -> set[Any]: - return set() - - def visit_Cut(self, cut: Cut) -> set[Any]: - return set() - - def visit_NameLeaf(self, node: NameLeaf) -> set[Any]: - return {node.value} - - def visit_StringLeaf(self, node: StringLeaf) -> set[Any]: - return set() - - -def compute_left_recursives( - rules: dict[str, Rule] -) -> tuple[dict[str, Set[str]], list[Set[str]]]: - graph = make_first_graph(rules) - sccs = list(sccutils.strongly_connected_components(graph.keys(), graph)) - for scc in sccs: - if len(scc) > 1: - for name in scc: - rules[name].left_recursive = True - # Try to find a leader such that all cycles go through it. - leaders = set(scc) - for start in scc: - for cycle in sccutils.find_cycles_in_scc(graph, scc, start): - # print("Cycle:", " -> ".join(cycle)) - leaders -= scc - set(cycle) - if not leaders: - raise ValueError( - f"SCC {scc} has no leadership candidate (no element is included in all cycles)" - ) - # print("Leaders:", leaders) - leader = min(leaders) # Pick an arbitrary leader from the candidates. - rules[leader].leader = True - else: - name = min(scc) # The only element. - if name in graph[name]: - rules[name].left_recursive = True - rules[name].leader = True - return graph, sccs - - -def make_first_graph(rules: dict[str, Rule]) -> dict[str, Set[str]]: - """Compute the graph of left-invocations. - - There's an edge from A to B if A may invoke B at its initial - position. - - Note that this requires the nullable flags to have been computed. - """ - initial_name_visitor = InitialNamesVisitor(rules) - graph = {} - vertices: set[str] = set() - for rulename, rhs in rules.items(): - graph[rulename] = names = initial_name_visitor.visit(rhs) - vertices |= names - for vertex in vertices: - graph.setdefault(vertex, set()) - return graph