xiiregexbuilder
FPGA-Accelerated Regular Expression Matching Engineprocessor/src/compile_regex.py (8893B)
1import os
2
3# Glushkov Construction for Regex CPU
4
5
6class Node:
7 def __init__(self, type):
8 self.type = type
9 self.nullable = False
10 self.first = set()
11 self.last = set()
12 self.pos = None
13 self.char = None
14
15
16def get_positions(node):
17 if node.type == "lit":
18 return {node.pos: node.char}
19 res = {}
20 if hasattr(node, "left"):
21 res.update(get_positions(node.left))
22 if hasattr(node, "right"):
23 res.update(get_positions(node.right))
24 if hasattr(node, "child"):
25 res.update(get_positions(node.child))
26 return res
27
28
29def compute_follow(node, follow):
30 if node.type == "cat":
31 compute_follow(node.left, follow)
32 compute_follow(node.right, follow)
33 for i in node.left.last:
34 follow[i] |= node.right.first
35 elif node.type == "or":
36 compute_follow(node.left, follow)
37 compute_follow(node.right, follow)
38 elif node.type == "star" or node.type == "plus":
39 compute_follow(node.child, follow)
40 for i in node.child.last:
41 follow[i] |= node.child.first
42 elif node.type == "quest":
43 compute_follow(node.child, follow)
44
45
46def parse_regex(pattern):
47 tokens = list(pattern[::-1])
48
49 def parse_expr():
50 node = parse_cat()
51 while tokens and tokens[-1] == "|":
52 tokens.pop()
53 right = parse_cat()
54 new_node = Node("or")
55 new_node.left = node
56 new_node.right = right
57 new_node.nullable = node.nullable or right.nullable
58 new_node.first = node.first | right.first
59 new_node.last = node.last | right.last
60 node = new_node
61 return node
62
63 def parse_cat():
64 node = parse_unary()
65 while tokens and tokens[-1] not in ")|":
66 right = parse_unary()
67 new_node = Node("cat")
68 new_node.left = node
69 new_node.right = right
70 new_node.nullable = node.nullable and right.nullable
71 new_node.first = node.first | (right.first if node.nullable else set())
72 new_node.last = right.last | (node.last if right.nullable else set())
73 node = new_node
74 return node
75
76 def parse_unary():
77 node = parse_primary()
78 while tokens and tokens[-1] in "*+?":
79 op = tokens.pop()
80 new_node = Node("star" if op == "*" else ("plus" if op == "+" else "quest"))
81 new_node.child = node
82 if op == "*":
83 new_node.nullable = True
84 new_node.first = node.first
85 new_node.last = node.last
86 elif op == "+":
87 new_node.nullable = node.nullable
88 new_node.first = node.first
89 new_node.last = node.last
90 else: # ?
91 new_node.nullable = True
92 new_node.first = node.first
93 new_node.last = node.last
94 node = new_node
95 return node
96
97 pos_counter = [1]
98
99 def parse_primary():
100 c = tokens.pop()
101 if c == "(":
102 node = parse_expr()
103 if tokens:
104 tokens.pop() # )
105 return node
106 else:
107 node = Node("lit")
108 node.char = c
109 node.pos = pos_counter[0]
110 pos_counter[0] += 1
111 node.nullable = False
112 node.first = {node.pos}
113 node.last = {node.pos}
114 return node
115
116 return parse_expr()
117
118
119def generate_rasm(root, match_id, base_pc):
120 positions = get_positions(root)
121 follow = {i: set() for i in positions}
122 compute_follow(root, follow)
123
124 # Map Glushkov positions to absolute PCs
125 pc_map = {0: base_pc} # Start state
126 for i in range(1, len(positions) + 1):
127 pc_map[i] = base_pc + i
128
129 match_state_pc = base_pc + len(positions) + 1
130 next_free_pc = match_state_pc + 1
131
132 rasm = []
133
134 # MATCH State (PC: match_state_pc)
135 # The MATCH instruction in your HW: char=0, term=1, mid=match_id
136 rasm.append(f"{match_state_pc}: MATCH({match_id})")
137
138 # Start chain (Position 0)
139 start_targets = list(root.first)
140 if root.nullable:
141 # If the regex matches empty string, the start state should point to match
142 start_targets.append(match_state_pc)
143
144 if not start_targets:
145 rasm.append(f"{pc_map[0]}: MATCH({match_id})")
146 else:
147
148 def write_split_chain(targets, start_pc, free_pc):
149 if not targets:
150 return [], free_pc
151 if len(targets) == 1:
152 target_pc = pc_map[targets[0]] if targets[0] in pc_map else targets[0]
153 return [f"{start_pc}: JMP({target_pc})"], free_pc
154 elif len(targets) == 2:
155 t1 = pc_map[targets[0]] if targets[0] in pc_map else targets[0]
156 t2 = pc_map[targets[1]] if targets[1] in pc_map else targets[1]
157 return [f"{start_pc}: SPLIT({t1}, {t2})"], free_pc
158 else:
159 t1 = pc_map[targets[0]] if targets[0] in pc_map else targets[0]
160 lines = [f"{start_pc}: SPLIT({t1}, {free_pc})"]
161 more, last = write_split_chain(targets[1:], free_pc, free_pc + 1)
162 return lines + more, last
163
164 chain, next_free_pc = write_split_chain(start_targets, pc_map[0], next_free_pc)
165 rasm.extend(chain)
166
167 # Character positions
168 for i in range(1, len(positions) + 1):
169 char = positions[i]
170 targets = list(follow[i])
171 if i in root.last:
172 # If this is a terminal position, it MUST lead to the MATCH state
173 targets.append(match_state_pc)
174
175 p_any = 1 if char == "." else 0
176 p_char = f"'{char}'" if char != "." else 0
177
178 if not targets:
179 rasm.append(f"{pc_map[i]}: CHAR({p_char}, 0, 0, 0, 0, {p_any})")
180 elif len(targets) == 1:
181 t = pc_map[targets[0]] if targets[0] in pc_map else targets[0]
182 rasm.append(f"{pc_map[i]}: CHAR({p_char}, {t}, 0, 0, 0, {p_any})")
183 elif len(targets) == 2:
184 t1 = pc_map[targets[0]] if targets[0] in pc_map else targets[0]
185 t2 = pc_map[targets[1]] if targets[1] in pc_map else targets[1]
186 rasm.append(f"{pc_map[i]}: CHAR({p_char}, {t1}, {t2}, 0, 0, {p_any})")
187 else:
188 chain_start = next_free_pc
189 rasm.append(f"{pc_map[i]}: CHAR({p_char}, {chain_start}, 0, 0, 0, {p_any})")
190 chain, next_free_pc = write_split_chain(
191 targets, chain_start, next_free_pc + 1
192 )
193 rasm.extend(chain)
194
195 return rasm, next_free_pc
196
197
198def main():
199 import sys
200
201 input_file = sys.argv[1] if len(sys.argv) > 1 else "regex.txt"
202 output_file = sys.argv[2] if len(sys.argv) > 2 else "regexes.rasm"
203
204 if not os.path.exists(input_file):
205 print(f"Error: {input_file} not found.")
206 return
207 with open(input_file, "r") as f:
208 # Limit to 16 regexes max
209 patterns = [l.strip() for l in f if l.strip() and not l.startswith("#")][:16]
210
211 if not patterns:
212 print("No patterns found in input file.")
213 return
214
215 # Reserve space at the beginning for the "Start Chain" (SPLITs)
216 # If we have N patterns, we need N-1 SPLIT instructions.
217 # We will start the actual regex logic at base_pc = N (to be safe).
218 num_patterns = len(patterns)
219 base_pc = num_patterns
220
221 regex_rasm_blocks = []
222 regex_start_pcs = []
223 current_pc = base_pc
224
225 for i, pat in enumerate(patterns):
226 try:
227 root = parse_regex(pat)
228 lines, next_pc = generate_rasm(root, i, current_pc)
229 regex_rasm_blocks.append((f"# Regex {i}: {pat}", lines))
230 regex_start_pcs.append(current_pc) # The start state for this regex
231 current_pc = next_pc
232 except Exception as e:
233 print(f"Error compiling '{pat}': {e}")
234
235 full_rasm = []
236
237 # Generate the Start Chain at PC 0
238 # PC 0: SPLIT(Regex0_Start, PC 1)
239 # PC 1: SPLIT(Regex1_Start, PC 2)
240 # ...
241 # PC N-2: SPLIT(RegexN-2_Start, RegexN-1_Start)
242
243 if num_patterns == 1:
244 full_rasm.append(f"0: JMP({regex_start_pcs[0]})")
245 else:
246 for i in range(num_patterns - 1):
247 target_regex_pc = regex_start_pcs[i]
248 next_chain_pc = i + 1 if i < num_patterns - 2 else regex_start_pcs[i + 1]
249 full_rasm.append(f"{i}: SPLIT({target_regex_pc}, {next_chain_pc})")
250
251 full_rasm.append("")
252 for header, lines in regex_rasm_blocks:
253 full_rasm.append(header)
254 full_rasm.extend(lines)
255 full_rasm.append("")
256
257 with open(output_file, "w") as f:
258 f.write("\n".join(full_rasm))
259 print(
260 f"Glushkov compilation successful. Generated {current_pc} instructions for {num_patterns} patterns from {input_file} to {output_file}."
261 )
262
263
264if __name__ == "__main__":
265 main()