diff options
Diffstat (limited to 'research/experiments/030-app-lifecycle/stress.py')
| -rw-r--r-- | research/experiments/030-app-lifecycle/stress.py | 170 |
1 files changed, 170 insertions, 0 deletions
diff --git a/research/experiments/030-app-lifecycle/stress.py b/research/experiments/030-app-lifecycle/stress.py new file mode 100644 index 0000000..d713f38 --- /dev/null +++ b/research/experiments/030-app-lifecycle/stress.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Modules replaced and freed while they are read (research 115's next step: the modules an app ran are let go). A +host of apps (--threads 3, --view-workers 1) holds an app of items-body.aup; its line names in turn items-body.pmod, +items-third.pmod (the same app) and items-same.pmod (a view renamed, total and tally; count's fold the same instructions at the same place, counting +n > 1 and adding 2000), so that a module freed and the next loaded take the same address, and a fold remembered for +one (a reader's pass, kept by its module and place) gives the other a value neither has; and picked, a view with a body +whose list is of a record of one field in one module and of two in the other at the same place (a reader laid out for +one and reading the other gives neither's value: 11 and 100,222); total and tally at the same place, tally 500,000 more +(a request resolved with one module and shown with the other at the same address gives the other's). Meanwhile, until CHANGES changes are done: +- readers on connections of their own GET /x/views/count (in both), /x/views/summed and picked (views with a body: the + view workers), and total and tally (one in each module): each answer 200 with the value the items give, 404 for a view + the module lacks, or 503 while it changes; +- subscribers stream count and total (Accept: text/event-stream), again each time a stream ends: every count pushed + one of the two, every total below 500,000 (a push group asked whether its app runs the module it was made in by + address would take items-same, come where its module was freed, for that module and push tally); +- a writer sends POST /x/inputs/add with its body in two parts, a change between them: 200, or 503 refused as stopped. +Then the host is stopped, and its stats say how many modules it freed and how many it loaded at an address one freed +had (the case comparing by address would get wrong). Run with an ASan build: a module read after it +was freed stops the host there. + +stress.py PURE_SERVE CHANGES one JSON line""" +import json +import select +import shutil +import signal +import socket +import subprocess +import sys +import tempfile +import threading +import time +from pathlib import Path + +from window import AUP, HERE, grow + + +def answer(sock): + head = b'' + while b'\r\n\r\n' not in head: + part = sock.recv(4096) + if not part: + raise ConnectionError('closed') + head += part + top, rest = head.split(b'\r\n\r\n', 1) + status = int(top.split()[1]) + length = next(int(line.split(b':')[1]) for line in top.split(b'\r\n') if line.lower().startswith(b'content-length')) + while len(rest) < length: + rest += sock.recv(4096) + return status, rest[:length] + + +def main(): + serve, changes = Path(sys.argv[1]).resolve(), int(sys.argv[2]) + work = Path(tempfile.mkdtemp(prefix='pure-030-', dir='/dev/shm')) + try: + modules = {} + for name in ('items-body', 'items-third', 'items-same'): + modules[name] = work / f'{name}.pmod' + subprocess.run([str(AUP), 'build', str(HERE / f'{name}.aup'), '-o', str(modules[name])], check=True) + conf = work / 'apps.conf' + cycle = ('items-body', 'items-third', 'items-same') # items-same can come where items-body was freed + write = lambda n: conf.write_text(f"x {modules[cycle[n % 3]]} {work / 'store'}\n") + write(0) + host = subprocess.Popen([str(serve), '--apps', str(conf), '--port', '0', '--threads', '3', '--view-workers', '1'], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + port = json.loads(host.stdout.readline())['port'] + items = grow(port, 1000) + counts = (items, items - 1 + 2000) # count in items-body (every n is 1 or more), in items-same (n > 1, + 2000) + done = threading.Event() + seen = {'answers': 0, 'not_found': 0, 'changing': 0, 'pushed': 0, 'writes': 0, 'writes_refused': 0, 'wrong': []} + lock = threading.Lock() + + def tally(key, wrong=None): + with lock: + seen[key] += 1 + if wrong is not None and len(seen['wrong']) < 10: + seen['wrong'].append(wrong) + + def reader(view): + sock = socket.create_connection(('127.0.0.1', port)) + while not done.is_set(): + sock.sendall(f'GET /x/views/{view} HTTP/1.1\r\nHost: x\r\n\r\n'.encode()) + status, body = answer(sock) + if status == 503: + tally('changing') + elif status == 404: + tally('not_found', None if view in ('total', 'tally') else (view, status, body[:80])) + elif status == 200: + value = json.loads(body) + good = (value in counts if view == 'count' else value >= items if view == 'summed' else + value in (11, 100222) if view == 'picked' else value < 500000 if view == 'total' else + value >= 500000) + tally('answers', None if good else (view, value)) + else: + tally('answers', (view, status, body[:80])) + + def subscriber(view, good): # streams again when its stream ends (its view gone from the module that runs) + while not done.is_set(): + sock = socket.create_connection(('127.0.0.1', port)) + sock.sendall(f'GET /x/views/{view} HTTP/1.1\r\nHost: x\r\nAccept: text/event-stream\r\n\r\n'.encode()) + sock.settimeout(0.5) + buffer = b'' + while not done.is_set(): + try: + part = sock.recv(4096) + except socket.timeout: + continue + if not part: + break + buffer += part + while b'\n\n' in buffer: + event, buffer = buffer.split(b'\n\n', 1) + for line in event.split(b'\n'): + if line.startswith(b'data:'): + value = json.loads(line[5:]) + tally('pushed', None if good(value) else ('pushed', view, value)) + sock.close() + + def writer(): + while not done.is_set(): + sock = socket.create_connection(('127.0.0.1', port)) + body = b'{"step": 1}' + sock.sendall(b'POST /x/inputs/add HTTP/1.1\r\nHost: x\r\nContent-Type: application/json\r\n' + b'Content-Length: %d\r\n\r\n' % len(body) + body[:4]) + time.sleep(0.003) # a change may come between the parts + sock.sendall(body[4:]) + status, said = answer(sock) + if status == 200: + tally('writes') + elif status == 503: + tally('writes_refused') + else: + tally('writes', ('write', status, said[:80])) + sock.close() + + threads = [threading.Thread(target=reader, args=(v,)) for v in ('count', 'picked', 'summed', 'total', 'tally', 'picked')] + threads += [threading.Thread(target=subscriber, args=('count', lambda v: v in counts)), + threading.Thread(target=subscriber, args=('total', lambda v: isinstance(v, (int, float)) and v < 500000)), + threading.Thread(target=writer)] + for t in threads: + t.start() + for n in range(1, changes + 1): + write(n) + host.send_signal(signal.SIGHUP) + ready, _, _ = select.select([host.stdout], [], [], 60) + line = host.stdout.readline() if ready else b'' + if not line: + done.set() + raise SystemExit(f'the host stopped at change {n}: {host.stderr.read().decode()[-3000:]}') + said = json.loads(line) + assert said['apps'][0]['did'] == 'changed' and said['apps'][0]['running'], said + time.sleep(0.002) + time.sleep(1) + done.set() + for t in threads: + t.join(timeout=10) + host.terminate() + _, errors = host.communicate(timeout=60) + stats = next((json.loads(l) for l in errors.decode().splitlines() if '"modules_freed"' in l), {}) + print(json.dumps({'changes': changes, **{k: v for k, v in seen.items()}, 'host_exit': host.returncode, + 'modules_freed': stats.get('modules_freed'), 'modules_retired': stats.get('modules_retired'), + 'modules_at_freed': stats.get('modules_at_freed'), + 'asan': [l for l in errors.decode().splitlines() if 'AddressSanitizer' in l][:3]})) + sys.exit(0 if not seen['wrong'] and host.returncode in (0, -15) else 1) + finally: + shutil.rmtree(work) + + +if __name__ == '__main__': + main() |