diff options
Diffstat (limited to 'research/experiments/037-view-workers/views.py')
| -rwxr-xr-x | research/experiments/037-view-workers/views.py | 177 |
1 files changed, 177 insertions, 0 deletions
diff --git a/research/experiments/037-view-workers/views.py b/research/experiments/037-view-workers/views.py new file mode 100755 index 0000000..ea9aa30 --- /dev/null +++ b/research/experiments/037-view-workers/views.py @@ -0,0 +1,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() |