armdb 0.5.4

sharded bitcask key-value storage optimized for NVMe
Documentation 
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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216

//! Benchmark gate for П-9 (review 26-06-10): does the tree-global `write_lock`
//! cap multi-threaded insert throughput on a single SkipList / ConstTree?
//!
//! Run: cargo run --release -p armdb --features bench-internals --example insert_scaling
//!
//! Phases:
//!   A. raw SkipList insert scaling (index only, no disk) — write_lock isolated
//!   B. ConstTree insert scaling (end-to-end: shard mutex + write buffer + index)
//!   C. ConstTree update scaling (SeqLock in-place path, bypasses write_lock) — control
//!
//! Each phase inserts a fixed TOTAL of new unique keys split across T threads,
//! so perfect scaling shows wall time dropping ~1/T. CPU ratio = process CPU
//! time / wall time (≈ number of effectively running threads).

use std::time::Instant;

use armdb::skiplist::node::{ConstNode, random_height};
use armdb::skiplist::{InsertResult, SkipList};
use armdb::{Config, ConstTree, DiskLoc};
use tempfile::TempDir;

const PREFILL_LIST: u64 = 8_000_000;
const PREFILL_TREE: u64 = 4_000_000;
const TOTAL_OPS: u64 = 1_000_000;
const THREADS: &[usize] = &[1, 2, 4, 8, 16];

/// splitmix64 — bijection on u64, gives unique pseudo-random keys per index.
fn key64(i: u64) -> u64 {
    let mut z = i.wrapping_add(0x9E3779B97F4A7C15);
    z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
    z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
    z ^ (z >> 31)
}

fn key_bytes(i: u64) -> [u8; 8] {
    key64(i).to_be_bytes()
}

/// Process CPU time (user + sys) in seconds, from /proc/self/stat.
/// Linux-only; returns 0.0 elsewhere — the "CPU x" column is then meaningless.
fn cpu_seconds() -> f64 {
    #[cfg(target_os = "linux")]
    {
        let stat = std::fs::read_to_string("/proc/self/stat").unwrap();
        let rest = stat.rsplit(')').next().unwrap();
        let fields: Vec<&str> = rest.split_whitespace().collect();
        let utime: f64 = fields[11].parse().unwrap();
        let stime: f64 = fields[12].parse().unwrap();
        let ticks = 100.0; // CLK_TCK on Linux
        (utime + stime) / ticks
    }
    #[cfg(not(target_os = "linux"))]
    {
        0.0
    }
}

struct PhaseResult {
    threads: usize,
    wall_s: f64,
    mops: f64,
    cpu_ratio: f64,
}

fn print_table(title: &str, results: &[PhaseResult]) {
    println!("\n=== {title} ===");
    println!(
        "{:>8} {:>10} {:>10} {:>10} {:>9}",
        "threads", "wall s", "Mops/s", "speedup", "CPU x"
    );
    let base = results[0].mops;
    for r in results {
        println!(
            "{:>8} {:>10.2} {:>10.2} {:>9.2}x {:>9.1}",
            r.threads,
            r.wall_s,
            r.mops,
            r.mops / base,
            r.cpu_ratio
        );
    }
}

fn run_phase(threads: usize, total: u64, f: impl Fn(u64, u64) + Sync) -> PhaseResult {
    let per_thread = total / threads as u64;
    let cpu0 = cpu_seconds();
    let t0 = Instant::now();
    std::thread::scope(|s| {
        for t in 0..threads {
            let f = &f;
            let start = t as u64 * per_thread;
            s.spawn(move || f(start, start + per_thread));
        }
    });
    let wall_s = t0.elapsed().as_secs_f64();
    let cpu_s = cpu_seconds() - cpu0;
    PhaseResult {
        threads,
        wall_s,
        mops: (per_thread * threads as u64) as f64 / wall_s / 1e6,
        cpu_ratio: cpu_s / wall_s,
    }
}

// ---------------------------------------------------------------------------
// Phase A: raw SkipList
// ---------------------------------------------------------------------------

type Node = ConstNode<[u8; 8], 8>;

fn phase_a() {
    println!("\n[A] raw SkipList: prefill {PREFILL_LIST} keys…");
    let list: SkipList<Node> = SkipList::new(false);
    let t0 = Instant::now();
    {
        let guard = list.collector().enter();
        for i in 0..PREFILL_LIST {
            let node = Node::alloc(
                key_bytes(i),
                [0u8; 8],
                DiskLoc::new(0, 0, 0),
                random_height(),
            );
            match list.insert(node, &guard) {
                InsertResult::Inserted => {}
                InsertResult::Exists(_) => panic!("duplicate key in prefill"),
            }
        }
    }
    println!("    prefill done in {:.1}s", t0.elapsed().as_secs_f64());

    let mut next_base = PREFILL_LIST;
    let mut results = Vec::new();
    for &t in THREADS {
        let base = next_base;
        next_base += TOTAL_OPS;
        let r = run_phase(t, TOTAL_OPS, |start, end| {
            let guard = list.collector().enter();
            for i in start..end {
                let node = Node::alloc(
                    key_bytes(base + i),
                    [0u8; 8],
                    DiskLoc::new(0, 0, 0),
                    random_height(),
                );
                match list.insert(node, &guard) {
                    InsertResult::Inserted => {}
                    InsertResult::Exists(_) => panic!("duplicate key"),
                }
            }
        });
        results.push(r);
    }
    print_table(
        "Phase A: raw SkipList insert (new keys, write_lock path)",
        &results,
    );
}

// ---------------------------------------------------------------------------
// Phases B & C: ConstTree end-to-end
// ---------------------------------------------------------------------------

fn phase_bc() {
    let dir = TempDir::new().unwrap();
    let mut config = Config::balanced().build();
    config.shard_count = 16; // shard mutexes must not be the suspect
    let tree = ConstTree::<[u8; 8], 16>::open(dir.path(), config).unwrap();

    println!("\n[B] ConstTree: prefill {PREFILL_TREE} keys…");
    let t0 = Instant::now();
    for i in 0..PREFILL_TREE {
        tree.insert(&key_bytes(i), &[0u8; 16]).unwrap();
    }
    println!("    prefill done in {:.1}s", t0.elapsed().as_secs_f64());

    let mut next_base = PREFILL_TREE;
    let mut results = Vec::new();
    for &t in THREADS {
        let base = next_base;
        next_base += TOTAL_OPS;
        let r = run_phase(t, TOTAL_OPS, |start, end| {
            for i in start..end {
                tree.insert(&key_bytes(base + i), &[0u8; 16]).unwrap();
            }
        });
        results.push(r);
    }
    print_table("Phase B: ConstTree insert (new keys, end-to-end)", &results);

    // Phase C: updates over existing keys — SeqLock in-place, no write_lock.
    let mut results = Vec::new();
    for &t in THREADS {
        let r = run_phase(t, TOTAL_OPS, |start, end| {
            for i in start..end {
                // threads touch disjoint ranges inside the prefill keyspace
                let k = key_bytes(i % PREFILL_TREE);
                tree.put(&k, &[1u8; 16]).unwrap();
            }
        });
        results.push(r);
    }
    print_table(
        "Phase C: ConstTree put over existing keys (SeqLock path) — control",
        &results,
    );
}

fn main() {
    println!(
        "insert_scaling gate: {} logical cores",
        std::thread::available_parallelism().unwrap()
    );
    phase_a();
    phase_bc();
}