Null result: A hand-written character-scanning parser will parse a…

import sys
import re
import time
import random
import statistics
import tracemalloc
import resource
import json

# ---------- Synthetic log generation ----------

METHODS = ["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS"]
PATHS = ["/", "/index.html", "/api/v1/users", "/api/v1/items", "/static/app.js",
 "/static/style.css", "/images/logo.png", "/health", "/metrics",
 "/api/v2/search", "/login", "/logout", "/dashboard"]
PROTOS = ["HTTP/1.0", "HTTP/1.1", "HTTP/2.0"]
STATUSES = [200, 200, 200, 200, 201, 204, 301, 302, 304, 400, 401, 403, 404, 500, 502, 503]
UAS = [
 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15",
 "curl/7.68.0",
 "python-requests/2.28.1",
 "Googlebot/2.1 (+http://www.google.com/bot.html)",
]

def make_ip(rng):
 return "%d.%d.%d.%d" % (rng.randint(1, 223), rng.randint(0, 255),
 rng.randint(0, 255), rng.randint(1, 254))

def generate_log(n_lines, seed=1337):
 rng = random.Random(seed)
 lines = []
 for _ in range(n_lines):
 ip = make_ip(rng)
 # Apache-ish timestamp
 day = rng.randint(1, 28)
 mon = rng.choice(["Jan", "Feb", "Mar", "Apr", "May", "Jun",
 "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"])
 year = rng.choice([2022, 2023, 2024])
 hh = rng.randint(0, 23)
 mm = rng.randint(0, 59)
 ss = rng.randint(0, 59)
 tz = rng.choice(["+0000", "-0500", "+0200"])
 ts = "%02d/%s/%d:%02d:%02d:%02d %s" % (day, mon, year, hh, mm, ss, tz)
 method = rng.choice(METHODS)
 path = rng.choice(PATHS)
 proto = rng.choice(PROTOS)
 status = rng.choice(STATUSES)
 size = rng.randint(0, 50000)
 ua = rng.choice(UAS)
 line = '%s - - [%s] "%s %s %s" %d %d "%s"' % (
 ip, ts, method, path, proto, status, size, ua)
 lines.append(line)
 return lines

# ---------- Parsers ----------

# Compiled regex (naive-ish but reasonable)
LOG_RE = re.compile(
 r'^(?P<ip>\S+) \S+ \S+ \[(?P<ts>[^\]]+)\] '
 r'"(?P<method>\S+) (?P<path>\S+) (?P<proto>[^"]+)" '
 r'(?P<status>\d{3}) (?P<size>\d+) "(?P<ua>[^"]*)"$'
)

def parse_regex(line):
 m = LOG_RE.match(line)
 if m is None:
 return None
 g = m.groupdict()
 return (g["ip"], g["ts"], g["method"], g["path"], g["proto"],
 int(g["status"]), int(g["size"]), g["ua"])

def parse_split(line):
 # Split-based extraction
 try:
 # ip - - [ts] "method path proto" status size "ua"
 # First split on space, but ts has spaces inside brackets.
 # Use a pragmatic split approach.
 parts = line.split(" ")
 ip = parts[0]
 # find bracket
 lb = line.index("[")
 rb = line.index("]")
 ts = line[lb + 1:rb]
 # quoted request
 q1 = line.index('"')
 q2 = line.index('"', q1 + 1)
 req = line[q1 + 1:q2]
 rparts = req.split(" ")
 method, path, proto = rparts[0], rparts[1], rparts[2]
 # after closing quote
 rest = line[q2 + 2:]
 # rest = status size "ua"
 rq1 = rest.index('"')
 rq2 = rest.rindex('"')
 ua = rest[rq1 + 1:rq2]
 pre = rest[:rq1].strip()
 sp = pre.split(" ")
 status = int(sp[0])
 size = int(sp[1])
 return (ip, ts, method, path, proto, status, size, ua)
 except (ValueError, IndexError):
 return None

def parse_scanner(line):
 # Hand-written character scanner
 n = len(line)
 i = 0
 # ip
 start = i
 while i < n and line[i] != ' ':
 i += 1
 ip = line[start:i]
 # skip two tokens ("- -")
 for _ in range(2):
 while i < n and line[i] == ' ':
 i += 1
 while i < n and line[i] != ' ':
 i += 1
 # skip space, expect '['
 while i < n and line[i] == ' ':
 i += 1
 if i >= n or line[i] != '[':
 return None
 i += 1
 start = i
 while i < n and line[i] != ']':
 i += 1
 ts = line[start:i]
 i += 1 # skip ']'
 # skip space, expect '"'
 while i < n and line[i] == ' ':
 i += 1
 if i >= n or line[i] != '"':
 return None
 i += 1
 # method
 start = i
 while i < n and line[i] != ' ':
 i += 1
 method = line[start:i]
 i += 1
 # path
 start = i
 while i < n and line[i] != ' ':
 i += 1
 path = line[start:i]
 i += 1
 # proto
 start = i
 while i < n and line[i] != '"':
 i += 1
 proto = line[start:i]
 i += 1 # skip '"'
 # skip space
 while i < n and line[i] == ' ':
 i += 1
 # status
 start = i
 while i < n and line[i] != ' ':
 i += 1
 status = int(line[start:i])
 i += 1
 # size
 start = i
 while i < n and line[i] != ' ':
 i += 1
 size = int(line[start:i])
 i += 1
 # skip space, expect '"'
 while i < n and line[i] == ' ':
 i += 1
 if i >= n or line[i] != '"':
 return None
 i += 1
 start = i
 while i < n and line[i] != '"':
 i += 1
 ua = line[start:i]
 return (ip, ts, method, path, proto, status, size, ua)

PARSERS = {
 "regex": parse_regex,
 "split": parse_split,
 "scanner": parse_scanner,
}

# ---------- Benchmark helpers ----------

def run_parser(parser, lines):
 count = 0
 for line in lines:
 r = parser(line)
 if r is not None:
 count += 1
 return count

def measure(parser, lines, repeats=3):
 times = []
 counts = []
 for _ in range(repeats):
 t0 = time.perf_counter()
 c = run_parser(parser, lines)
 t1 = time.perf_counter()
 times.append(t1 - t0)
 counts.append(c)
 return times, counts

def peak_rss_kb():
 return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss

# ---------- Adversarial input ----------

def make_adversarial():
 # Long run of spaces inside the quoted request section, plus a long
 # run of quotes. Designed to trigger backtracking in the naive regex.
 spaces = " " * 20000
 quotes = '"' * 20000
 # A line that starts plausibly but then has a huge run of spaces where
 # the regex expects a single space between tokens.
 line1 = '1.2.3.4 - - [01/Jan/2024:00:00:00 +0000] "GET' + spaces + '/x HTTP/1.1" 200 1 "ua"'
 # A line with a huge run of quotes in the UA field
 line2 = '1.2.3.4 - - [01/Jan/2024:00:00:00 +0000] "GET /x HTTP/1.1" 200 1 "' + quotes + '"'
 return [line1, line2]

def time_adversarial(parser, adv_lines, repeats=3):
 times = []
 for _ in range(repeats):
 t0 = time.perf_counter()
 for line in adv_lines:
 try:
 parser(line)
 except Exception:
 pass
 t1 = time.perf_counter()
 times.append(t1 - t0)
 return times

# ---------- Main ----------

def main():
 N_LINES = 200_000
 REPEATS = 3

 print("[gen] generating %d synthetic log lines..." % N_LINES, file=sys.stderr)
 t0 = time.perf_counter()
 lines = generate_log(N_LINES)
 t1 = time.perf_counter()
 print("[gen] done in %.2fs" % (t1 - t0), file=sys.stderr)

 # Sanity check: all parsers agree on a sample
 sample = lines[:200]
 for name, p in PARSERS.items():
 ok = sum(1 for l in sample if p(l) is not None)
 print("[sanity] %s parsed %d/%d" % (name, ok, len(sample)), file=sys.stderr)

 # Warm up (JIT-ish, cache effects)
 for p in PARSERS.values():
 run_parser(p, lines[:1000])

 results = {}
 for name, p in PARSERS.items():
 print("[bench] running %s..." % name, file=sys.stderr)
 tracemalloc.start()
 times, counts = measure(p, lines, repeats=REPEATS)
 cur, peak = tracemalloc.get_traced_memory()
 tracemalloc.stop()
 rss = peak_rss_kb()
 med_t = statistics.median(times)
 lps = N_LINES / med_t if med_t > 0 else float("inf")
 results[name] = {
 "times_s": [float(x) for x in times],
 "median_time_s": float(med_t),
 "lines_per_second": float(lps),
 "parsed_count": int(counts[0]),
 "tracemalloc_peak_bytes": int(peak),
 "process_peak_rss_kb": int(rss),
 }
 print("[bench] %s: median %.3fs, %.0f lines/s, peak_rss=%dKB"
 % (name, med_t, lps, rss), file=sys.stderr)

 # Adversarial
 print("[adv] building adversarial input...", file=sys.stderr)
 adv = make_adversarial()
 adv_results = {}
 for name, p in PARSERS.items():
 print("[adv] timing %s..." % name, file=sys.stderr)
 try:
 at = time_adversarial(p, adv, repeats=REPEATS)
 med = statistics.median(at)
 adv_results[name] = {
 "times_s": [float(x) for x in at],
 "median_time_s": float(med),
 }
 print("[adv] %s: median %.4fs" % (name, med), file=sys.stderr)
 except Exception as e:
 adv_results[name] = {"error": str(e)}
 print("[adv] %s failed: %s" % (name, e), file=sys.stderr)

 # Evaluate hypothesis
 scanner_lps = results["scanner"]["lines_per_second"]
 regex_lps = results["regex"]["lines_per_second"]
 split_lps = results["split"]["lines_per_second"]
 scanner_rss = results["scanner"]["process_peak_rss_kb"]
 regex_rss = results["regex"]["process_peak_rss_kb"]
 split_rss = results["split"]["process_peak_rss_kb"]

 fastest = max(results.items(), key=lambda kv: kv[1]["lines_per_second"])[0]
 lowest_rss = min(results.items(), key=lambda kv: kv[1]["process_peak_rss_kb"])[0]

 adv_regex = adv_results.get("regex", {}).get("median_time_s")
 adv_scanner = adv_results.get("scanner", {}).get("median_time_s")
 adv_ratio = None
 if adv_regex is not None and adv_scanner is not None and adv_scanner > 0:
 adv_ratio = adv_regex / adv_scanner

 speed_ok = (fastest == "scanner")
 rss_ok = (lowest_rss == "scanner")
 adv_ok = (adv_ratio is not None and adv_ratio >= 10.0)

 supported = bool(speed_ok and rss_ok and adv_ok)

 evidence = {
 "n_lines": N_LINES,
 "repeats": REPEATS,
 "parsers": results,
 "fastest_parser": fastest,
 "lowest_rss_parser": lowest_rss,
 "adversarial": adv_results,
 "adversarial_regex_vs_scanner_ratio": (float(adv_ratio) if adv_ratio is not None else None),
 "criteria": {
 "scanner_fastest": bool(speed_ok),
 "scanner_lowest_rss": bool(rss_ok),
 "adversarial_regex_10x_slower": bool(adv_ok),
 },
 }

 interpretation = (
 "Scanner median %.0f lines/s vs regex %.0f and split %.0f; "
 "scanner peak RSS %dKB vs regex %dKB and split %dKB. "
 "Adversarial regex/scanner time ratio: %s."
 % (scanner_lps, regex_lps, split_lps,
 scanner_rss, regex_rss, split_rss,
 ("%.2f" % adv_ratio) if adv_ratio is not None else "n/a")
 )

 result = {
 "hypothesis": (
 "A hand-written character-scanning parser parses a synthetic 200k-line "
 "Apache-style access log faster (higher lines/second) and with lower peak "
 "memory than both `re` and `str.split`, and a crafted pathological input "
 "causes catastrophic backtracking in a naive `re` pattern, making it at "
 "least 10x slower than the hand-written parser."
 ),
 "hypothesis_supported": supported,
 "evidence": evidence,
 "interpretation": interpretation,
 }

 print(json.dumps(result, indent=2,
 default=lambda o: o.item() if hasattr(o, "item") else str(o)))
 return 0

if __name__ == "__main__":
 sys.exit(main())