The modules an app ran are freed (research 115's next step): every app's module a Loaded with holds (each snapshot of it, each request read with it); one replaced is retired and freed on the apps thread once no app runs it, every connection thread has begun a pass since (woken for it), and nothing holds it; who keeps a module to ask later whether it is the same asks its serial (core: PureModule.serial, a reader's module_serial, pure_snapshot_module, pure_snapshot_release says it freed; serve: routes, groups, links, bindings, queries, the published snapshot), as an address may come back; a link binds under its receiver's lock, a link naming an input under binding_lock; 030's memory.py: 4,000 changes 0 bytes a change (was 19,284), 4,100 freed; stress.py: readers, view workers, push streams and a body across changes, 586 of 600 loads at a freed module's address, every answer right, ASan 1,000 changes clean; a push group that looked while its app resumed missed the new snapshot (published before the app took requests) and waited for its next save: woken again once it takes them (server_view_workers had failed once under load)

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()