1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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()
|
