Record the messenger between two nodes and its latency (089)

The roadmap's E2EE messenger made of the runtime's general parts, with
no messenger code in it: sealed, authenticated sessions between nodes
(086, 088), Vega on each node's display as a surface (design 010), the
app in Vega's examples. Vega's check runs two nodes through a recording
relay: a line typed on one display is drawn on the other, frames match
the vega command's, and the relay sees no text.

A line takes 0.79 ms at p50 and 1.1 ms at p99 from one node's program
to the other node's views with stores in RAM, both commits, the HTTP
delivery and the seal included (sealing adds nothing measurable); on the
Pi's SD card 3.5 ms at p50, the long tail being the card's syncs.

1 files changed, 120 insertions, 0 deletions

diff --git a/research/experiments/015-messenger-latency/latency.py b/research/experiments/015-messenger-latency/latency.py
new file mode 100644
index 0000000..5c67f88
--- /dev/null
+++ b/research/experiments/015-messenger-latency/latency.py
@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""How long a line takes from one messenger node to the other (docs/research/089-messenger.md).
+
+Two nodes of Vega's examples/messenger.aup on this machine, Alice's and Bob's,
+wired to each other directly: sealed (each with a key of its own, sealing to
+the other's, naming the other as its one sender) or plain. A program posts
+--lines lines to Alice's node one after another; for each it times the post
+(answered once Alice's node saved the line) and the moment Bob's node shows
+it (its title view counts the line; polled every 0.2 ms). Nodes on cores 0-2,
+stores on --directory. Prints one JSON line per variant and round.
+"""
+import argparse
+import http.client
+import json
+import os
+import select
+import shutil
+import signal
+import statistics
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+
+
+def start(serve, module, work, name, *extra):
+    process = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(module), '--store',
+                                str(work / f'{name}-store'), '--port', '0', '--name', name, *map(str, extra)],
+                               stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    ready, _, _ = select.select([process.stdout], [], [], 10)
+    assert ready, 'node did not start'
+    return process, json.loads(process.stdout.readline())
+
+
+def stop(process):
+    process.send_signal(signal.SIGTERM)
+    _, error = process.communicate(timeout=30)
+    assert process.returncode == 0, error[-500:]
+
+
+class Client:
+    def __init__(self, port):
+        self.connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10)
+
+    def call(self, method, path, body=None):
+        self.connection.request(method, path, body)
+        answer = self.connection.getresponse()
+        return answer.status, answer.read().decode()
+
+
+def run(serve, module, work, sealed, lines):
+    for name in ('alice', 'bob'):
+        shutil.rmtree(work / f'{name}-store', ignore_errors=True)
+    keys = {}
+    for name in ('alice', 'bob'):
+        extra = ['--key', work / f'{name}.key'] if sealed else []
+        process, started = start(serve, module, work, name, *extra)
+        keys[name] = started.get('public_key')
+        assert Client(started['port']).call('POST', '/inputs/rename', name.encode())[0] == 200
+        stop(process)
+    # Bob first, on a port of its own; then Alice, wired to it; then Bob again, wired to Alice.
+    def wiring(name, other, port):
+        wire = ['--send', f'said=http://127.0.0.1:{port}/inputs/heard']
+        if sealed:
+            wire += ['--key', work / f'{name}.key', '--seal', f'said={keys[other]}', '--sender',
+                     f'{other}.said={keys[other]}']
+        return wire
+    bob, bob_started = start(serve, module, work, 'bob', *(['--key', work / 'bob.key'] if sealed else []))
+    alice, alice_started = start(serve, module, work, 'alice', *wiring('alice', 'bob', bob_started['port']))
+    to_alice, to_bob = Client(alice_started['port']), Client(bob_started['port'])
+    saved, seen = [], []
+    try:
+        for i in range(lines):
+            began = time.perf_counter()
+            assert to_alice.call('POST', '/inputs/say', f'line {i}'.encode())[0] == 200
+            saved.append(time.perf_counter() - began)
+            want = f'bob: {i + 1} lines'
+            while to_bob.call('GET', '/views/title')[1] != want:
+                time.sleep(0.0002)
+            seen.append(time.perf_counter() - began)
+    finally:
+        stop(alice)
+        stop(bob)
+
+    def ms(values, q):
+        return round(statistics.quantiles(values, n=100)[q - 1] * 1000, 2)
+    delivered = [b - a for a, b in zip(saved, seen)]  # from Alice's answer to Bob's view
+    return {'sealed': sealed, 'lines': lines, 'saved_p50_ms': ms(saved, 50), 'saved_p99_ms': ms(saved, 99),
+            'delivered_p50_ms': ms(delivered, 50), 'delivered_p99_ms': ms(delivered, 99),
+            'shown_p50_ms': ms(seen, 50), 'shown_p90_ms': ms(seen, 90), 'shown_p99_ms': ms(seen, 99),
+            'shown_max_ms': round(max(seen) * 1000, 2)}
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--aup', type=Path, default=ROOT.parent / 'aup' / 'build' / 'aup')
+    parser.add_argument('--vega', type=Path, default=ROOT.parent / 'vega')
+    parser.add_argument('--serve', type=Path, default=ROOT / 'build' / 'pure-serve')
+    parser.add_argument('--directory', type=Path, default=Path('/var/tmp'))
+    parser.add_argument('--lines', type=int, default=300)
+    parser.add_argument('--rounds', type=int, default=3)
+    args = parser.parse_args()
+    work = Path(tempfile.mkdtemp(prefix='pure-messenger-', dir=args.directory))
+    try:
+        module = work / 'messenger.pmod'
+        subprocess.run([str(args.aup), 'build', str(args.vega / 'examples' / 'messenger.aup'), '-o', str(module)],
+                       env=dict(os.environ, AUP_PATH=str(args.vega / 'lib')), check=True)
+        for round_ in range(args.rounds):
+            for sealed in (True, False) if round_ % 2 == 0 else (False, True):
+                print(json.dumps({'round': round_, 'directory': str(args.directory),
+                                  **run(args.serve, module, work, sealed, args.lines)}), flush=True)
+    finally:
+        shutil.rmtree(work)
+
+
+if __name__ == '__main__':
+    main()