Say where a node's memory goes, and the budget it now keeps (104, 013)

104 measures a node's memory under a receiver that is down and under a
snapshot held for long (021's run.py: resident memory against what was
posted, 30 s a case). A held snapshot grew the node without end (6.4 to
678 MiB, fixed in d1e8f38); a slow receiver grows it by what was sent,
221 B a send, which is state until delivered, so the budget bounds state
and refuses past it.

013 is that budget's design: what it counts (state bytes, deterministic),
what an event over it does, why an answer takes its send's room, how
links and HTTP wait, what it does not count and those bounds, and the
checks. With --memory 4000000 the same load fills it in about 4 s and
the node stays at 9.6-9.8 MiB (45.9 MiB and growing without).

012 records the hole the budget found in the link (the full flag), and
103 that its clock fix fell short (385754c). CHECKPOINT gains the row.

1 files changed, 221 insertions, 0 deletions

diff --git a/research/experiments/021-memory/run.py b/research/experiments/021-memory/run.py
new file mode 100644
index 0000000..49bb5a9
--- /dev/null
+++ b/research/experiments/021-memory/run.py
@@ -0,0 +1,221 @@
+#!/usr/bin/env python3
+"""Where a Pure node's memory goes when what it keeps cannot leave (review
+004: an app's whole memory budget, not its requests' alone). pure-serve runs
+an app while clients post as fast as they may for --seconds; its resident
+memory (VmRSS) is read every 0.5 s beside what was posted, and the store's
+size at the end.
+
+  outbox      the sender's output wired nowhere: every send stays in its outbox
+  budget      the same with --memory 4000000 (docs/design/013-memory-budget.md):
+              sends past it are refused, 503 memory
+  link-down   the output linked (pure://) to a port no one listens on
+  wire-down   the output sent over HTTP to a port no one listens on
+  held        a list of items whose notes are set again and again, a surface
+              holding the snapshot of all of them (hold-surface.so) the whole time
+  unheld      the same, no surface
+
+    python3 research/experiments/021-memory/run.py [--seconds 30] [--cases outbox,budget,held] [--out FILE]
+        [--label NAME]
+
+PURE_SERVE=PATH runs another build's pure-serve (--label names it in the rows).
+"""
+import argparse
+import http.client
+import json
+import os
+import random
+import select
+import signal
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+HERE = Path(__file__).resolve().parent
+SERVE = Path(os.environ.get('PURE_SERVE', ROOT / 'build/pure-serve'))  # another build's, to compare
+sys.path.insert(0, str(ROOT / 'tests'))
+import check  # noqa: E402
+
+ITEMS = """record Item {
+  id = 0
+  note: text
+}
+state items: list of Item by id
+input add = message {
+  id: number
+  note: text
+}
+input set = message {
+  id: number
+  note: text
+}
+on add { items.add(Item { id: add.id, note: add.note }) }
+on set { items[set.id].note = set.note }
+view count = items.count
+"""
+SEEDED, NOTE, MEMORY = 20000, 100, 4000000
+
+
+def build(work):
+    if 'PURE_SERVE' not in os.environ:
+        subprocess.run(['make', '-s', '-C', str(ROOT), 'build/pure-serve'], check=True)
+    aup = Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup'))
+    for name, text in (('sender', check.LINK_SENDER), ('items', ITEMS)):
+        (work / f'{name}.aup').write_text(text)
+        check.compile_module(aup, work / f'{name}.aup', work / f'{name}.pmod')
+    subprocess.run(['cc', '-O2', '-std=c17', '-Wall', '-Wextra', '-Werror', '-fPIC', '-shared', f'-I{ROOT}/core',
+                    f'-I{ROOT}/serve', str(HERE / 'hold-surface.c'), '-o', str(work / 'hold-surface.so')], check=True)
+    key = subprocess.Popen([str(SERVE), '--module', str(work / 'items.pmod'), '--store',
+                            str(work / 'key-store'), '--key', str(work / 'node.key'), '--port', '0'],
+                           stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    select.select([key.stdout], [], [], 10)
+    key.stdout.readline()
+    key.terminate()
+    key.communicate(timeout=30)
+
+
+def start(args):
+    args = [str(a) for a in args]
+    at = args.index('--') if '--' in args else len(args)  # the surface's arguments come last
+    process = subprocess.Popen(args[:at] + ['--port', '0'] + args[at:], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    ready, _, _ = select.select([process.stdout], [], [], 10)
+    line = process.stdout.readline() if ready else b''
+    assert line, process.stderr.read(2000) if process.poll() is not None else 'no start'
+    return process, json.loads(line)['port']
+
+
+def rss_kib(pid):
+    with open(f'/proc/{pid}/status') as status:
+        for line in status:
+            if line.startswith('VmRSS:'):
+                return int(line.split()[1])
+    return 0
+
+
+def store_bytes(path):
+    return sum(f.stat().st_size for f in Path(path).rglob('*') if f.is_file())
+
+
+def post_all(port, bodies, stop, counts, index, refused):
+    """Posts (path, body) from bodies() on one keep-alive connection until
+    stop: counts[index] the ones taken, refused[index] those refused 503."""
+    connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30)
+    while not stop.is_set():
+        path, body = bodies()
+        data = json.dumps(body).encode()
+        try:
+            connection.request('POST', path, data, {'Content-Length': str(len(data)),
+                                                    'Content-Type': 'application/json'})
+            response = connection.getresponse()
+            response.read()
+            if response.status == 200:
+                counts[index] += 1
+            elif response.status == 503:
+                refused[index] += 1
+        except OSError:
+            connection.close()
+            connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30)
+    connection.close()
+
+
+def run_case(case, work, seconds, clients=4):
+    store = work / f'store-{case}'
+    args = [SERVE, '--store', store, '--threads', '2']
+    note = 'n' * NOTE
+    if case in ('outbox', 'budget', 'link-down', 'wire-down'):
+        args += ['--module', work / 'sender.pmod']
+        if case == 'budget':
+            args += ['--memory', MEMORY]
+        if case == 'link-down':
+            args += ['--key', work / 'node.key', '--send', 'ordered=pure://127.0.0.1:9/order',
+                     '--seal', f'ordered={os.urandom(32).hex()}']
+        elif case == 'wire-down':
+            args += ['--send', 'ordered=http://127.0.0.1:9/inputs/order']
+
+        def bodies():
+            return '/inputs/buy', {'item': note, 'count': 1}
+    else:
+        args += ['--module', work / 'items.pmod']
+        if case == 'held':  # the snapshot of all the items, held from then on
+            args += ['--surface', work / 'hold-surface.so', '--', 'count', str(SEEDED)]
+        rng = random.Random(7)
+
+        def bodies():
+            return '/inputs/set', {'id': rng.randrange(SEEDED), 'note': f'{rng.getrandbits(32):08x}' + note[8:]}
+    process, port = start(args)
+    try:
+        if case in ('held', 'unheld'):  # the items first, then the sets
+            seeded = [0]
+            done = threading.Event()
+            batches = [iter(range(k, SEEDED, clients)) for k in range(clients)]
+
+            def seeding(k):
+                connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30)
+                for i in batches[k]:
+                    data = json.dumps({'id': i, 'note': note}).encode()
+                    connection.request('POST', '/inputs/add', data, {'Content-Length': str(len(data)),
+                                                                      'Content-Type': 'application/json'})
+                    response = connection.getresponse()
+                    response.read()
+                    assert response.status == 200, response.status
+                connection.close()
+            threads = [threading.Thread(target=seeding, args=(k,)) for k in range(clients)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+            del seeded, done
+            time.sleep(1.0)  # the surface reads the first snapshot after the seeding's saves
+        stop = threading.Event()
+        counts, refused = [0] * clients, [0] * clients
+        began, rss_began = time.time(), rss_kib(process.pid)
+        samples = [(0.0, rss_began, 0)]
+        threads = [threading.Thread(target=post_all, args=(port, bodies, stop, counts, k, refused))
+                   for k in range(clients)]
+        for t in threads:
+            t.start()
+        while time.time() - began < seconds:
+            time.sleep(0.5)
+            samples.append((round(time.time() - began, 2), rss_kib(process.pid), sum(counts)))
+        stop.set()
+        for t in threads:
+            t.join()
+        rss_end = rss_kib(process.pid)
+    finally:
+        process.send_signal(signal.SIGTERM)
+        _, error = process.communicate(timeout=60)
+    stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')]
+    posted = sum(counts)
+    return {'case': case, 'seconds': seconds, 'posted': posted, 'refused': sum(refused), 'rss_began_kib': rss_began,
+            'rss_end_kib': rss_end,
+            'rss_per_post_bytes': round((rss_end - rss_began) * 1024 / posted, 1) if posted else None,
+            'store_bytes': store_bytes(store), 'samples': samples[::4] + [samples[-1]],
+            'stats': stats[-1] if stats else None}
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--seconds', type=int, default=30)
+    parser.add_argument('--cases', default='outbox,budget,link-down,wire-down,unheld,held')
+    parser.add_argument('--out', type=Path, default=None)
+    parser.add_argument('--label', default=None, help='the build, named in each row')
+    args = parser.parse_args()
+    with tempfile.TemporaryDirectory(prefix='pure-memory-') as directory:
+        work = Path(directory)
+        build(work)
+        for case in args.cases.split(','):
+            row = run_case(case, work, args.seconds)
+            if args.label:
+                row = {'label': args.label, **row}
+            line = json.dumps(row)
+            print(line, flush=True)
+            if args.out:
+                with open(args.out, 'a') as out:
+                    out.write(line + '\n')
+
+
+if __name__ == '__main__':
+    main()