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
//! Compaction workers must never claim the same input files twice.
//!
//! A level below L0 is read as a single sorted run: the iterator walks its
//! files in key order and expects each key to be greater than the last. Two
//! workers that pick the same inputs produce two output sets covering the
//! same key range, both of which land in the target level. The level is
//! then not sorted, and a scan across it either reports that iteration went
//! backwards or silently stops at the first key that does not advance.
//!
//! The invariant is checked the way a reader would meet it: scan everything
//! and require that the keys strictly increase and that every key written
//! comes back exactly once.
// Native-only. wasm-pack builds every test target for wasm32, and these use
// threads and the filesystem, neither of which exists there.
#![cfg(not(target_arch = "wasm32"))]
use regolith::{Db, Options, WriteBatch};
/// Small enough that a few megabytes reach L1 through many separate
/// compaction jobs, which is what gives the workers something to race over.
fn contended_options(workers: usize) -> Options {
Options {
// Deliberately tiny. Small files and a trigger of one turn a few
// tens of megabytes into hundreds of separate compaction jobs, and
// it is the number of independent picks, not the volume of data,
// that decides whether two workers ever choose the same input.
write_buffer_size: 128 * 1024,
block_size: 4 * 1024,
block_cache_size: 256 * 1024,
target_file_size: 64 * 1024,
level_base_bytes: 128 * 1024,
l0_compaction_trigger: 1,
level0_slowdown_writes_trigger: 0,
level0_stop_writes_trigger: 0,
max_background_compactions: workers,
max_subcompactions: workers,
..Options::default()
}
}
fn write_dense_keys(db: &Db, count: u64) {
let value = [b'v'; 256];
let mut batch = WriteBatch::new();
for i in 0..count {
batch.put(&i.to_be_bytes(), &value);
if batch.buffered_bytes() >= 128 * 1024 {
db.write(std::mem::take(&mut batch)).unwrap();
}
}
db.write(batch).unwrap();
}
/// Read the whole database back and hold the reader's own invariant to it.
fn assert_single_sorted_run(db: &Db, expected: u64) {
let mut seen: u64 = 0;
let mut last: Option<Vec<u8>> = None;
for (key, _value) in db.scan_stream(None, None).unwrap() {
if let Some(prev) = &last {
assert!(
key.as_slice() > prev.as_slice(),
"scan went backwards at entry {seen}: {:?} did not follow {:?}",
key,
prev
);
}
last = Some(key);
seen += 1;
}
// A level holding two overlapping runs also truncates the scan, because
// the iterator stops at the key that fails to advance. Counting catches
// that even when the ordering assertion above does not fire first.
assert_eq!(
seen, expected,
"scan returned {seen} keys, expected {expected}"
);
}
#[test]
fn many_workers_keep_every_level_a_single_sorted_run() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open(dir.path(), contended_options(16)).unwrap();
const KEYS: u64 = 300_000;
write_dense_keys(&db, KEYS);
db.flush().unwrap();
while db.compact_step().unwrap() {}
assert_single_sorted_run(&db, KEYS);
}
#[test]
fn a_reopened_database_reads_back_what_the_workers_wrote() {
let dir = tempfile::tempdir().unwrap();
const KEYS: u64 = 200_000;
{
let db = Db::open(dir.path(), contended_options(16)).unwrap();
write_dense_keys(&db, KEYS);
db.flush().unwrap();
while db.compact_step().unwrap() {}
db.close().unwrap();
}
// Duplicated outputs survive in the manifest, so a reopen is where a
// corrupt level structure shows up even if the writing process saw
// consistent reads from its own cache.
let db = Db::open(dir.path(), contended_options(2)).unwrap();
assert_single_sorted_run(&db, KEYS);
}