1 files changed, 82 insertions, 0 deletions
diff --git a/research/experiments/028-one-sending-loop/threads.py b/research/experiments/028-one-sending-loop/threads.py
new file mode 100755
index 0000000..b890c67
--- /dev/null
+++ b/research/experiments/028-one-sending-loop/threads.py
@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+"""The threads of a host whose app sends to ten links and ten wires,
+against the same host with no output wired (docs/design/017, 2b): many
+sends one message to each of its twenty outputs; the links go to the
+host's own --link-port, the wires to its own HTTP port, each to an app of
+its own (d1..d20), all sealed (a host that names its senders takes
+nothing unsealed). The threads are counted once every receiver has its
+message. One connection thread, two workers.
+
+threads.py NAME=PURE_SERVE [NAME=PURE_SERVE ...]   a JSON line per host
+"""
+import http.client
+import json
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+from sends import AUP, HERE, free_port, started, stopped, threads
+
+
+def host(serve, work, modules, key, public, wired):
+    link_port, http_port = free_port(socket.SOCK_DGRAM), free_port(socket.SOCK_STREAM)
+    sends = [f'send=l{n}=pure://127.0.0.1:{link_port}/d{n}/take seal=l{n}={public}' for n in range(1, 11)] + \
+        [f'send=w{n}=http://127.0.0.1:{http_port}/d{n + 10}/inputs/take seal=w{n}={public}' for n in range(1, 11)] \
+        if wired else []
+    (work / 'apps.conf').write_text(f"many {modules['many']} {work / 'many'} {' '.join(sends)}\n" +
+                                    ''.join(f"d{n} {modules['dst']} {work / f'd{n}'}\n" for n in range(1, 21)))
+    args = [serve, '--apps', work / 'apps.conf', '--threads', 1, '--workers', 2, '--port', http_port, '--key', key]
+    if wired:
+        args += ['--link-port', link_port, *sum((['--sender', f'many.{o}{n}={public}'] for o in 'lw' for n in range(1, 11)), [])]
+    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, answer.getheader('X-Pure-Error'))
+        return body
+
+    try:
+        ask('POST', '/many/inputs/ping?k=7')
+        deadline = time.time() + 20
+        while wired and any(int(ask('GET', f'/d{n}/views/count')) < 1 for n in range(1, 21)):
+            assert time.time() < deadline, 'not every receiver has its message'
+            time.sleep(0.01)
+        return threads(process.pid)
+    finally:
+        client.close()
+        stopped(process)
+
+
+def main():
+    serves = {name: Path(path).resolve() for name, path in (a.split('=', 1) for a in sys.argv[1:])}
+    work = Path(tempfile.mkdtemp(prefix='pure-threads-', dir='/dev/shm'))
+    try:
+        modules = {}
+        for name in ('many', '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([list(serves.values())[-1], '--apps', work / 'one.conf', '--key', key, '--port', 0])
+        public = info['public_key']
+        stopped(process)
+        for name, serve in serves.items():
+            for wired in (False, True):
+                runs = work / f'{name}-{wired}'
+                runs.mkdir()
+                print(json.dumps({'version': name, 'outputs': 'ten links, ten wires' if wired else 'none',
+                                  'threads': host(serve, runs, modules, key, public, wired)}), flush=True)
+                shutil.rmtree(runs)
+    finally:
+        shutil.rmtree(work)
+
+
+if __name__ == '__main__':
+    main()