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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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()
|
