Wake each thread at most once a push, end streams only of stopped apps, and show a subscriber costs writes nothing (019)

3 files changed, 93 insertions, 0 deletions

diff --git a/research/experiments/035-push/2026-10-04-under-load.jsonl b/research/experiments/035-push/2026-10-04-under-load.jsonl
new file mode 100644
index 0000000..7ab19f6
--- /dev/null
+++ b/research/experiments/035-push/2026-10-04-under-load.jsonl
@@ -0,0 +1,7 @@
+{"round": 1, "way": "none", "writes_per_second": 138759, "p99_us": 896, "reads_by_subscribers": 0}
+{"round": 1, "way": "subscribed", "writes_per_second": 139164, "p99_us": 832, "reads_by_subscribers": 1000}
+{"round": 2, "way": "subscribed", "writes_per_second": 139052, "p99_us": 896, "reads_by_subscribers": 1000}
+{"round": 2, "way": "none", "writes_per_second": 139115, "p99_us": 1024, "reads_by_subscribers": 0}
+{"round": 3, "way": "none", "writes_per_second": 139012, "p99_us": 1216, "reads_by_subscribers": 0}
+{"round": 3, "way": "subscribed", "writes_per_second": 139320, "p99_us": 800, "reads_by_subscribers": 998}
+{"none_median": 139012, "subscribed_median": 139164, "change": 0.1, "pushes": 3000, "pushed": 3000}
diff --git a/research/experiments/035-push/push.py b/research/experiments/035-push/push.py
index 28334dd..f854bb3 100755
--- a/research/experiments/035-push/push.py
+++ b/research/experiments/035-push/push.py
@@ -40,6 +40,7 @@ def main():
     runs = sys.argv[5:]
     work = Path(tempfile.mkdtemp(prefix='pure-035-', dir='/dev/shm'))
     try:
+        subprocess.run(['cc', '-O2', '-Wall', '-Wextra', '-Werror', '-o', str(WATCH), str(WATCH) + '.c'], check=True)
         subprocess.run([str(aup), 'build', str(HERE / 'counter.aup'), '-o', str(work / 'counter.pmod')], check=True)
         for run in runs:
             clients, rate, way = run.split(':', 2)
diff --git a/research/experiments/035-push/under_load.py b/research/experiments/035-push/under_load.py
new file mode 100755
index 0000000..bcf1554
--- /dev/null
+++ b/research/experiments/035-push/under_load.py
@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+"""What a subscriber costs writes (docs/design/019, research 120). pure-serve
+(connection threads 3, store in RAM, cores 0-2) holds the scoreboard
+(benchmarks/http/scores.aup); benchmarks/http/load (core 3) posts points
+on 64 connections for SECONDS, with no subscriber, or with SUBSCRIBERS
+following its total view (which every write changes), in turns, ROUNDS
+times.
+
+under_load.py PURE_SERVE AUP SECONDS ROUNDS SUBSCRIBERS   a JSON line per run, then medians
+"""
+import json
+import shutil
+import socket
+import statistics
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+LOAD = ROOT / 'benchmarks' / 'http' / 'load'
+
+
+def main():
+    serve, aup, seconds, rounds, subscribers = Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve(), *sys.argv[3:6]
+    work = Path(tempfile.mkdtemp(prefix='pure-035-load-', dir='/dev/shm'))
+    try:
+        subprocess.run([str(aup), 'build', str(ROOT / 'benchmarks' / 'http' / 'scores.aup'), '-o', str(work / 'scores.pmod')],
+                       check=True)
+        host = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(work / 'scores.pmod'), '--store',
+                                 str(work / 'store'), '--port', '0', '--threads', '3'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        port = json.loads(host.stdout.readline())['port']
+        rates = {}
+        for round_ in range(1, int(rounds) + 1):
+            ways = ['none', 'subscribed']
+            for way in (ways if round_ % 2 else ways[::-1]):
+                socks, taken, stop = [], [0], threading.Event()
+                if way == 'subscribed':
+                    for _ in range(int(subscribers)):
+                        s = socket.create_connection(('127.0.0.1', port))
+                        s.sendall(b'GET /views/total HTTP/1.1\r\nHost: x\r\nAccept: text/event-stream\r\n\r\n')
+                        socks.append(s)
+
+                    def follow():
+                        while not stop.is_set():
+                            for s in socks:
+                                s.setblocking(False)
+                                try:
+                                    while s.recv(65536):
+                                        taken[0] += 1
+                                except BlockingIOError:
+                                    pass
+                            time.sleep(0.005)
+
+                    reader = threading.Thread(target=follow)
+                    reader.start()
+                    time.sleep(0.5)
+                out = subprocess.run(['taskset', '-c', '3', str(LOAD), str(port), '64', seconds, 'score', '64', 'keep-alive'],
+                                     capture_output=True, text=True, check=True).stdout
+                result = json.loads(out)
+                stop.set()
+                for s in socks:
+                    s.close()
+                if socks:
+                    reader.join()
+                assert result['2xx'] == result['completed'] and not result['errors'], result
+                rates.setdefault(way, []).append(result['requests_per_second'])
+                print(json.dumps({'round': round_, 'way': way, 'writes_per_second': result['requests_per_second'],
+                                  'p99_us': result['p99_us'], 'reads_by_subscribers': taken[0]}), flush=True)
+                time.sleep(0.5)
+        host.terminate()
+        _, err = host.communicate(timeout=60)
+        stats = json.loads(err.decode().strip().splitlines()[-1])
+        none, subscribed = statistics.median(rates['none']), statistics.median(rates['subscribed'])
+        print(json.dumps({'none_median': none, 'subscribed_median': subscribed, 'change': round((subscribed / none - 1) * 100, 1),
+                          'pushes': stats['pushes'], 'pushed': stats['pushed']}), flush=True)
+    finally:
+        shutil.rmtree(work)
+
+
+if __name__ == '__main__':
+    main()