Regex Engine

কেন এই প্রজেক্ট

cs-theory/finite-automata-dfa-nfa আর cs-theory/regular-expressions-and-pumping-lemma লেসনে আপনি হাতে-কলমে Thompson’s construction আর subset construction করেছেন — কাগজে, ছোট উদাহরণে। এই প্রজেক্ট সেই দুইটা algorithm-কে একটা সম্পূর্ণ, কাজ-করা regex engine-এ রূপান্তর করে — এমন একটা engine যেটা কখনো catastrophic backtracking-এ পড়বে না, কারণ এটা ব্যাকট্র্যাকই করে না।

এটাই সেই লেসনের Cloudflare outage গল্পের সরাসরি উত্তর — RE2 (Google), Rust-এর regex crate ঠিক এই কৌশলে বানানো: automaton-first, backtracking কখনো না। এই প্রজেক্ট শেষে আপনার কাছে থাকবে এই একই স্থাপত্যের একটা ছোট, বোধগম্য সংস্করণ — আর একটা সংখ্যা, যেটা প্রমাণ করবে কেন এই স্থাপত্যটাই সঠিক: (a+)+b-জাতীয় pattern-এ আপনার engine Python-এর re-এর চেয়ে হাজার গুণ দ্রুত হবে, ইনপুট যত বড়ই হোক।

ধাপে ধাপে

১. Regex parser — string থেকে AST

compilers/parsing-এর recursive descent কৌশল সরাসরি প্রয়োগ করুন। Grammar (precedence low-to-high: union, concat, star):

expr   → term ('|' term)*
term   → factor+
factor → atom ('*' | '+' | '?')?
atom   → CHAR | '(' expr ')'

class Char:  __slots__ = ("c",);      __init__ = lambda s, c: setattr(s, "c", c)
class Concat: __slots__ = ("parts",); __init__ = lambda s, p: setattr(s, "parts", p)
class Union:  __slots__ = ("opts",);  __init__ = lambda s, o: setattr(s, "opts", o)
class Star:   __slots__ = ("inner",); __init__ = lambda s, i: setattr(s, "inner", i)

def parse(pattern: str):
    pos = 0
    def peek(): return pattern[pos] if pos < len(pattern) else None
    def expr():
        nonlocal pos
        opts = [term()]
        while peek() == '|':
            pos += 1; opts.append(term())
        return opts[0] if len(opts) == 1 else Union(opts)
    def term():
        nonlocal pos
        parts = []
        while peek() not in (None, '|', ')'):
            parts.append(factor())
        return parts[0] if len(parts) == 1 else Concat(parts)
    def factor():
        nonlocal pos
        a = atom()
        while peek() in ('*', '+', '?'):
            op = pattern[pos]; pos += 1
            if op == '*': a = Star(a)
            elif op == '+': a = Concat([a, Star(a)])       # desugar: a+ = aa*
            elif op == '?': a = Union([a, Concat([])])      # desugar: a? = (a|ε)
        return a
    def atom():
        nonlocal pos
        if peek() == '(':
            pos += 1; e = expr(); pos += 1
            return e
        c = pattern[pos]; pos += 1
        return Char(c)
    return expr()

+ আর ?-কে parser-এই *, |-এ desugar করে ফেললাম — তাই Thompson’s construction-এ নতুন কোনো টেমপ্লেট লাগবে না, শুধু মূল তিনটা (Char, Concat, Union, Star — চারটা node type, তিনটা অপারেটর)।

২. Thompson’s construction — AST থেকে ε-NFA

লেসনের টেমপ্লেট সরাসরি কোডে। প্রতিটা fragment একটা (start, accept, delta, eps) — বাইরের কেউ শুধু start/accept নিয়ে কাজ করে, ভেতরের state সংখ্যা যত খুশি হোক।

import itertools
_counter = itertools.count()
def new_state(): return f"q{next(_counter)}"

def thompson(node):
    if isinstance(node, Char):
        s, a = new_state(), new_state()
        return {"start": s, "accept": a, "delta": {(s, node.c): {a}}, "eps": {}}
    if isinstance(node, Concat):
        if not node.parts:                          # ε (from a? desugar)
            s = new_state(); return {"start": s, "accept": s, "delta": {}, "eps": {}}
        frags = [thompson(p) for p in node.parts]
        delta, eps = {}, {}
        for f in frags: delta.update(f["delta"]); eps.update(f["eps"])
        for f1, f2 in zip(frags, frags[1:]):
            eps.setdefault(f1["accept"], set()).add(f2["start"])
        return {"start": frags[0]["start"], "accept": frags[-1]["accept"], "delta": delta, "eps": eps}
    if isinstance(node, Union):
        frags = [thompson(o) for o in node.opts]
        s, a = new_state(), new_state()
        delta, eps = {}, {s: set(), a: set()}
        for f in frags:
            delta.update(f["delta"]); eps.update(f["eps"])
            eps[s].add(f["start"]); eps.setdefault(f["accept"], set()).add(a)
        return {"start": s, "accept": a, "delta": delta, "eps": eps}
    if isinstance(node, Star):
        f = thompson(node.inner)
        s, a = new_state(), new_state()
        eps = dict(f["eps"]); eps[s] = {f["start"], a}
        eps.setdefault(f["accept"], set()).update({f["start"], a})
        return {"start": s, "accept": a, "delta": f["delta"], "eps": eps}

Checkpoint: thompson(parse("a(b|c)*"))-এর delta/eps হাতে-করা লেসনের ট্রেসের সাথে (state নাম বাদে) কাঠামোগতভাবে মিলছে কি না যাচাই করুন — start-এ ঢুকে a পড়ে, তারপর b/c লুপ, শেষে accept।

৩. Subset construction — NFA থেকে DFA

লেসনের worklist algorithm, generic করে লেখা:

def eclose(nfa, states):
    stack, result = list(states), set(states)
    while stack:
        s = stack.pop()
        for t in nfa["eps"].get(s, ()):
            if t not in result: result.add(t); stack.append(t)
    return frozenset(result)

def to_dfa(nfa, alphabet):
    start = eclose(nfa, {nfa["start"]})
    delta, worklist, seen = {}, [start], {start}
    while worklist:
        S = worklist.pop()
        for a in alphabet:
            raw = set()
            for q in S: raw |= nfa["delta"].get((q, a), set())
            T = eclose(nfa, raw)
            if T:
                delta[(S, a)] = T
                if T not in seen: seen.add(T); worklist.append(T)
    accept = {S for S in seen if nfa["accept"] in S}
    return {"delta": delta, "start": start, "accept": accept}

alphabet-টা pattern-এ ব্যবহৃত সব literal character থেকে বের করুন (compile()-এ একবার scan করে)।

৪. DFA matcher — backtracking নেই

def compile_regex(pattern: str):
    ast = parse(pattern)
    alphabet = set(pattern) - set("()|*+?")
    nfa = thompson(ast)
    return to_dfa(nfa, alphabet)

def fullmatch(dfa, s: str) -> bool:
    state = dfa["start"]
    for ch in s:
        state = dfa["delta"].get((state, ch))
        if state is None:
            return False                # কোনো backtrack — সরাসরি reject
    return state in dfa["accept"]

fullmatch প্রতি character-এ ঠিক একটা dict lookup — O(n), প্রতিবার, pattern যত জটিলই হোক (compile-time cost আলাদা, সেটাও worst-case subset construction-এর exponential bound সাপেক্ষে, কিন্তু match-time সবসময় linear)।

Checkpoint: fullmatch(compile_regex("a(b|c)*"), "abcbc") → True, fullmatch(..., "abd") → False। re.fullmatch("a(b|c)*", s)-এর সাথে দশটা হাতে-বাছা string-এ ফলাফল মিলিয়ে দেখুন।

৫. DFA minimization

def minimize(dfa, alphabet):
    states = {s for s, _ in dfa["delta"]} | set(dfa["delta"].values()) | {dfa["start"]}
    accept, nonaccept = dfa["accept"], states - dfa["accept"]
    partitions = [p for p in (accept, nonaccept) if p]
    changed = True
    while changed:
        changed = False
        new_partitions = []
        for group in partitions:
            buckets = {}
            for s in group:
                sig = tuple(next((i for i, g in enumerate(partitions) if dfa["delta"].get((s, a)) in g), -1)
                             for a in alphabet)
                buckets.setdefault(sig, set()).add(s)
            if len(buckets) > 1: changed = True
            new_partitions.extend(buckets.values())
        partitions = new_partitions
    return partitions   # প্রতিটা group একটা মিনিমাইজড state

Checkpoint: একটা redundant-state-সহ ছোট DFA হাতে বানিয়ে চালান — partition সংখ্যা মূল state সংখ্যার চেয়ে কম হলে minimization কাজ করছে।

৬. ReDoS বেঞ্চমার্ক

import re, time

def bench(pattern, n):
    s = "a" * n
    dfa = compile_regex(pattern.replace("+", "*"))  # + ইতিমধ্যে desugar হয়েছে compile_regex-এ
    t0 = time.perf_counter(); fullmatch(dfa, s + "x"); t_dfa = time.perf_counter() - t0
    t0 = time.perf_counter(); re.fullmatch(pattern, s + "x"); t_re = time.perf_counter() - t0
    print(f"n={n:3}  own-engine={t_dfa*1000:8.3f}ms  python-re={t_re*1000:10.2f}ms")

for n in [10, 15, 20, 25]:
    bench("(a+)+b", n)

`n`	নিজের engine	Python `re`
10	?	?
15	?	?
20	?	?
25	?	?

নিজের সংখ্যা ভরুন — re-এর কলাম exponentially বাড়বে, আপনার engine-এর কলাম প্রায় স্থির থাকবে। এই টেবিলটাই আপনার প্রজেক্টের মূল দাবির প্রমাণ।

৭. Correctness test suite

import random, string

def random_pattern(depth=3):
    if depth == 0 or random.random() < 0.3:
        return random.choice(string.ascii_lowercase[:3])
    op = random.choice(["concat", "union", "star"])
    if op == "star":
        return f"({random_pattern(depth-1)})*"
    a, b = random_pattern(depth-1), random_pattern(depth-1)
    return f"{a}|{b}" if op == "union" else f"{a}{b}"

mismatches = 0
for _ in range(500):
    p = random_pattern()
    dfa = compile_regex(p)
    for _ in range(20):
        s = "".join(random.choice("abc") for _ in range(random.randint(0, 6)))
        own = fullmatch(dfa, s)
        ref = bool(re.fullmatch(p, s))
        if own != ref:
            mismatches += 1
            print("MISMATCH", p, s, own, ref)
print("mismatches:", mismatches)

Checkpoint: mismatches == 0 — এটাই Kleene’s theorem-এর ব্যবহারিক যাচাই, হাজার হাজার random case-এ।

নিজেকে চ্যালেঞ্জ করুন

Character class ([a-z], [^abc]) সমর্থন যোগ করুন — parser-এ নতুন atom case, Thompson-এ multi-symbol union হিসেবে desugar
Anchor (^, $) আর . (যেকোনো character) সমর্থন যোগ করুন
Lazy DFA — পুরো subset construction আগে থেকে না চালিয়ে, শুধু প্রয়োজনমতো state বানান আর cache করুন (RE2-র lazy DFA-র সরলীকৃত সংস্করণ) — memory ব্যবহার তুলনা করুন eager বনাম lazy-তে
Capture group ((...)) — এটা কঠিন: pure DFA-তে group boundary ট্র্যাক করা যায় না (কোন state-এ ছিলাম তা “ভুলে” যায় subset construction-এ), তাই একটা tagged-NFA বা parallel Thompson-VM কৌশল লাগবে (Russ Cox-এর reference পড়ুন)
{n,m} repetition quantifier — desugar কৌশলে (a{2,4} = aa(a(a)?)?) implement করুন, বড় n-এ state সংখ্যা কীভাবে বাড়ে মাপুন
একই pattern-এর জন্য নিজের DFA state সংখ্যা বনাম minimized state সংখ্যা বনাম Python re-এর internal compiled size (যদি বের করা যায়) — একটা তুলনামূলক রিপোর্ট লিখুন

যা এখানে শিখলেন, পরে যেখানে লাগবে

এখানে	পরে
Regex parser (recursive descent)	যেকোনো DSL/config format parser — `compilers/parsing`
Thompson’s construction (structural recursion)	Compiler IR generation — AST থেকে recursively code বানানো
Subset construction (NFA→DFA)	Lexer generator internals (flex/re2c) — `compilers/lexical-analysis`
DFA minimization	Protocol/FSM state reduction, embedded system-এ compact table
ReDoS awareness	Input validation code review, security audit — `security` module
Test-against-reference methodology	Property-based testing, compiler correctness testing সাধারণভাবে

মাইলস্টোন

আগে যা পড়া দরকার