1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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()
|
