diff --git a/bench/README.md b/bench/README.md
new file mode 100644
index 00000000..3390cff8
--- /dev/null
+++ b/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
diff --git a/bench/_repo.py b/bench/_repo.py
new file mode 100644
index 00000000..5518edeb
--- /dev/null
+++ b/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)
diff --git a/bench/ab_compare.py b/bench/ab_compare.py
new file mode 100644
index 00000000..eeb3dc52
--- /dev/null
+++ b/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()
diff --git a/bench/bench_katcp.py b/bench/bench_katcp.py
new file mode 100644
index 00000000..e10d4335
--- /dev/null
+++ b/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()
diff --git a/bench/devm_repro.py b/bench/devm_repro.py
new file mode 100644
index 00000000..1bf6e736
--- /dev/null
+++ b/bench/devm_repro.py
@@ -0,0 +1,181 @@
+# -*- coding: utf-8 -*-
+"""MT-7273 devbox CBF+ proxy reproduction script.
+
+Queries existing per-input gains and re-applies those same values, keeping state
+neutral while exercising large (~60 KiB) message latency against the proxy.
+
+Stdlib only (raw sockets) to ensure client-side overhead is negligible.
+
+The proxy's katcp port is required. Find it from the proxy's command line,
+e.g.:  ps aux | grep data-cbfplus-proxy   ->  ... --addr :5056
+
+Usage:
+    python devm_repro.py --port <proxy-port>              # 880 msgs, serial
+    python devm_repro.py --port <proxy-port> --smoke      # quick 3-pass run
+    python devm_repro.py --port <proxy-port> --pipelined  # all msgs in flight
+"""
+from __future__ import print_function
+import argparse
+import socket
+import time
+
+STREAM_CANDIDATES = [
+    b"wide-antenna-channelised-voltage",
+    b"wide.antenna-channelised-voltage",
+    b"antenna-channelised-voltage",
+]
+
+
+class Katcp(object):
+    def __init__(self, host, port):
+        self.sock = socket.create_connection((host, port), timeout=30)
+        self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        self.buf = b""
+        time.sleep(0.5)
+        self.drain()
+        self.mid = 0
+
+    def drain(self):
+        self.sock.setblocking(False)
+        try:
+            while True:
+                d = self.sock.recv(65536)
+                if not d:
+                    break
+                self.buf += d
+        except Exception:
+            pass
+        self.sock.setblocking(True)
+        self.sock.settimeout(30)
+        self.buf = b""
+
+    def readline(self):
+        while True:
+            nl = self.buf.find(b"\n")
+            if nl >= 0:
+                line = self.buf[:nl]
+                self.buf = self.buf[nl + 1:]
+                return line.replace(b"\r", b"")
+            d = self.sock.recv(65536)
+            if not d:
+                raise RuntimeError("connection closed")
+            self.buf += d
+
+    def send(self, name, args):
+        """Send ?name[mid] args...; return the mid used."""
+        self.mid += 1
+        line = b"?%s[%d]" % (name, self.mid)
+        if args:
+            line += b" " + b" ".join(args)
+        self.sock.sendall(line + b"\n")
+        return self.mid
+
+    def wait_reply(self, name, mid):
+        """Collect informs until the !name[mid] reply; return (reply, informs)."""
+        informs = []
+        want_r = b"!%s[%d]" % (name, mid)
+        want_i = b"#%s[%d]" % (name, mid)
+        while True:
+            line = self.readline()
+            if line.startswith(want_r):
+                return line, informs
+            if line.startswith(want_i):
+                informs.append(line)
+
+    def request(self, name, args=()):
+        mid = self.send(name, list(args))
+        return self.wait_reply(name, mid)
+
+
+def detect_stream(kc, an_input):
+    """Find the stream name the proxy's wide-gain request accepts."""
+    for stream in STREAM_CANDIDATES:
+        reply, _ = kc.request(b"wide-gain", [stream, an_input])
+        if b" ok" in reply or reply.split(b" ")[1:2] == [b"ok"]:
+            return stream, reply
+    raise RuntimeError("no stream candidate accepted; last reply: %r" % reply)
+
+
+def query_gains(kc, stream, inp, nchans):
+    """Query current gains for an input; expand a uniform single value."""
+    reply, _ = kc.request(b"wide-gain", [stream, inp])
+    parts = reply.split(b" ")
+    assert parts[1] == b"ok", "gain query failed: %r" % reply
+    vals = parts[2:]
+    if len(vals) == 1:
+        vals = vals * nchans
+    assert len(vals) == nchans, "expected %d gains, got %d" % (nchans, len(vals))
+    return vals
+
+
+def pct(sorted_vals, p):
+    return sorted_vals[min(len(sorted_vals) - 1, int(len(sorted_vals) * p))]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument('--host', default='localhost')
+    ap.add_argument('--port', type=int, required=True,
+                    help='katcp port of the CBF+ proxy under test')
+    ap.add_argument('--passes', type=int, default=110)
+    ap.add_argument('--nchans', type=int, default=4096)
+    ap.add_argument('--pipelined', action='store_true',
+                    help='send everything, then collect all replies')
+    ap.add_argument('--smoke', action='store_true', help='3 passes only')
+    opts = ap.parse_args()
+    if opts.smoke:
+        opts.passes = 3
+
+    kc = Katcp(opts.host, opts.port)
+
+    reply, informs = kc.request(b"sensor-value", [b"input-labels"])
+    assert b" ok" in reply, "sensor-value input-labels failed: %r" % reply
+    inputs = informs[0].split(b" ")[-1].split(b",")
+    print("inputs: %s" % b",".join(inputs).decode())
+
+    stream, _ = detect_stream(kc, inputs[0])
+    print("stream accepted by wide-gain: %s" % stream.decode())
+
+    print("Querying current gains...")
+    t0 = time.time()
+    gains = {}
+    for inp in inputs:
+        gains[inp] = query_gains(kc, stream, inp, opts.nchans)
+    print("  %d queries took %.2fs" % (len(inputs), time.time() - t0))
+
+    total = opts.passes * len(inputs)
+    line_len = len(b" ".join([stream, inputs[0]] + gains[inputs[0]]))
+    print("Running %d passes x %d inputs = %d set-gain messages (~%.0f KiB each), %s..."
+          % (opts.passes, len(inputs), total, line_len / 1024.0,
+             "pipelined" if opts.pipelined else "serial"))
+
+    lat = []
+    t0 = time.time()
+    if opts.pipelined:
+        mids = []
+        for _p in range(opts.passes):
+            for inp in inputs:
+                mids.append(kc.send(b"wide-gain", [stream, inp] + gains[inp]))
+        for mid in mids:
+            reply, _ = kc.wait_reply(b"wide-gain", mid)
+            assert b" ok" in reply, reply[:200]
+    else:
+        for _p in range(opts.passes):
+            for inp in inputs:
+                t1 = time.time()
+                mid = kc.send(b"wide-gain", [stream, inp] + gains[inp])
+                reply, _ = kc.wait_reply(b"wide-gain", mid)
+                lat.append(time.time() - t1)
+                assert b" ok" in reply, reply[:200]
+    dt = time.time() - t0
+
+    print("Total: %.2fs for %d messages -> %.1f ms/msg, %.1f msg/s"
+          % (dt, total, dt / total * 1e3, total / dt))
+    if lat:
+        lat.sort()
+        print("Per-message latency: p50 %.1f ms  p95 %.1f ms  max %.1f ms"
+              % (pct(lat, 0.50) * 1e3, pct(lat, 0.95) * 1e3, lat[-1] * 1e3))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/bench/e2e_proxy_bench.py b/bench/e2e_proxy_bench.py
new file mode 100644
index 00000000..08909f1b
--- /dev/null
+++ b/bench/e2e_proxy_bench.py
@@ -0,0 +1,173 @@
+# -*- coding: utf-8 -*-
+"""MT-7273 end-to-end proxy benchmark script.
+
+Chain: test client -> mini-proxy -> device server
+
+Simulates KATCP proxied request flow (copy -> rename -> callback -> reply)
+with immediate device responses to measure proxy stack overhead.
+
+Usage:
+    python e2e_proxy_bench.py [-n N] [--chans C1,C2,...]
+"""
+from __future__ import print_function
+import argparse
+import socket
+import sys
+import time
+
+from _repo import katcp_repo
+sys.path.insert(0, katcp_repo())
+
+parser = argparse.ArgumentParser()
+parser.add_argument('-n', type=int, default=200, help='messages per run')
+parser.add_argument('--chans', default='1,4096')
+cli_args = parser.parse_args()
+
+from katcp import AsyncReply, CallbackClient, DeviceServer  # noqa: E402
+
+
+class Device(DeviceServer):
+    VERSION_INFO = ("mt7273-device", 0, 1)
+    BUILD_INFO = ("mt7273-device", 0, 1, "")
+
+    def setup_sensors(self):
+        pass
+
+    def request_wide_gain(self, req, msg):
+        """Accept gains and reply ok immediately."""
+        return req.make_reply("ok")
+
+
+class MiniProxy(DeviceServer):
+    """Proxy handler mimicking katcore DeviceHandler request flow."""
+
+    VERSION_INFO = ("mt7273-proxy", 0, 1)
+    BUILD_INFO = ("mt7273-proxy", 0, 1, "")
+
+    def __init__(self, device_addr, *a, **kw):
+        super(MiniProxy, self).__init__(*a, **kw)
+        self._dev_client = CallbackClient(device_addr[0], device_addr[1])
+
+    def setup_sensors(self):
+        pass
+
+    def start(self, *a, **kw):
+        self._dev_client.start()
+        self._dev_client.wait_protocol(timeout=5)
+        super(MiniProxy, self).start(*a, **kw)
+
+    def stop(self, *a, **kw):
+        super(MiniProxy, self).stop(*a, **kw)
+        self._dev_client.stop()
+
+    def request_wide_gain(self, req, msg):
+        """Forward gains to the device and relay its reply."""
+        devmsg = msg.copy()
+        devmsg.name = "wide-gain"
+
+        def reply_cb(devreply, orig_msg, orig_req):
+            m = devreply.copy()
+            m.name = orig_msg.name
+            orig_req.reply_with_message(m)
+
+        self._dev_client.callback_request(
+            devmsg,
+            reply_cb=lambda r: reply_cb(r, msg, req),
+            timeout=30,
+        )
+        raise AsyncReply()
+
+
+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_line(nchans, mid):
+    vals = [make_gain_value(i) for i in range(nchans)]
+    return (b"?wide-gain[%d] wide.antenna-channelised-voltage m001h " % mid
+            + b" ".join(vals) + b"\n")
+
+
+class RawClient(object):
+    def __init__(self, addr):
+        self.sock = socket.create_connection(addr)
+        self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        self.buf = b""
+
+    def read_reply(self):
+        """Read lines until a !wide-gain reply arrives."""
+        while True:
+            nl = self.buf.find(b"\n")
+            if nl >= 0:
+                line, self.buf = self.buf[:nl], self.buf[nl + 1:]
+                if line.startswith(b"!wide-gain"):
+                    return line
+                continue
+            data = self.sock.recv(65536)
+            if not data:
+                raise RuntimeError("connection closed")
+            self.buf += data
+
+    def close(self):
+        self.sock.close()
+
+
+def run_serial(addr, nchans, n):
+    c = RawClient(addr)
+    t0 = time.time()
+    for i in range(n):
+        c.sock.sendall(make_line(nchans, i + 1))
+        c.read_reply()
+    dt = time.time() - t0
+    c.close()
+    return dt
+
+
+def run_pipelined(addr, nchans, n):
+    c = RawClient(addr)
+    lines = [make_line(nchans, i + 1) for i in range(n)]
+    t0 = time.time()
+    c.sock.sendall(b"".join(lines))
+    for _ in range(n):
+        c.read_reply()
+    dt = time.time() - t0
+    c.close()
+    return dt
+
+
+def main():
+    n = cli_args.n
+    chan_list = [int(x) for x in cli_args.chans.split(',')]
+
+    device = Device("127.0.0.1", 0)
+    device.start(timeout=5)
+    dev_addr = device.bind_address
+
+    proxy = MiniProxy(dev_addr, "127.0.0.1", 0)
+    proxy.start(timeout=5)
+    proxy_addr = proxy.bind_address
+    time.sleep(0.5)
+
+    print("python %s  n=%d" % (sys.version.split()[0], n))
+    print("%8s %10s %14s %14s %16s" % (
+        "nchans", "mode", "total_s", "per_msg_ms", "msgs_per_s"))
+    try:
+        for nchans in chan_list:
+            # Warmup run
+            run_serial(proxy_addr, nchans, 5)
+            dt = run_serial(proxy_addr, nchans, n)
+            print("%8d %10s %14.2f %14.2f %16.1f" % (
+                nchans, "serial", dt, dt / n * 1e3, n / dt))
+            dt = run_pipelined(proxy_addr, nchans, n)
+            print("%8d %10s %14.2f %14.2f %16.1f" % (
+                nchans, "pipelined", dt, dt / n * 1e3, n / dt))
+    finally:
+        proxy.stop()
+        device.stop()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/bench/fuzz_parse_branch.py b/bench/fuzz_parse_branch.py
new file mode 100644
index 00000000..c105551c
--- /dev/null
+++ b/bench/fuzz_parse_branch.py
@@ -0,0 +1,94 @@
+# -*- coding: utf-8 -*-
+"""Differential fuzz test: baseline MessageParser vs updated parser.
+
+Extracts baseline katcp modules directly from git into a temporary package
+to compare outputs (including exception types and messages) against current HEAD.
+
+Usage:
+    python fuzz_parse_branch.py
+"""
+from __future__ import print_function
+import os
+import random
+import shutil
+import subprocess
+import sys
+import tempfile
+
+from _repo import katcp_repo
+
+KATCP_REPO = katcp_repo()
+BASELINE_REF = os.environ.get('KATCP_BASELINE_REF', 'master')
+sys.path.insert(0, KATCP_REPO)
+
+# Extract baseline katcp modules into temporary pristinekatcp package
+tmpdir = tempfile.mkdtemp(prefix='mt7273-pristine-')
+pkg = os.path.join(tmpdir, 'pristinekatcp')
+os.mkdir(pkg)
+open(os.path.join(pkg, '__init__.py'), 'w').close()
+
+for fname in ('core.py', 'compat.py', 'kattypes.py', 'sampling.py'):
+    src = subprocess.check_output(
+        ['git', 'show', '%s:katcp/%s' % (BASELINE_REF, fname)], cwd=KATCP_REPO)
+    with open(os.path.join(pkg, fname), 'wb') as f:
+        f.write(src)
+sys.path.insert(0, tmpdir)
+
+from katcp.core import MessageParser  # noqa: E402
+
+from pristinekatcp.core import MessageParser as OrigParser  # noqa: E402
+
+p_orig = OrigParser()
+p_fast = MessageParser()
+
+random.seed(1234)
+
+CHARS = [b"a", b"z", b"0", b"9", b"-", b"+", b".", b"j", b"_",
+         b"\\_", b"\\\\", b"\\0", b"\\n", b"\\r", b"\\e", b"\\t", b"\\@",
+         b"\\", b"\\x", b"\0", b"\x1b", b"\t", b" "]
+
+NAMES = [b"wide-gain", b"gain", b"a", b"9bad", b"x-1", b"ba\0d", b"b\\_d", b""]
+
+
+def rand_arg():
+    n = random.randint(0, 12)
+    return b"".join(random.choice(CHARS) for _ in range(n))
+
+
+def rand_line():
+    t = random.choice([b"?", b"!", b"#", b"%", b""])
+    name = random.choice(NAMES)
+    mid = random.choice([b"", b"[123]", b"[0]", b"[x]", b"[]"])
+    nargs = random.randint(0, 6)
+    args = b" ".join(rand_arg() for _ in range(nargs))
+    line = t + name + mid
+    if args:
+        line += b" " + args
+    return line
+
+
+def outcome(parser, line):
+    try:
+        m = parser.parse(line)
+        return ("ok", m.mtype, m.name, m.mid, m.arguments)
+    except Exception as e:
+        return ("err", type(e).__name__, str(e))
+
+
+mismatches = 0
+N = 300000
+try:
+    for _i in range(N):
+        line = rand_line()
+        a = outcome(p_orig, line)
+        b = outcome(p_fast, line)
+        if a != b:
+            mismatches += 1
+            print("MISMATCH on %r:\n  orig: %r\n  new:  %r" % (line, a, b))
+            if mismatches > 10:
+                break
+    print("done: %d lines, %d mismatches" % (N, mismatches))
+finally:
+    shutil.rmtree(tmpdir, ignore_errors=True)
+
+sys.exit(1 if mismatches else 0)
diff --git a/bench/run_core_tests.py b/bench/run_core_tests.py
new file mode 100644
index 00000000..2f4670b3
--- /dev/null
+++ b/bench/run_core_tests.py
@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+"""Run katcp core + BNF test suites.
+
+Usage:
+    python run_core_tests.py
+"""
+from __future__ import print_function
+import sys
+import unittest
+
+from _repo import katcp_repo
+sys.path.insert(0, katcp_repo())
+
+loader = unittest.TestLoader()
+suite = unittest.TestSuite()
+for mod in ('katcp.test.test_core', 'katcp.test.test_katcp_bnf'):
+    __import__(mod)
+    suite.addTests(loader.loadTestsFromModule(sys.modules[mod]))
+
+result = unittest.TextTestRunner(verbosity=1).run(suite)
+sys.exit(0 if result.wasSuccessful() else 1)
diff --git a/bench/test_latency_target.py b/bench/test_latency_target.py
new file mode 100644
index 00000000..cb3d5e59
--- /dev/null
+++ b/bench/test_latency_target.py
@@ -0,0 +1,217 @@
+#!/usr/bin/env python3
+"""Gain-set latency/throughput against a CBF+ proxy or a CBF product directly.
+
+Derived from Bruce Merry's original aiokatcp latency script, corrected for use
+against a live product and parameterised so the identical client code, message
+construction and timing can be pointed at either end of the chain:
+
+  proxy    ?wide-gain on the CBF+ proxy          (katcp-python)
+  product  ?gain      on the product controller  (aiokatcp)
+
+The product's ?gain takes the same arguments as the proxy's ?wide-gain, so only
+the request name and the address differ. Running both and subtracting isolates
+what the proxy itself costs.
+
+Corrections to the original, all of which matter against a live product:
+  - real input labels, discovered from the proxy's input-labels sensor rather
+    than assumed (the original's inp0..inpN are rejected by a real product);
+  - gains formatted with an explicit '+' before non-negative imaginary parts,
+    without which roughly half of them are not valid complex values;
+  - reply status is checked, so failed requests cannot be timed as if they had
+    succeeded;
+  - existing gains are saved and restored, leaving product state as found.
+
+Requires aiokatcp, numpy and uvloop (all present in the katsdpcontroller venv).
+
+Usage:
+    # via the proxy (input labels discovered automatically)
+    python test_latency_target.py --port 5056
+
+    # direct to the product controller; labels must be given explicitly
+    python test_latency_target.py --host <cmc-host> --port <product-port> \
+        --request gain --inputs m011h,m011v,m022h,m022v
+
+    # per-request latency instead of pipelined throughput
+    python test_latency_target.py --port 5056 --serial
+"""
+
+import argparse
+import asyncio
+import logging
+import signal
+
+import aiokatcp
+import numpy as np
+import uvloop
+
+DEFAULT_STREAM = "wide-antenna-channelised-voltage"
+
+
+async def read_reply(reader, name):
+    """Read lines until the reply to `name` arrives, discarding informs."""
+    want = b"!" + name.encode()
+    while True:
+        line = await reader.readuntil()
+        if line.startswith(want):
+            return line
+
+
+async def discover_inputs(reader, writer):
+    """Read the input labels from the proxy's input-labels sensor.
+
+    Only works against the proxy; the product controller does not carry this
+    sensor, so --inputs must be given explicitly for that target.
+    """
+    writer.write(bytes(aiokatcp.Message.request("sensor-value", "input-labels")))
+    labels = None
+    while True:
+        line = await reader.readuntil()
+        if line.startswith(b"#sensor-value"):
+            labels = line.strip().split(b" ")[-1]
+        elif line.startswith(b"!sensor-value"):
+            break
+    if not labels or b"," not in labels:
+        raise RuntimeError(
+            "could not discover input labels (got %r); pass --inputs explicitly, "
+            "which is required when targeting the product controller" % labels)
+    return [label.decode() for label in labels.split(b",")]
+
+
+async def query_gains(reader, writer, request, stream, label):
+    """Return the current gains for one input, as wire-format byte strings."""
+    writer.write(bytes(aiokatcp.Message.request(request, stream, label)))
+    line = await read_reply(reader, request)
+    parts = line.strip().split(b" ")
+    if parts[1] != b"ok":
+        raise RuntimeError("gain query for %s failed: %r" % (label, line[:200]))
+    return parts[2:]
+
+
+async def set_gains(reader, writer, request, stream, label, vals):
+    """Write gains back for one input."""
+    writer.write(bytes(aiokatcp.Message.request(request, stream, label, *vals)))
+    line = await read_reply(reader, request)
+    if b" ok" not in line:
+        raise RuntimeError("gain restore for %s failed: %r" % (label, line[:200]))
+
+
+def make_gains(n_inputs, channels):
+    """Random complex64 gains, formatted as the wire expects."""
+    shape = (n_inputs, channels)
+    rng = np.random.default_rng(seed=2)
+    mag = rng.uniform(0.5, 2.0, size=shape)
+    phase = rng.uniform(-np.pi, np.pi, size=shape)
+    gains = (mag * np.exp(1j * phase)).astype(np.complex64)
+    return [[f"{g.real}{g.imag:+}j".encode() for g in row] for row in gains]
+
+
+async def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="localhost")
+    ap.add_argument("--port", type=int, required=True,
+                    help="katcp port of the proxy, or of the product controller")
+    ap.add_argument("--request", default="wide-gain",
+                    help="wide-gain for the proxy, gain for the product")
+    ap.add_argument("--stream", default=DEFAULT_STREAM)
+    ap.add_argument("--inputs", default=None,
+                    help="comma-separated input labels; discovered from the "
+                         "proxy if omitted, required for the product")
+    ap.add_argument("--passes", type=int, default=110,
+                    help="passes over all inputs; 110 x 8 inputs = 880 messages")
+    ap.add_argument("--channels", type=int, default=4096)
+    ap.add_argument("--serial", action="store_true",
+                    help="one request at a time (latency) instead of "
+                         "pipelined bursts (throughput)")
+    ap.add_argument("--tag", default="")
+    opts = ap.parse_args()
+
+    logging.basicConfig(level=logging.WARNING)
+    asyncio.get_event_loop().add_signal_handler(
+        signal.SIGINT, asyncio.current_task().cancel
+    )
+
+    loop = asyncio.get_running_loop()
+    reader, writer = await asyncio.open_connection(opts.host, opts.port, limit=2 ** 22)
+
+    labels = (
+        opts.inputs.split(",") if opts.inputs
+        else await discover_inputs(reader, writer)
+    )
+    gains_text = make_gains(len(labels), opts.channels)
+    messages = [
+        bytes(aiokatcp.Message.request(opts.request, opts.stream, label, *row))
+        for row, label in zip(gains_text, labels)
+    ]
+    msg_bytes = sum(len(m) for m in messages)
+
+    saved = {}
+    for label in labels:
+        saved[label] = await query_gains(reader, writer, opts.request, opts.stream, label)
+
+    ok = fail = 0
+    first_fail = None
+    latencies = []
+
+    async def drain_one():
+        nonlocal ok, fail, first_fail
+        while True:
+            line = await reader.readuntil()
+            if line[:1] == b"!":
+                if b" ok" in line:
+                    ok += 1
+                else:
+                    fail += 1
+                    if first_fail is None:
+                        first_fail = line[:200]
+                return
+
+    async def once():
+        if opts.serial:
+            for message in messages:
+                t0 = loop.time()
+                writer.write(message)
+                await drain_one()
+                latencies.append((loop.time() - t0) * 1e3)
+        else:
+            for message in messages:
+                writer.write(message)
+            for _ in messages:
+                await drain_one()
+
+    try:
+        await once()  # warmup pass, not timed
+        ok = fail = 0
+        latencies.clear()
+        start = loop.time()
+        for _ in range(opts.passes):
+            await once()
+        elapsed = loop.time() - start
+    finally:
+        for label, vals in saved.items():
+            await set_gains(reader, writer, opts.request, opts.stream, label, vals)
+
+    n = opts.passes * len(messages)
+    size = opts.passes * msg_bytes
+    mode = "serial" if opts.serial else "pipelined"
+    print("%-10s %-10s %7.2f s  %8.2f ms/msg  %7.1f msg/s  %6.1f MB/s  ok=%d fail=%d%s"
+          % (opts.tag, mode, elapsed, elapsed / n * 1e3, n / elapsed,
+             size / elapsed / 1e6, ok, fail,
+             "  first_fail=%r" % first_fail if first_fail else ""))
+    if latencies:
+        latencies.sort()
+
+        def pct(q):
+            return latencies[min(len(latencies) - 1, int(len(latencies) * q))]
+
+        print("%-10s per-request latency: p50 %.2f ms  p95 %.2f ms  max %.2f ms"
+              % ("", pct(0.50), pct(0.95), latencies[-1]))
+    writer.close()
+    await writer.wait_closed()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+        asyncio.run(main())
+    except asyncio.CancelledError:
+        pass
diff --git a/katcp/client.py b/katcp/client.py
index a34191eb..0ba34f74 100644
--- a/katcp/client.py
+++ b/katcp/client.py
@@ -11,7 +11,9 @@
 standard_library.install_aliases()  # noqa: E402
 
 import logging
+import os
 import sys
+import time
 import traceback
 
 from builtins import object
@@ -37,6 +39,22 @@
 # logging.basicConfig(level=logging.DEBUG)
 log = logging.getLogger("katcp.client")
 
+# Request names for which the outgoing encode is timed, taken from the
+# KATCP_LATENCY_LOG_REQUESTS environment variable as a comma-separated list
+# (e.g. "wide-gain"). Empty, the default, disables the instrumentation.
+#
+# This complements the proxying intervals logged by katcore's DeviceHandler.
+# Those cannot see this cost: callback_request() is @make_threadsafe, so off
+# the ioloop thread it only enqueues and returns, leaving the encode to run
+# here afterwards -- inside what that instrumentation reports as the device
+# round trip.
+LATENCY_LOG_REQUESTS = frozenset(
+    name.strip()
+    for name in os.environ.get("KATCP_LATENCY_LOG_REQUESTS", "").split(",")
+    if name.strip()
+)
+
+
 def address_to_string(addr_tuple):
     return ":".join(str(part) for part in addr_tuple)
 
@@ -425,7 +443,15 @@ def send_message(self, msg):
 
         """
         assert get_thread_ident() == self.ioloop_thread_id
-        data = bytes(msg) + b"\n"
+        if msg.name in LATENCY_LOG_REQUESTS:
+            encode_start = time.time()
+            data = bytes(msg) + b"\n"
+            self._logger.info(
+                '[LATENCY] %s encode=%.6fs bytes=%d',
+                msg.name, time.time() - encode_start, len(data),
+            )
+        else:
+            data = bytes(msg) + b"\n"
 
         # Log all sent messages here so no one else has to.
         if self._logger.isEnabledFor(logging.DEBUG):
diff --git a/katcp/core.py b/katcp/core.py
index c34423ce..629d0e98 100644
--- a/katcp/core.py
+++ b/katcp/core.py
@@ -321,6 +321,16 @@ def format_argument(self, arg):
             else:
                 return encode_utf8_with_error_log(str(arg))
 
+    @classmethod
+    def _from_parsed(cls, mtype, name, arguments, mid):
+        """Construct a Message from pre-validated parts, bypassing __init__."""
+        msg = cls.__new__(cls)
+        msg.mtype = mtype
+        msg.name = name
+        msg.mid = mid
+        msg.arguments = arguments
+        return msg
+
     def copy(self):
         """Return a shallow copy of the message object and its arguments.
 
@@ -330,7 +340,8 @@ def copy(self):
             A copy of the message object.
 
         """
-        return Message(self.mtype, self.name, self.arguments)
+        return self._from_parsed(self.mtype, self.name, list(self.arguments),
+                                 None)
 
     def __bytes__(self):
         """Return Message serialized for transmission.
@@ -342,9 +353,14 @@ def __bytes__(self):
 
         """
         if self.arguments:
-            escaped_args = [self.ESCAPE_RE.sub(self._escape_match, x)
-                            for x in self.arguments]
-            escaped_args = [x or b"\\@" for x in escaped_args]
+            search = self.ESCAPE_RE.search
+            sub = self.ESCAPE_RE.sub
+            escape_match = self._escape_match
+            escaped_args = []
+            for x in self.arguments:
+                if search(x):
+                    x = sub(escape_match, x)
+                escaped_args.append(x or b"\\@")
             arg_str = b" " + b" ".join(escaped_args)
         else:
             arg_str = b""
@@ -541,6 +557,9 @@ class MessageParser(object):
     ## @brief Regular expression matching all escapes.
     UNESCAPE_RE = re.compile(br"\\(.?)")
 
+    ## @brief Non-whitespace specials that require escaping within arguments.
+    SPECIAL_NO_WHITESPACE_RE = re.compile(br"[\0\n\r\x1b]")
+
     ## @brief Regular expression matching KATCP whitespace (just space and tab)
     WHITESPACE_RE = re.compile(br"[ \t]+")
 
@@ -602,7 +621,14 @@ def parse(self, line):
             del parts[-1]
 
         name = parts[0][1:]
-        arguments = [self._parse_arg(x) for x in parts[1:]]
+        if b"\\" not in line:
+            match = self.SPECIAL_NO_WHITESPACE_RE.search(line, len(parts[0]))
+            if match:
+                raise KatcpSyntaxError(
+                    "Un-escaped special %r." % (match.group(),))
+            arguments = parts[1:]
+        else:
+            arguments = [self._parse_arg(x) for x in parts[1:]]
 
         # split out message id
         match = self.NAME_RE.match(name)
@@ -615,7 +641,7 @@ def parse(self, line):
 
         name = ensure_native_str(name)
 
-        return Message(mtype, name, arguments, mid)
+        return Message._from_parsed(mtype, name, arguments, mid)
 
 
 class ProtocolFlags(object):
diff --git a/katcp/server.py b/katcp/server.py
index f9190e25..2ab64e34 100644
--- a/katcp/server.py
+++ b/katcp/server.py
@@ -1024,7 +1024,7 @@ def handle_message(self, client_conn, msg):
 
         """
         # log messages received so that no one else has to
-        self._logger.debug('received: {0!s}'.format(msg))
+        self._logger.debug('received: %s', msg)
 
         if msg.mtype == msg.REQUEST:
             return self.handle_request(client_conn, msg)
diff --git a/tox.ini b/tox.ini
index cc1fa57b..b5fb64f0 100644
--- a/tox.ini
+++ b/tox.ini
@@ -10,7 +10,10 @@ commands =
     coverage html
     coverage report -m --skip-covered
 deps =
-    coverage
+    # coverage >= 5 stores data via sqlite3, which is missing from the
+    # build node's source-built Python (no _sqlite3 extension); 4.x uses a
+    # flat file and is also the last major supporting the py27 env.
+    coverage<5
     mock==3.0.5
     nose
     nosexcover