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()