Measure sending between two apps of a host: local outlet or link (112)

Two apps of one host, src sending one message for each of 1,000 items
in a burst event, 2,000 bursts a run, five rounds with the ways in
turns: the local outlet stores 1,141,642 sends a second at 1.17 us of
the host's CPU each, a loopback pure:// link to the host's own port
578,879 at 2.93 us (1.97 times the sends, 0.40 times the CPU). 017's
second stage asked for at least the link's rate at half its CPU or
less: it goes in. Every run checks dst's sum: each send came once.

1 files changed, 117 insertions, 0 deletions

diff --git a/research/experiments/027-apps-sending/sends.py b/research/experiments/027-apps-sending/sends.py
new file mode 100644
index 0000000..454dc2b
--- /dev/null
+++ b/research/experiments/027-apps-sending/sends.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Two apps of one host (docs/design/017, stage 2), src sending to dst
+either through the local outlet (send=out=dst/take) or over a loopback
+pure:// link to the host's own --link-port (sealed with its key). src
+holds 1,000 items; each burst event walks them and sends one message
+each, so sending (not HTTP) is what the host does. A run: BURSTS bursts
+posted one after another, then the time until dst has stored every send,
+and the host's CPU over it (utime + stime). dst's sum says each send came
+once. In RAM, the host on cores 0-2.
+
+sends.py SERVE ROUNDS [BURSTS]   prints a JSON line per run
+"""
+import http.client
+import json
+import os
+import select
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+AUP = ROOT.parent / 'aup' / 'build' / 'aup'
+ITEMS = 1000
+
+
+def started(args):
+    process = subprocess.Popen(['taskset', '-c', '0-2', *map(str, args), '--port', '0'], stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE)
+    ready, _, _ = select.select([process.stdout], [], [], 10)
+    assert ready, process.stderr.read()
+    return process, json.loads(process.stdout.readline())
+
+
+def stopped(process):
+    process.terminate()
+    _, error = process.communicate(timeout=60)
+    assert process.returncode == 0, error[-500:]
+    return json.loads(error.decode().strip().splitlines()[-1])
+
+
+def cpu_seconds(pid):
+    fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()
+    return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')
+
+
+def run(way, serve, work, modules, key, public, bursts):
+    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
+        probe.bind(('127.0.0.1', 0))
+        link_port = probe.getsockname()[1]
+    to = 'dst/take' if way == 'local' else f'pure://127.0.0.1:{link_port}/dst/take seal=out={public}'
+    (work / 'apps.conf').write_text(f"src {modules['src']} {work / 'src'} send=out={to}\n"
+                                    f"dst {modules['dst']} {work / 'dst'}\n")
+    args = [serve, '--apps', work / 'apps.conf', '--threads', 1, '--workers', 2]
+    if way == 'link':
+        args += ['--key', key, '--link-port', link_port, '--sender', f'src.out={public}']
+    process, info = started(args)
+    client = http.client.HTTPConnection('127.0.0.1', info['port'], timeout=60)
+
+    def ask(method, path):
+        client.request(method, path, b'' if method == 'POST' else None)
+        answer = client.getresponse()
+        body = answer.read()
+        assert answer.status == 200, (path, answer.status)
+        return body
+
+    try:
+        for k in range(1, ITEMS + 1):
+            ask('POST', f'/src/inputs/fill?k={k}')
+        began, cpu = time.perf_counter(), cpu_seconds(process.pid)
+        for _ in range(bursts):
+            ask('POST', '/src/inputs/burst?step=1')
+        sent = bursts * ITEMS
+        while int(ask('GET', '/dst/views/count')) < sent:
+            time.sleep(0.005)
+        took, used = time.perf_counter() - began, cpu_seconds(process.pid) - cpu
+        total = int(ask('GET', '/dst/views/sum'))
+        assert int(ask('GET', '/dst/views/count')) == sent and total == bursts * (ITEMS * (ITEMS + 1) // 2 + ITEMS), total
+    finally:
+        client.close()
+        stats = stopped(process)
+    return {'way': way, 'sends': sent, 'seconds': round(took, 3), 'sends_per_second': round(sent / took),
+            'cpu_us_per_send': round(used * 1e6 / sent, 2), 'handed': stats.get('handed', 0),
+            'link_packets': stats['link_packets']}
+
+
+def main():
+    serve, rounds = Path(sys.argv[1]).resolve(), int(sys.argv[2])
+    bursts = int(sys.argv[3]) if len(sys.argv) > 3 else 60
+    work = Path(tempfile.mkdtemp(prefix='pure-sends-', dir='/dev/shm'))
+    try:
+        modules = {}
+        for name in ('src', 'dst'):
+            modules[name] = work / f'{name}.pmod'
+            subprocess.run([str(AUP), 'build', str(HERE / f'{name}.aup'), '-o', str(modules[name])], check=True)
+        key = work / 'host.key'
+        (work / 'one.conf').write_text(f"x {modules['dst']} {work / 'keys'}\n")
+        process, info = started([serve, '--apps', work / 'one.conf', '--key', key])
+        public = info['public_key']
+        stopped(process)
+        for round_ in range(1, rounds + 1):
+            for way in (['local', 'link'] if round_ % 2 else ['link', 'local']):
+                runs = work / f'run-{round_}-{way}'
+                runs.mkdir()
+                row = run(way, serve, runs, modules, key, public, bursts)
+                print(json.dumps({'round': round_, **row}), flush=True)
+                shutil.rmtree(runs)
+    finally:
+        shutil.rmtree(work)
+
+
+if __name__ == '__main__':
+    main()