MT-7273: Optimize Message parsing, copying and encoding for large messages

bench/README.md

@@ -0,0 +1,30 @@
+# katcp message-codec benchmarks (MT-7273)
+
+Verification and benchmarking scripts for the `katcp/core.py` large-message fast paths (parse / copy / encode).
+
+Scripts resolve the target repository via the `KATCP_REPO` environment variable, defaulting to the parent repository containing this `bench/` directory.
+
+(The pre-existing `benchtx*`/`runner.py`/`results.py` scripts in this directory are legacy twisted-based benchmarks, unrelated to these tools.)
+
+### Requirements
+
+Requires repository dependencies (`future`, `tornado>=4.3,<5` for e2e tests).
+
+### Scripts
+
+| Script | Description | Run Time |
+|---|---|---|
+| `ab_compare.py` | Compares baseline (default `master` via `git worktree`) against working tree. Prints per-operation speedup factors. Options: `--e2e`, `--ref <sha>`, `--py <python>`. | ~1 min |
+| `fuzz_parse_branch.py` | Differential fuzz test: 300k generated lines against both the working tree parser and pristine master pulled from git history. Exits 1 on any divergence. Custom ref via `KATCP_BASELINE_REF`. | ~2 min |
+| `run_core_tests.py` | Runs katcp core + BNF grammar test suites (62 tests). | seconds |
+| `bench_katcp.py` | Standalone parse/encode/init/copy timings across channel counts for current checkout. | ~30 s |
+| `e2e_proxy_bench.py` | Local client → mini-proxy → mock device chain to measure serial vs pipelined per-message latency. Options: `-n`, `--chans`. | ~1 min |
+| `devm_repro.py` | Live test against a running CBF+ proxy. Queries and re-sets existing gains to keep state neutral. Requires `--port <proxy-port>` (see the proxy's `--addr` on its command line); `--smoke` (24 msgs), default 880 msgs. | ~5 min |
+| `test_latency_target.py` | Gain-set timing against either the CBF+ proxy (`?wide-gain`) or the product controller (`?gain`), so the two can be subtracted to isolate proxy overhead. Input labels discovered from the proxy, or given via `--inputs` (required for the product). `--serial` for per-request latency, default pipelined. Needs `aiokatcp`, `numpy`, `uvloop` — use the katsdpcontroller venv. | ~1 min per run |
+
+### Quick Start
+
+```bash
+cd bench
+python ab_compare.py
+python fuzz_parse_branch.py

bench/_repo.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+"""Locate the katcp-python repository root.
+
+Checks KATCP_REPO env var, git root of bench/, or falls back to the parent directory.
+"""
+from __future__ import absolute_import
+import os
+import subprocess
+
+
+def katcp_repo():
+    env = os.environ.get('KATCP_REPO')
+    if env:
+        return os.path.abspath(env)
+    here = os.path.dirname(os.path.abspath(__file__))
+    try:
+        out = subprocess.check_output(
+            ['git', '-C', here, 'rev-parse', '--show-toplevel'],
+            stderr=subprocess.STDOUT)
+        return out.decode('utf-8').strip()
+    except Exception:
+        return os.path.dirname(here)

bench/ab_compare.py

@@ -0,0 +1,115 @@
+# -*- coding: utf-8 -*-
+"""MT-7273 A/B runner: benchmark baseline ref vs working tree.
+
+Creates a temporary git worktree for the baseline ref (default: master),
+runs the benchmark against both trees, and prints a side-by-side comparison
+table with speedup ratios.
+
+Usage:
+    python ab_compare.py               # micro-benchmark (parse/encode/copy)
+    python ab_compare.py --e2e         # include end-to-end proxy benchmark
+    python ab_compare.py --ref <sha>   # custom baseline ref
+    python ab_compare.py --py <python> # python interpreter path
+"""
+
+from __future__ import print_function
+import argparse
+import os
+import re
+import subprocess
+import sys
+import tempfile
+
+from _repo import katcp_repo
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = katcp_repo()
+DEFAULT_PY = sys.executable
+
+ROW_RE = re.compile(r'^\s*(\d+)\s')
+
+
+def run_bench(py, script, katcp_path, extra=()):
+    env = dict(os.environ, KATCP_REPO=katcp_path)
+    out = subprocess.check_output([py, os.path.join(HERE, script)] + list(extra),
+                                  env=env, stderr=subprocess.STDOUT)
+    return out.decode('utf-8', 'replace')
+
+
+def parse_micro(out):
+    """-> {nchans: (parse_ms, encode_ms, init_ms, copy_ms)}"""
+    rows = {}
+    for ln in out.splitlines():
+        m = ROW_RE.match(ln)
+        parts = ln.split()
+        if m and len(parts) == 6:
+            rows[int(parts[0])] = tuple(float(x) for x in parts[2:])
+    return rows
+
+
+def parse_e2e(out):
+    """-> {(nchans, mode): per_msg_ms}"""
+    rows = {}
+    for ln in out.splitlines():
+        parts = ln.split()
+        if len(parts) == 5 and ROW_RE.match(ln) and parts[1] in ('serial', 'pipelined'):
+            rows[(int(parts[0]), parts[1])] = float(parts[3])
+    return rows
+
+
+def ratio(a, b):
+    return ('%.1fx' % (a / b)) if b else '-'
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument('--ref', default='master', help='baseline git ref')
+    ap.add_argument('--py', default=DEFAULT_PY)
+    ap.add_argument('--e2e', action='store_true')
+    ap.add_argument('--e2e-n', type=int, default=100)
+    opts = ap.parse_args()
+
+    tmpdir = tempfile.mkdtemp(prefix='mt7273-ab-')
+    baseline_tree = os.path.join(tmpdir, 'baseline')
+    subprocess.check_call(['git', '-C', REPO, 'worktree', 'add', '--detach',
+                           baseline_tree, opts.ref])
+    try:
+        head = subprocess.check_output(
+            ['git', '-C', REPO, 'rev-parse', '--short', 'HEAD']).decode().strip()
+        base = subprocess.check_output(
+            ['git', '-C', REPO, 'rev-parse', '--short', opts.ref]).decode().strip()
+        print('baseline: %s (%s)   fix: checkout HEAD (%s)\n'
+              % (opts.ref, base, head))
+
+        print('== micro-benchmark (ms) ==')
+        a = parse_micro(run_bench(opts.py, 'bench_katcp.py', baseline_tree))
+        b = parse_micro(run_bench(opts.py, 'bench_katcp.py', REPO))
+        print('%8s | %22s | %22s | %22s' % ('nchans', 'parse (base/fix/x)',
+                                            'encode (base/fix/x)', 'copy (base/fix/x)'))
+        for n in sorted(set(a) & set(b)):
+            pa, ea, _, ca = a[n]
+            pb, eb, _, cb = b[n]
+            print('%8d | %9.4f %7.4f %4s | %9.4f %7.4f %4s | %9.4f %7.4f %4s'
+                  % (n, pa, pb, ratio(pa, pb), ea, eb, ratio(ea, eb),
+                     ca, cb, ratio(ca, cb)))
+
+        if opts.e2e:
+            print('\n== end-to-end proxied round trip (ms/msg) ==')
+            extra = ['-n', str(opts.e2e_n)]
+            ea_ = parse_e2e(run_bench(opts.py, 'e2e_proxy_bench.py',
+                                      baseline_tree, extra))
+            eb_ = parse_e2e(run_bench(opts.py, 'e2e_proxy_bench.py',
+                                      REPO, extra))
+            print('%8s %10s | %10s %10s %6s' % ('nchans', 'mode', 'baseline',
+                                                'fix', 'x'))
+            for key in sorted(set(ea_) & set(eb_)):
+                print('%8d %10s | %10.2f %10.2f %6s'
+                      % (key[0], key[1], ea_[key], eb_[key],
+                         ratio(ea_[key], eb_[key])))
+    finally:
+        subprocess.call(['git', '-C', REPO, 'worktree', 'remove', '--force',
+                         baseline_tree])
+
+
+if __name__ == '__main__':
+    main()

bench/bench_katcp.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+"""katcp Message parse/encode/copy/init micro-benchmark vs payload size.
+
+Benchmarks the katcp checkout resolved by _repo.katcp_repo() (KATCP_REPO
+env var, else the repo containing this bench/ directory).
+
+Usage:
+    python bench_katcp.py
+"""
+from __future__ import print_function
+import sys
+import timeit
+
+from _repo import katcp_repo
+
+sys.path.insert(0, katcp_repo())
+
+from katcp.core import Message, MessageParser  # noqa: E402
+
+PARSER = MessageParser()
+
+CHANNEL_COUNTS = [1, 256, 1024, 4096, 32768]
+
+
+def make_gain_value(i):
+    """One complex gain as wire bytes, varied slightly per channel."""
+    real = 0.9 + (i % 7) * 0.01
+    imag = 0.05 + (i % 5) * 0.01
+    return ("%.4f+%.4fj" % (real, imag)).encode('ascii')
+
+
+def make_gain_line(nchans, mid=123):
+    """Wire-format ?gain request: stream, input, then one complex gain per channel."""
+    vals = [make_gain_value(i) for i in range(nchans)]
+    args = [b"wide.antenna-channelised-voltage", b"m001h"] + vals
+    line = b"?wide-gain[%d] " % mid + b" ".join(args)
+    return line, args
+
+
+def bench(fn, number=None):
+    """Return best-of-3 seconds per call."""
+    if number is None:
+        # autoscale: aim for ~0.2s per repeat
+        t = timeit.timeit(fn, number=1)
+        number = max(1, int(0.2 / max(t, 1e-9)))
+    best = min(timeit.repeat(fn, number=number, repeat=3))
+    return best / number
+
+
+def main():
+    print("Python %s  repo %s" % (sys.version.split()[0], katcp_repo()))
+    hdr = ("nchans", "line_kB", "parse_ms", "encode_ms", "init_ms", "copy_ms")
+    print("%8s %8s %10s %10s %10s %10s" % hdr)
+    for n in CHANNEL_COUNTS:
+        line, args = make_gain_line(n)
+        msg = PARSER.parse(line)
+
+        parse_s = bench(lambda line=line: PARSER.parse(line))
+        encode_s = bench(lambda msg=msg: msg.__bytes__())
+        init_s = bench(lambda msg=msg: Message(Message.REQUEST, "wide-gain",
+                                               msg.arguments, b"123"))
+        copy_s = bench(lambda msg=msg: msg.copy())
+
+        print("%8d %8.1f %10.4f %10.4f %10.4f %10.4f" % (
+            n, len(line) / 1024.0,
+            parse_s * 1e3, encode_s * 1e3, init_s * 1e3, copy_s * 1e3))
+
+    # Baseline small message comparison
+    small = b"!wide-gain[123] ok"
+    small_s = bench(lambda: PARSER.parse(small))
+    print("\nsmall reply '!wide-gain[123] ok' parse: %.1f us" % (small_s * 1e6))
+
+
+if __name__ == '__main__':
+    main()