https://github.com/python/cpython/commit/b7f81d4fab8f2d7f096642433967ce0e713802dd
commit: b7f81d4fab8f2d7f096642433967ce0e713802dd
branch: main
author: Pablo Galindo Salgado <[email protected]>
committer: pablogsal <[email protected]>
date: 2026-09-24T16:27:08Z
summary:
Refactor pegen C generation into explicit compilation phases (#157501)
* Move pegen grammar analysis into its own module
* Record pegen helper rule kinds explicitly
* Introduce immutable C parser data and separate lowering
* Add C emitters with local rule and output state
* Route C generation through the prepared parser model
* Fix pegen import formatting for lint
* Preserve and test shared-prefix preparation after rebase
files:
A Lib/test/test_peg_generator/test_c_generator.py
A Tools/peg_generator/pegen/c_generator_calls.py
A Tools/peg_generator/pegen/c_generator_file.py
A Tools/peg_generator/pegen/c_generator_model.py
A Tools/peg_generator/pegen/c_generator_rules.py
A Tools/peg_generator/pegen/grammar_analysis.py
M Lib/test/test_peg_generator/test_c_parser.py
M Lib/test/test_peg_generator/test_pegen.py
M Tools/peg_generator/pegen/c_generator.py
M Tools/peg_generator/pegen/grammar.py
M Tools/peg_generator/pegen/parser_generator.py
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
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
