diff options
Diffstat (limited to 'research/experiments')
| -rw-r--r-- | research/experiments/037-view-workers/plain.py | 65 | ||||
| -rw-r--r-- | research/experiments/037-view-workers/runs.sh | 41 | ||||
| -rwxr-xr-x | research/experiments/037-view-workers/views.py | 177 |
3 files changed, 283 insertions, 0 deletions
diff --git a/research/experiments/037-view-workers/plain.py b/research/experiments/037-view-workers/plain.py new file mode 100644 index 0000000..96c937f --- /dev/null +++ b/research/experiments/037-view-workers/plain.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""A view of one expression read, before and after (docs/design/020's stage +2, research 122): the scoreboard (benchmarks/http/scores.aup) on pure-serve +BEFORE and AFTER in turns (connection threads 3, store in RAM, cores 0-2, +a fresh host each run), 1,000 points posted first, then benchmarks/http/load +(core 3) reads GET /views/total on 64 connections for SECONDS. ROUNDS +pairs, the first in each even round the after one. + +plain.py BEFORE AFTER AUP SECONDS ROUNDS a JSON line per run, then medians +""" +import json +import shutil +import statistics +import subprocess +import sys +import tempfile +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[2] +LOAD = ROOT / 'benchmarks' / 'http' / 'load' + + +def run(serve, module, work, seconds): + store = work / 'store' + shutil.rmtree(store, ignore_errors=True) + host = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(module), '--store', str(store), + '--port', '0', '--threads', '3'], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + try: + port = json.loads(host.stdout.readline())['port'] + filled = json.loads(subprocess.run(['taskset', '-c', '3', str(LOAD), str(port), '8', '1', 'score', '64', + 'keep-alive'], capture_output=True, text=True, check=True).stdout) + assert filled['2xx'] == filled['completed'] and filled['completed'] > 0, filled + result = json.loads(subprocess.run(['taskset', '-c', '3', str(LOAD), str(port), '64', str(seconds), 'total', '64', + 'keep-alive'], capture_output=True, text=True, check=True).stdout) + assert result['2xx'] == result['completed'] and not result['errors'], result + return result + finally: + host.terminate() + host.communicate(timeout=60) + + +def main(): + before, after, aup = (Path(a).resolve() for a in sys.argv[1:4]) + seconds, rounds = int(sys.argv[4]), int(sys.argv[5]) + work = Path(tempfile.mkdtemp(prefix='pure-037-plain-', dir='/dev/shm')) + try: + module = work / 'scores.pmod' + subprocess.run([str(aup), 'build', str(ROOT / 'benchmarks' / 'http' / 'scores.aup'), '-o', str(module)], check=True) + rates = {'before': [], 'after': []} + for round_ in range(1, rounds + 1): + order = ['before', 'after'] if round_ % 2 else ['after', 'before'] + for version in order: + result = run(before if version == 'before' else after, module, work, seconds) + rates[version].append(result['requests_per_second']) + print(json.dumps({'round': round_, 'version': version, 'reads_per_second': result['requests_per_second'], + 'p50_us': result['p50_us'], 'p99_us': result['p99_us']}), flush=True) + b, a = statistics.median(rates['before']), statistics.median(rates['after']) + print(json.dumps({'before_median': b, 'after_median': a, 'change': round((a / b - 1) * 100, 1)}), flush=True) + finally: + shutil.rmtree(work) + + +if __name__ == '__main__': + main() diff --git a/research/experiments/037-view-workers/runs.sh b/research/experiments/037-view-workers/runs.sh new file mode 100644 index 0000000..bb91997 --- /dev/null +++ b/research/experiments/037-view-workers/runs.sh @@ -0,0 +1,41 @@ +#!/bin/sh +# The runs of research 122 (docs/design/020's stage 2), from the repository's +# root, in steps, each under research/guard.py: +# runs.sh build NAME COMMIT pure-serve of COMMIT, with its pure-crypto.so, +# in ~/.cache/pure-037/NAME +# runs.sh tail ROUND views.py on the after build: writes alone, beside +# 2 askers of the BFS view, and beside 2 askers +# of another host (apart); 20,000 nodes, 100,000 +# edges, connection threads 3, 10 s, 3 rounds +# runs.sh control ROUND the same with --view-workers 0 (the views on the +# connection threads, as before) +# runs.sh scaling ROUND views.py, connection threads 1: one asker, two +# runs.sh plain plain.py: GET /views/total (64 connections, +# 10 s), before (9484700) and after, 6 pairs +set -e +here=research/experiments/037-view-workers +builds=$HOME/.cache/pure-037 +day=$(date +%F) +views() { + until awk '{ exit !($1 < 1.5) }' /proc/loadavg; do sleep 10; done + python3 research/guard.py --most 2048 -- python3 $here/views.py "$builds/after/pure-serve" ../aup/build/aup \ + 20000 100000 "$1" 2 10 3 "$2" +} +case "$1" in +build) + rm -rf "$builds/src-$2" + git worktree add -f -q --detach "$builds/src-$2" "$3" + make -s -C "$builds/src-$2" build/pure-serve build/pure-crypto.so + mkdir -p "$builds/$2" + cp "$builds/src-$2/build/pure-serve" "$builds/src-$2/build/pure-crypto.so" "$builds/$2/" + git worktree remove --force "$builds/src-$2" ;; +tail) views 3 writes,both,apart > "$here/$day-tail-$2.jsonl" ;; +control) VIEWS_SERVE_ARGS='--view-workers 0' views 3 writes,both,apart > "$here/$day-control-$2.jsonl" ;; +scaling) views 1 one,many > "$here/$day-scaling-$2.jsonl" ;; +plain) + until awk '{ exit !($1 < 1.5) }' /proc/loadavg; do sleep 10; done + python3 research/guard.py -- python3 $here/plain.py "$builds/before/pure-serve" "$builds/after/pure-serve" \ + ../aup/build/aup 10 6 > "$here/$day-plain.jsonl" ;; +*) echo "runs.sh build NAME COMMIT | tail ROUND | control ROUND | scaling ROUND | plain" >&2 + exit 2 ;; +esac 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() |