Say what views that compute cost the writes beside them (018 stage 3, research 119)

1 files changed, 168 insertions, 0 deletions

diff --git a/research/experiments/034-view-bodies/views.py b/research/experiments/034-view-bodies/views.py
new file mode 100755
index 0000000..92bdc7c
--- /dev/null
+++ b/research/experiments/034-view-bodies/views.py
@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""Views that compute, beside writes (docs/design/018 stage 3, research
+119). pure-serve (THREADS connection threads, store in RAM, cores 0-2)
+holds 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 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
+LOAD = HERE.parents[2] / 'benchmarks' / 'http' / 'load'
+
+
+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-034-', dir='/dev/shm'))
+    try:
+        subprocess.run([str(aup), 'build', str(HERE / 'graph.aup'), '-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)],
+                                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)],
+                                 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 = {}, {}
+        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'])
+                    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()}
+        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}
+        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)
+        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()