diff options
| -rw-r--r-- | docs/research/120-pushing-views.md | 15 | ||||
| -rw-r--r-- | research/experiments/035-push/2026-10-04-under-load.jsonl | 7 | ||||
| -rwxr-xr-x | research/experiments/035-push/push.py | 1 | ||||
| -rwxr-xr-x | research/experiments/035-push/under_load.py | 85 | ||||
| -rw-r--r-- | serve/serve.c | 18 | ||||
| -rw-r--r-- | tests/check.py | 8 |
6 files changed, 130 insertions, 4 deletions
diff --git a/docs/research/120-pushing-views.md b/docs/research/120-pushing-views.md index 53339b8..05352bb 100644 --- a/docs/research/120-pushing-views.md +++ b/docs/research/120-pushing-views.md @@ -14,6 +14,8 @@ 100번 바뀔 때 같은 CPU로 더 신선하다(p99 11 ms 대 20 ms). - **크기.** 구독자 1만에서 호스트의 메모리는 구독자마다 0.29–0.42 KiB 늘었다(기준 1 KiB). 값 하나가 1만에게 다 닿기까지 p99는 37–45 ms다(기준 100 ms). +- **쓰기 옆의 구독자.** 초당 13만 9천 번의 쓰기 옆에서 구독자 하나가 매번 바뀌는 보기를 따라가도 쓰기는 그대로다(+0.1%). 미는 일은 + 10 ms마다 한 번이다. - **지금의 길.** TPC-C는 +0.3%, 은행 질의의 명령 수는 같다(±0.1%). 질의 하나(in_credit, 0.84 ms)의 시간이 처음 여섯 쌍에서 +4.2%, 다음 여섯 쌍에서 +2.4%, 열두 쌍 모두 +3.0%였다. 명령 수가 같으니 일이 아니라 자리의 차이다. @@ -55,6 +57,19 @@ CPU는 `/proc/PID/stat`의 utime+stime, 메모리는 클라이언트를 붙이� 구독자는 요청을 읽던 연결 기록(6,192 바이트)을 돌려주고 소켓과 구독만 남는다. 구독자 1,000의 런에서 구독자마다 1.5–4 KiB로 보인 것은 붙는 동안 한꺼번에 생긴 연결 기록이 남은 것이다(돌려준 기록은 다음 연결이 다시 쓴다). 1만이면 그 몫이 묽어진다. +**쓰기 옆의 구독자**(`under_load.py`, `under-load.jsonl`). 점수판(`scores.aup`)에 64연결이 10초 동안 점수를 보낸다. 구독자 없이와, +구독자 하나가 `total`(쓰기마다 바뀐다)을 따라가며. 번갈아 세 라운드. + +| 중앙값 | 쓰기(초당) | 쓰기 p99 | +|---|---:|---:| +| 구독자 없음 | 139,012 | 1,024 µs | +| 구독자 하나 | 139,164 (+0.1%) | 832 µs | + +호스트는 10초 동안 미는 일을 3,000번 했다(스레드 셋이 10 ms마다). 이것은 처음 구현을 다시 읽다 찾은 것을 고친 뒤에 쟀다: 스레드가 미는 +일을 미루는 동안에도 깨우라는 말을 매번 지워서, 쓰기가 많으면 소유자가 저장마다 스레드마다 eventfd에 썼다. 이제 그 말은 미는 일이 +돌 때만 지운다(미는 일 한 번에 스레드마다 많아야 한 번 쓴다). 또 앱이 조용한 동안(모듈을 바꾸는 중)에 미는 일이 돌면 스트림을 끊던 +것을, 조용하고 스냅숏도 놓은 때(멈춘 때)만 끊게 고쳤다. 검사에 앱을 구성에서 빼면 스트림이 끝나는 것을 더했다. + **지금의 길**(전 `84483fe`, 후 `a7de450` 다음의 `watch_subscriber`를 넣은 빌드). TPC-C(RAM, W=1, 64터미널)와 은행 질의를 번갈아 여섯 쌍, 은행 질의만 여섯 쌍 더. 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() diff --git a/serve/serve.c b/serve/serve.c index 4e62d3d..f4eea46 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -2476,7 +2476,7 @@ static void subscriber_event(Host *host, Subscriber *s, uint32_t events) { * group's value at least. */ static void push_group(Host *host, Group *g) { App *app = g->app; - if (!atomic_load(&app->live) || atomic_load(&app->gone)) { close_group(host, g); return; } + if (atomic_load(&app->gone)) { close_group(host, g); return; } const Tokens *tokens = atomic_load_explicit(&app->tokens, memory_order_acquire); for (Subscriber *s = g->first, *after; s; s = after) { /* a tokens file changed: who may still read */ after = s->next; @@ -2486,7 +2486,12 @@ static void push_group(Host *host, Group *g) { unsubscribe(host, s); } else s->checked = tokens; } + /* Quiet (taking no request: its module changing, or stopping) it waits + * for the next word; stopped (quiet and its snapshot let go) its streams end. */ + bool quiet = !atomic_load(&app->live); PureSnapshot *snapshot = read_published(host, app); + if (quiet && !snapshot) { close_group(host, g); return; } + if (quiet) { pure_snapshot_release(snapshot); return; } if (!snapshot) return; /* none yet: the next publication wakes this */ PureApp *reader = &host->readers[app - apps]; if (reader->module != g->route.module) { /* the app changed module: the view of that name, if it is the same kind */ @@ -3429,10 +3434,15 @@ static void *run_shard(void *context) { drive(c); } serve_bursts(host); - bool told = atomic_exchange_explicit(&host->push_due, false, memory_order_acq_rel); - if (host->groups && (told || host->push_waiting)) { /* at most one pass every --push-every-ms */ + /* At most one pass every --push-every-ms; the word stays given until + * one runs, so owners write to this thread at most once a pass. */ + bool told = atomic_load_explicit(&host->push_due, memory_order_acquire); + if (host->groups && (told || host->push_waiting)) { host->push_waiting = host->now - host->pushed_at < push_every_ms; - if (!host->push_waiting) push_pass(host); + if (!host->push_waiting) { + atomic_store_explicit(&host->push_due, false, memory_order_release); /* a word given meanwhile runs the next */ + push_pass(host); + } } uint64_t now = host->now; if (now >= next_sweep) { diff --git a/tests/check.py b/tests/check.py index 91aec12..f599fb6 100644 --- a/tests/check.py +++ b/tests/check.py @@ -4723,6 +4723,14 @@ def server_push(aup, pure, work): write(work / 'push-gone.pmod') reading() assert stream.ends(), 'a stream of a view that went still runs' + write(module) + reading() + stream = EventStream(host.port, '/p/views/total', token=token) + assert stream.next()[2] == '2' + conf.write_text(f"q {module} {work / 'push-q'}\n") # p goes: its streams end + said = reading() + assert {app['name']: app['did'] for app in said['apps']}['p'] == 'removed', said + assert stream.ends(), 'a stream of an app that went still runs' finally: stats = host.stop() assert stats[0]['subscribers'] == 0, stats[0] |