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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
#!/usr/bin/env python3
"""Views that compute, beside writes: 034's views.py (research 119) for
docs/design/020's stage 2 (research 122), its summary with the writes'
max latency too (the criterion), the host's further arguments from
VIEWS_SERVE_ARGS (as --view-workers 0, the control). pure-serve (THREADS
connection threads, store in RAM, cores 0-2) holds 034's graph.aup with a
random graph of NODES nodes and EDGES edges (seed 119); a second one like
it (apart) shares the cores. In turns, for SECONDS each, ROUNDS rounds:
writes benchmarks/http/load posts points (16 connections, keep-alive,
core 3): the owner applies them
both the same, while ASKERS connections ask the distance view (its
body a breadth first search) without pause, random arguments
apart the same, the askers asking the second host: the cores shared,
not the host
events the same, the askers posting the search as events (route): on
the owner
one, many the askers alone, 1 and ASKERS of them
Every view answer is checked against a model, made before any run is
timed: the askers start from 64 nodes picked beforehand (a model made while
asking would time Python, not the host).
views.py PURE_SERVE AUP NODES EDGES THREADS ASKERS SECONDS ROUNDS [WAYS] a JSON line per run, then medians
(WAYS: the ways to run, by commas; all by default)
"""
import http.client
import json
import os
import random
import shutil
import statistics
import struct
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path
HERE = Path(__file__).resolve().parent
GRAPH = HERE.parent / '034-view-bodies' / 'graph.aup'
LOAD = HERE.parents[2] / 'benchmarks' / 'http' / 'load'
EXTRA = os.environ.get('VIEWS_SERVE_ARGS', '').split()
def main():
serve, aup = Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve()
nodes, edges_wanted, threads, askers, seconds, rounds = (int(a) for a in sys.argv[3:9])
chosen = sys.argv[9].split(',') if len(sys.argv) > 9 else ['writes', 'both', 'apart', 'events', 'one', 'many']
rng = random.Random(119)
pairs = set()
while len(pairs) < edges_wanted:
a, b = rng.randrange(nodes), rng.randrange(nodes)
if a != b:
pairs.add((a, b))
edges = sorted(pairs)
out = [[] for _ in range(nodes)]
for a, b in edges:
out[a].append(b)
model = {}
def far(a, b):
if a not in model:
depth, frontier = {a: 0}, [a]
for v in frontier:
for t in out[v]:
if t not in depth:
depth[t] = depth[v] + 1
frontier.append(t)
model[a] = depth
return model[a].get(b, -1)
work = Path(tempfile.mkdtemp(prefix='pure-037-', dir='/dev/shm'))
try:
subprocess.run([str(aup), 'build', str(GRAPH), '-o', str(work / 'graph.pmod')], check=True)
host = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(work / 'graph.pmod'), '--store',
str(work / 'store'), '--port', '0', '--threads', str(threads)] + EXTRA,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
port = json.loads(host.stdout.readline())['port']
apart = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(work / 'graph.pmod'), '--store',
str(work / 'apart'), '--port', '0', '--threads', str(threads)] + EXTRA,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
apart_port = json.loads(apart.stdout.readline())['port']
for target in (port, apart_port):
setup = http.client.HTTPConnection('127.0.0.1', target, timeout=600)
def post(path, body=b''):
setup.request('POST', path, body, {'Content-Length': str(len(body))})
answer = setup.getresponse()
assert answer.status == 200 and answer.read() is not None, path
post(f'/inputs/size?nodes={nodes}')
for k in range(0, len(edges), 80000):
post('/inputs/load', b''.join(struct.pack('<dd', a * 1048576 + b, b) for a, b in edges[k:k + 80000]))
starts = [rng.randrange(nodes) for _ in range(64)]
for a in starts:
far(a, a)
def ask(target, until, seed, done, event=False):
r = random.Random(seed)
c = http.client.HTTPConnection('127.0.0.1', target, timeout=600)
while time.monotonic() < until:
a, b = r.choice(starts), r.randrange(nodes)
if event:
c.request('POST', f'/inputs/route?start={a}&goal={b}', b'', {'Content-Length': '0'})
else:
c.request('GET', f'/views/distance?start={a}&goal={b}')
answer = c.getresponse()
body = answer.read()
assert answer.status == 200 and (event or int(body) == far(a, b)), (a, b, answer.status, body)
done.append(1)
def views(count):
done, until = [], time.monotonic() + seconds
ts = [threading.Thread(target=ask, args=(port, until, k * 7919 + len(model), done)) for k in range(count)]
for t in ts:
t.start()
for t in ts:
t.join()
return len(done) / seconds
def writes(way):
done, ts = [], []
if way != 'writes':
until = time.monotonic() + seconds + 0.5
ts = [threading.Thread(target=ask, args=(apart_port if way == 'apart' else port, until, k * 104729, done,
way == 'events')) for k in range(askers)]
for t in ts:
t.start()
got = subprocess.run(['taskset', '-c', '3', str(LOAD), str(port), '16', str(seconds), 'score', '64', 'keep-alive'],
capture_output=True, text=True, check=True).stdout
for t in ts:
t.join()
result = json.loads(got)
assert result['2xx'] == result['completed'] and not result['errors'], result
return result, len(done) / seconds
rates, p99, most = {}, {}, {}
for round_ in range(1, rounds + 1):
ways = chosen
for way in (ways if round_ % 2 else ways[::-1]):
if way in ('writes', 'both', 'apart', 'events'):
result, v = writes(way)
rates.setdefault(way, []).append(result['requests_per_second'])
p99.setdefault(way, []).append(result['p99_us'])
most.setdefault(way, []).append(result['max_us'])
line = {'round': round_, 'way': way, 'writes_per_second': result['requests_per_second'],
'write_p50_us': result['p50_us'], 'write_p99_us': result['p99_us'],
'write_max_us': result['max_us'], 'views_per_second': round(v, 1)}
else:
v = views(1 if way == 'one' else askers)
rates.setdefault(way, []).append(v)
line = {'round': round_, 'way': way, 'views_per_second': round(v, 1)}
print(json.dumps(line), flush=True)
for h in (host, apart):
h.terminate()
h.communicate(timeout=60)
med = {k: statistics.median(v) for k, v in rates.items()}
tail = {k: statistics.median(v) for k, v in p99.items()}
worst = {k: statistics.median(v) for k, v in most.items()}
summary = {'nodes': nodes, 'edges': len(edges), 'threads': threads, 'askers': askers,
'medians': {k: round(v, 1) for k, v in med.items()}, 'write_p99_medians_us': tail,
'write_max_medians_us': worst, 'serve_args': EXTRA}
if 'writes' in med:
summary['writes_kept'] = {k: round(med[k] / med['writes'] * 100, 1) for k in ('both', 'apart', 'events') if k in med}
if 'both' in tail and 'apart' in tail:
summary['p99_both_over_apart'] = round(tail['both'] / tail['apart'], 2)
summary['max_both_over_apart'] = round(worst['both'] / worst['apart'], 2)
if 'one' in med and 'many' in med:
summary['scaling'] = round(med['many'] / med['one'], 2)
print(json.dumps(summary), flush=True)
finally:
shutil.rmtree(work)
if __name__ == '__main__':
main()
|
