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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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()
|
