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
|
#!/usr/bin/env python3
"""Research 046: what a follower may hold when the owner's whole machine dies (run.py's kills seldom leave one with
part of a cut: a cut's records travel together). The owner takes EVENTS messages with the shim keeping every op it
shipped (SHIM_JOURNAL); then the ops are done again in order into a fresh directory, and at SAMPLES points (op
boundaries drawn at random, within cuts and at their ends, and the whole stream) a copy of what is there is opened by
pure-run as the owner (and around every rename: a compaction's or an image's install). Each must open without a fault, hold the first `total` messages' entries exactly (summed), hold
no fewer than at the sample before (a prefix never loses what a shorter one held), and the whole stream every message.
prefixes.py EVENTS SAMPLES one JSON line a sample, then one of the totals"""
import json
import os
import random
import shutil
import sys
import time
from pathlib import Path
import run as harness
OP_CREATE, OP_WRITE, OP_TRUNCATE, OP_ALLOCATE, OP_RENAME, OP_UNLINK, OP_MKDIR, OP_SYNC = range(1, 9)
def ops_of(journal):
"""The ops in order: (kind, path, rest) as bytes, and how many syncs (cuts) came before each."""
out, at, cuts = [], 0, 0
while at < len(journal):
kind, n = journal[at], int.from_bytes(journal[at + 1:at + 3], 'little')
path = journal[at + 3:at + 3 + n].decode()
p = at + 3 + n
if kind == OP_CREATE: end = p + 8
elif kind == OP_WRITE: end = p + 12 + int.from_bytes(journal[p + 8:p + 12], 'little')
elif kind == OP_TRUNCATE: end = p + 8
elif kind == OP_ALLOCATE: end = p + 16
elif kind == OP_RENAME: end = p + 2 + int.from_bytes(journal[p:p + 2], 'little')
elif kind in (OP_UNLINK, OP_MKDIR): end = p + 4
elif kind == OP_SYNC: end = p
else: raise SystemExit(f'unknown op {kind} at {at}')
out.append((kind, path, journal[p:end], cuts))
cuts += kind == OP_SYNC
at = end
return out
def apply(root, op):
kind, path, rest, _ = op
target = root / path if path else root
if kind == OP_CREATE:
flags = int.from_bytes(rest[:4], 'little')
fd = os.open(target, os.O_WRONLY | flags, int.from_bytes(rest[4:8], 'little'))
os.close(fd)
elif kind == OP_WRITE:
offset, length = int.from_bytes(rest[:8], 'little'), int.from_bytes(rest[8:12], 'little')
fd = os.open(target, os.O_WRONLY)
os.pwrite(fd, rest[12:12 + length], offset)
os.close(fd)
elif kind == OP_TRUNCATE:
os.truncate(target, int.from_bytes(rest, 'little'))
elif kind == OP_ALLOCATE:
fd = os.open(target, os.O_WRONLY)
os.posix_fallocate(fd, int.from_bytes(rest[:8], 'little'), int.from_bytes(rest[8:16], 'little'))
os.close(fd)
elif kind == OP_RENAME:
n = int.from_bytes(rest[:2], 'little')
os.rename(target, root / rest[2:2 + n].decode())
elif kind == OP_UNLINK:
(os.rmdir if int.from_bytes(rest, 'little') & 0x200 else os.unlink)(target)
elif kind == OP_MKDIR:
target.mkdir(mode=int.from_bytes(rest, 'little'), exist_ok=True)
def main():
events, samples = int(sys.argv[1]), int(sys.argv[2])
keys = harness.build()
journal = harness.WORK / 'journal'
harness.ENV['SHIM_JOURNAL'] = str(journal)
r = harness.Run(keys)
r.feed(events)
r.owner.wait()
r.reader.join(timeout=10)
r.shipper.communicate(timeout=60)
r.stop_follower()
del harness.ENV['SHIM_JOURNAL']
ops = ops_of(journal.read_bytes())
near = {i + d for i, op in enumerate(ops) if op[0] == OP_RENAME for d in range(-3, 4) if 0 < i + d <= len(ops)}
picks = sorted(set(random.sample(range(1, len(ops)), min(samples, len(ops) - 1))) | near | {len(ops)})
root, copy = harness.WORK / 'prefix', harness.WORK / 'prefix-open'
shutil.rmtree(root, ignore_errors=True)
root.mkdir()
done, last_total, failures, within = 0, -1, [], 0
started = time.time()
for pick in picks:
while done < pick:
apply(root, ops[done])
done += 1
mid = ops[pick - 1][0] != OP_SYNC # the prefix ends inside a cut
within += mid
shutil.rmtree(copy, ignore_errors=True)
shutil.copytree(root, copy)
took = harness.opened(copy)
total = int(took.get('total', -1)) if took.get('opened') else -1
good = (took.get('opened') and took.get('fault') == 'none' and took.get('summed') == harness.summed(total) and
total >= last_total and (pick < len(ops) or total == events))
if not good:
failures.append({'ops': pick, 'cut': ops[pick - 1][3], 'within_cut': mid, 'took': took, 'before': last_total})
print(json.dumps({'ops': pick, 'of': len(ops), 'cut': ops[pick - 1][3], 'within_cut': mid, 'total': total,
'passed': bool(good)}), flush=True)
last_total = max(last_total, total)
renamed = {}
for op in ops:
if op[0] == OP_RENAME:
to = op[2][2:].decode()
renamed[to] = renamed.get(to, 0) + 1
print(json.dumps({'run': 'prefixes', 'events': events, 'ops': len(ops), 'journal_bytes': journal.stat().st_size,
'renamed_to': renamed, 'samples': len(picks), 'near_renames': len(near), 'within_cuts': within, 'failed': len(failures), 'failures': failures[:5],
'seconds': round(time.time() - started, 1)}), flush=True)
if __name__ == '__main__':
main()
|
