Pike VM in Python
Runs Thompson's simulation with thread priorities so that it returns leftmost-first submatches.
- File
pike_vm.py, 165 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/pike-vm/. The authors are listed in prov/VF24KJ.jsonl.
"""The Pike VM: Thompson's NFA simulation with thread priorities and submatches.
A regular expression is a tuple:
("empty",) for the empty language, ("eps",) for the empty string,
("sym", a) for one symbol, ("alt", r, s), ("cat", r, s), ("star", r),
and two extras: ("lazy", r) for the non-greedy star r*?, and ("group", r)
for a capture group. Groups are numbered 1, 2, ... in the order of their
opening parentheses; group 0 is the whole match.
The expression is compiled to a program of instructions:
("char", a), ("split", x, y) which prefers x, ("jmp", x), ("save", k),
("fail",) and ("match",). The VM runs one thread per program counter,
keeps the threads in priority order, and returns the leftmost-first match.
"""
from __future__ import annotations
from typing import Any
Regex = tuple[Any, ...]
Program = list[tuple[Any, ...]]
Spans = list[tuple[int, int] | None]
def compile_regex(regex: Regex) -> tuple[Program, int]:
"""Compile to a program; return (program, number of groups including group 0).
The tree is walked with an explicit stack, so deep expressions need no recursion.
"""
prog: list[list[Any]] = [["save", 0]]
ngroups = 1
todo: list[tuple[str, Any]] = [("emit", regex)]
while todo:
task, arg = todo.pop()
if task == "emit":
tag = arg[0]
if tag == "sym":
prog.append(["char", arg[1]])
elif tag == "empty":
prog.append(["fail"])
elif tag == "cat":
todo.append(("emit", arg[2]))
todo.append(("emit", arg[1]))
elif tag == "alt":
split = len(prog)
prog.append(["split", split + 1, None])
cell = [split, None] # the split, then the jmp that ends the left branch
todo.append(("end_alt", cell))
todo.append(("emit", arg[2]))
todo.append(("mid_alt", cell))
todo.append(("emit", arg[1]))
elif tag in ("star", "lazy"):
split = len(prog)
if tag == "star":
prog.append(["split", split + 1, None])
else:
prog.append(["split", None, split + 1])
todo.append(("end_star", split))
todo.append(("emit", arg[1]))
elif tag == "group":
k = ngroups
ngroups += 1
prog.append(["save", 2 * k])
todo.append(("save", 2 * k + 1))
todo.append(("emit", arg[1]))
elif tag != "eps":
raise ValueError(f"unknown node {arg!r}")
elif task == "mid_alt":
arg[1] = len(prog)
prog.append(["jmp", None])
prog[arg[0]][2] = len(prog)
elif task == "end_alt":
prog[arg[1]][1] = len(prog)
elif task == "end_star":
prog.append(["jmp", arg])
ins = prog[arg]
if ins[2] is None:
ins[2] = len(prog)
else:
ins[1] = len(prog)
elif task == "save":
prog.append(["save", arg])
prog.append(["save", 1])
prog.append(["match"])
return [tuple(ins) for ins in prog], ngroups
def _add_thread(
prog: Program, threads: list[tuple[int, tuple[int, ...]]], seen: set[int], pc: int, caps: tuple[int, ...], i: int
) -> None:
"""Add the thread (pc, caps) at position i and follow jmp, split and save.
Depth-first in priority order with an explicit stack; a program counter
already in `seen` is skipped, so each list holds one thread per pc.
"""
stack = [(pc, caps)]
while stack:
pc, caps = stack.pop()
if pc in seen: # @line L14
continue
seen.add(pc)
ins = prog[pc]
op = ins[0]
if op == "jmp": # @line L15
stack.append((ins[1], caps))
elif op == "split": # @line L16
stack.append((ins[2], caps))
stack.append((ins[1], caps))
elif op == "save": # @line L17
k = ins[1]
stack.append((pc + 1, caps[:k] + (i,) + caps[k + 1 :]))
elif op != "fail": # @line L18
threads.append((pc, caps))
def pike_vm(prog: Program, ngroups: int, text: str, full: bool = False) -> Spans | None:
"""Return the spans of the leftmost-first match of the program in `text`, or None.
spans[k] is (start, end) for group k, or None if group k took no part.
With full=True the match must cover the whole text.
Cost: O(len(prog) * len(text) * ngroups) time and O(len(prog) * ngroups) space.
"""
n = len(text)
blank = (-1,) * (2 * ngroups)
clist: list[tuple[int, tuple[int, ...]]] = [] # @line L1
cseen: set[int] = set()
matched: tuple[int, ...] | None = None
for i in range(n + 1): # @line L2
if matched is None and (i == 0 or not full): # @line L3
_add_thread(prog, clist, cseen, 0, blank, i)
if not clist: # @line L4
break
nlist: list[tuple[int, tuple[int, ...]]] = [] # @line L5
nseen: set[int] = set()
for pc, caps in clist: # @line L6
ins = prog[pc]
if ins[0] == "char":
if i < n and text[i] == ins[1]: # @line L7
_add_thread(prog, nlist, nseen, pc + 1, caps, i + 1)
elif not full or i == n: # @line L8
matched = caps # @line L9
break # @line L10
clist, cseen = nlist, nseen # @line L11
if matched is None: # @line L12
return None
spans: Spans = []
for k in range(ngroups):
s, e = matched[2 * k], matched[2 * k + 1]
spans.append((s, e) if s >= 0 and e >= 0 else None)
return spans # @line L13
def search(regex: Regex, text: str) -> Spans | None:
"""The leftmost-first match of `regex` in `text`, with the spans of all groups."""
prog, ngroups = compile_regex(regex)
return pike_vm(prog, ngroups, text)
def fullmatch(regex: Regex, text: str) -> Spans | None:
"""The highest-priority match of `regex` that covers all of `text`, or None."""
prog, ngroups = compile_regex(regex)
return pike_vm(prog, ngroups, text, full=True)
test_pike_vm.py 365 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/pike-vm/. The authors are listed in prov/VF24KJ.jsonl.
import itertools
import random
import re
from pike_vm import compile_regex, fullmatch, pike_vm, search
EMPTY = ("empty",)
EPS = ("eps",)
def parse(text):
"""Parse | , concatenation, *, parentheses, ε and ∅ into the tuple form."""
pos = 0
def peek():
return text[pos] if pos < len(text) else None
def p_alt():
nonlocal pos
r = p_cat()
while peek() == "|":
pos += 1
r = ("alt", r, p_cat())
return r
def p_cat():
items = []
while peek() is not None and peek() not in "|)":
items.append(p_rep())
if not items:
return EPS
r = items[0]
for x in items[1:]:
r = ("cat", r, x)
return r
def p_rep():
nonlocal pos
r = p_atom()
while peek() == "*":
pos += 1
r = ("star", r)
return r
def p_atom():
nonlocal pos
c = peek()
pos += 1
if c == "(":
r = p_alt()
assert peek() == ")"
pos += 1
return r
if c == "ε":
return EPS
if c == "∅":
return EMPTY
return ("sym", c)
r = p_alt()
assert pos == len(text)
return r
def language(r, bound):
"""All strings of L(r) of length at most bound, by enumeration."""
t = r[0]
if t == "empty":
return set()
if t == "eps":
return {""}
if t == "sym":
return {r[1]} if bound >= 1 else set()
if t == "alt":
return language(r[1], bound) | language(r[2], bound)
if t == "cat":
a, b = language(r[1], bound), language(r[2], bound)
return {x + y for x in a for y in b if len(x) + len(y) <= bound}
body = language(r[1], bound) - {""}
result, frontier = {""}, {""}
while frontier:
frontier = {x + y for x in frontier for y in body if len(x) + len(y) <= bound} - result
result |= frontier
return result
def words(alphabet, bound):
for k in range(bound + 1):
for t in itertools.product(alphabet, repeat=k):
yield "".join(t)
def random_regex(rng, alphabet, depth):
if depth == 0 or rng.random() < 0.25:
x = rng.random()
if x < 0.08:
return EMPTY
if x < 0.18:
return EPS
return ("sym", rng.choice(alphabet))
k = rng.random()
if k < 0.35:
return ("alt", random_regex(rng, alphabet, depth - 1), random_regex(rng, alphabet, depth - 1))
if k < 0.75:
return ("cat", random_regex(rng, alphabet, depth - 1), random_regex(rng, alphabet, depth - 1))
return ("star", random_regex(rng, alphabet, depth - 1))
def decorate(rng, r):
"""Turn some nodes into groups and some stars into lazy stars."""
t = r[0]
if t in ("alt", "cat"):
r = (t, decorate(rng, r[1]), decorate(rng, r[2]))
elif t == "star":
r = ("lazy" if rng.random() < 0.3 else "star", decorate(rng, r[1]))
if rng.random() < 0.3:
r = ("group", r)
return r
def plain(r):
"""Drop groups and laziness, which do not change the language."""
t = r[0]
if t == "group":
return plain(r[1])
if t in ("star", "lazy"):
return ("star", plain(r[1]))
if t in ("alt", "cat"):
return (t, plain(r[1]), plain(r[2]))
return r
def to_python(r):
t = r[0]
if t == "empty":
return "(?!)"
if t == "eps":
return "(?:)"
if t == "sym":
return r[1]
if t == "alt":
return "(?:" + to_python(r[1]) + "|" + to_python(r[2]) + ")"
if t == "cat":
return "(?:" + to_python(r[1]) + ")(?:" + to_python(r[2]) + ")"
if t == "star":
return "(?:" + to_python(r[1]) + ")*"
if t == "lazy":
return "(?:" + to_python(r[1]) + ")*?"
return "(" + to_python(r[1]) + ")"
def nullable(r):
return "" in language(plain(r), 0)
def has_nullable_star_body(r):
t = r[0]
if t in ("star", "lazy"):
return nullable(r[1]) or has_nullable_star_body(r[1])
if t in ("alt", "cat"):
return has_nullable_star_body(r[1]) or has_nullable_star_body(r[2])
if t == "group":
return has_nullable_star_body(r[1])
return False
def bitstate(prog, ngroups, text, full=False):
"""Backtracking in priority order that never visits a (pc, position) pair twice."""
n = len(text)
visited = set()
for start in range(n + 1) if not full else [0]:
stack = [(0, start, (-1,) * (2 * ngroups))]
while stack:
pc, i, caps = stack.pop()
if (pc, i) in visited:
continue
visited.add((pc, i))
ins = prog[pc]
op = ins[0]
if op == "char":
if i < n and text[i] == ins[1]:
stack.append((pc + 1, i + 1, caps))
elif op == "jmp":
stack.append((ins[1], i, caps))
elif op == "split":
stack.append((ins[2], i, caps))
stack.append((ins[1], i, caps))
elif op == "save":
k = ins[1]
stack.append((pc + 1, i, caps[:k] + (i,) + caps[k + 1 :]))
elif op == "match":
if not full or i == n:
return [(caps[2 * k], caps[2 * k + 1]) if caps[2 * k] >= 0 and caps[2 * k + 1] >= 0 else None for k in range(ngroups)]
return None
def python_spans(r, text):
m = re.search(to_python(r), text)
if m is None:
return None
return [m.span(k) if m.span(k) != (-1, -1) else None for k in range((m.re.groups) + 1)]
def random_text(rng, alphabet):
return "".join(rng.choice(alphabet) for _ in range(rng.randrange(0, 9)))
def test_example_ambiguous_submatch():
r = parse_groups("(a|ab)(c|bcd)(d*)")
assert search(r, "abcd") == [(0, 4), (0, 1), (1, 4), (4, 4)]
assert fullmatch(r, "abcd") == [(0, 4), (0, 1), (1, 4), (4, 4)]
assert search(r, "xabcdd") == [(1, 6), (1, 2), (2, 5), (5, 6)]
def parse_groups(text):
"""Like parse, but ( ) makes a capture group and *? a lazy star."""
pos = 0
def peek():
return text[pos] if pos < len(text) else None
def p_alt():
nonlocal pos
r = p_cat()
while peek() == "|":
pos += 1
r = ("alt", r, p_cat())
return r
def p_cat():
items = []
while peek() is not None and peek() not in "|)":
items.append(p_rep())
if not items:
return EPS
r = items[0]
for x in items[1:]:
r = ("cat", r, x)
return r
def p_rep():
nonlocal pos
r = p_atom()
while peek() == "*":
pos += 1
if peek() == "?":
pos += 1
r = ("lazy", r)
else:
r = ("star", r)
return r
def p_atom():
nonlocal pos
c = peek()
pos += 1
if c == "(":
r = p_alt()
assert peek() == ")"
pos += 1
return ("group", r)
if c == "ε":
return EPS
if c == "∅":
return EMPTY
return ("sym", c)
r = p_alt()
assert pos == len(text)
return r
def test_greedy_lazy_and_groups():
assert search(parse_groups("(a*)(a*)"), "aaa") == [(0, 3), (0, 3), (3, 3)]
assert search(parse_groups("(a*?)(a*)"), "aaa") == [(0, 3), (0, 0), (0, 3)]
assert search(parse_groups("(a*?)b"), "aab") == [(0, 3), (0, 2)]
assert search(parse_groups("((a)|b)*"), "ab") == [(0, 2), (1, 2), (0, 1)]
assert search(parse_groups("(a|b)*c"), "ababc") == [(0, 5), (3, 4)]
assert search(parse_groups("b"), "aab") == [(2, 3)]
assert search(parse_groups("ab|b"), "bab") == [(0, 1)]
def test_edge_cases():
assert search(EMPTY, "") is None
assert search(EMPTY, "ab") is None
assert search(EPS, "") == [(0, 0)]
assert search(EPS, "ab") == [(0, 0)]
assert fullmatch(EPS, "a") is None
assert search(parse_groups("a"), "") is None
assert search(parse_groups("(∅)|b"), "b") == [(0, 1), None]
# An empty iteration that would revisit the star's split is cut, so these
# differ from Python's re, which reports (0, 0), (0, 0), (2, 2) and (2, 2).
assert search(parse_groups("(a*)*"), "") == [(0, 0), None]
assert search(parse_groups("(a*)*"), "b") == [(0, 0), None]
assert search(parse_groups("(a*)*"), "aa") == [(0, 2), (0, 2)]
assert search(parse_groups("(a|ε)*b"), "aab") == [(0, 3), (1, 2)]
assert fullmatch(parse_groups("((a*)*b*)*"), "abab") == bitstate(*compile_regex(parse_groups("((a*)*b*)*")), "abab", True)
def test_deep_inputs():
n = 3000
r = ("sym", "a")
for _ in range(n - 1):
r = ("cat", r, ("sym", "a"))
assert fullmatch(r, "a" * n) == [(0, n)]
assert fullmatch(r, "a" * (n - 1)) is None
s = ("sym", "b")
for _ in range(1500):
s = ("star", ("group", s))
spans = search(s, "b" * 50)
assert spans[0] == (0, 50)
assert search(("cat", s, ("sym", "c")), "b" * 40) is None
def test_random_against_backtracker_and_language():
rng = random.Random(2026_11_06)
for trial in range(1500):
alphabet = "ab" if trial % 2 == 0 else "abc"
r = decorate(rng, random_regex(rng, alphabet, 4))
prog, ngroups = compile_regex(r)
lang = language(plain(r), 6 if alphabet == "ab" else 4)
for w in words(alphabet, 6 if alphabet == "ab" else 4):
got = pike_vm(prog, ngroups, w, full=True)
assert (got is not None) == (w in lang), (r, w)
assert got == bitstate(prog, ngroups, w, True), (r, w)
for _ in range(4):
t = random_text(rng, alphabet)
got = pike_vm(prog, ngroups, t)
assert got == bitstate(prog, ngroups, t), (r, t)
subs = [t[j:k] for j in range(len(t) + 1) for k in range(j, len(t) + 1)]
big = language(plain(r), len(t))
assert (got is not None) == any(x in big for x in subs), (r, t)
if got is not None:
s0, s1 = got[0]
assert t[s0:s1] in big
for sp in got[1:]:
if sp is not None:
assert s0 <= sp[0] <= sp[1] <= s1
def test_random_against_python_re():
rng = random.Random(7)
compared = 0
while compared < 800:
alphabet = "ab" if compared % 2 == 0 else "abc"
r = decorate(rng, random_regex(rng, alphabet, 4))
if has_nullable_star_body(r):
continue
compared += 1
for _ in range(5):
t = random_text(rng, alphabet)
assert search(r, t) == python_spans(r, t), (r, to_python(r), t)
def test_nested_stars_against_backtracker():
for text in ["(a*)*", "((a*)*b)*", "(a*b*)*", "((ε|a)*)*", "(∅|a*)*b", "((a*?)*)*", "(a|ab)*(b*)"]:
r = parse_groups(text)
prog, ngroups = compile_regex(r)
for w in words("ab", 6):
assert pike_vm(prog, ngroups, w) == bitstate(prog, ngroups, w), (text, w)
assert pike_vm(prog, ngroups, w, True) == bitstate(prog, ngroups, w, True), (text, w)
What the next level needs
- the tests did not run in this build