hawdb-storage 0.5.0

An embedded Rust graph database for local data-plane workloads
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

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::*;
use std::sync::Barrier;
use std::time::Instant;

#[test]
#[ignore = "local release-mode multi-identity cached reader scaling comparison"]
fn sharded_cache_reader_benchmark() {
    const IDENTITIES: usize = 64;
    const READS_PER_THREAD: usize = 50_000;
    let config = StableIdentityMappingConfig {
        page_bytes: NonZeroUsize::new(16 * 1024).unwrap(),
        max_value_bytes: NonZeroUsize::new(512).unwrap(),
        ..StableIdentityMappingConfig::default()
    };
    let cache = Arc::new(SegmentCache::new((IDENTITIES * 16 * 1024) as u64));
    let mut readers = Vec::new();
    let mut paths = Vec::new();
    for index in 0..IDENTITIES {
        let path = test_path("sharded-cache-benchmark");
        let output =
            StableIdentityMappingWriter::publish(&path, 1, entries(&mapping(32)), config).unwrap();
        assert_eq!(output.header.page_count, 1);
        let reader = StableIdentityMappingReader::open_with_cache(
            &path,
            config,
            Arc::clone(&cache),
            StoreId(index as u128 + 1),
        )
        .unwrap();
        let (value, _) = reader
            .lookup(
                StableIdentityKey::node(17),
                StableIdentityReadLimits::default(),
            )
            .unwrap();
        assert_eq!(value, Some(Value::String("node-17".to_string())));
        readers.push(reader);
        paths.push(path);
    }
    assert_eq!(cache.snapshot().entry_count, IDENTITIES);
    for threads in [1, 2, 4, 8] {
        for distributed in [true, false] {
            let barrier = Barrier::new(threads + 1);
            let elapsed = std::thread::scope(|scope| {
                let handles: Vec<_> = (0..threads)
                    .map(|worker| {
                        let readers = &readers;
                        let barrier = &barrier;
                        scope.spawn(move || {
                            let checks = crate::cache::PAGE_INTEGRITY_CHECKS.get();
                            barrier.wait();
                            for iteration in 0..READS_PER_THREAD {
                                let index = if distributed {
                                    (iteration * 17 + worker * 7) % IDENTITIES
                                } else {
                                    0
                                };
                                let (value, report) = readers[index]
                                    .lookup(
                                        StableIdentityKey::node(17),
                                        StableIdentityReadLimits::default(),
                                    )
                                    .unwrap();
                                assert_eq!(report.storage_bytes_read, 0);
                                assert_eq!(report.cache_hits, 1);
                                assert!(value.is_some());
                                std::hint::black_box(value);
                            }
                            assert_eq!(crate::cache::PAGE_INTEGRITY_CHECKS.get(), checks);
                        })
                    })
                    .collect();
                let started = Instant::now();
                barrier.wait();
                for handle in handles {
                    handle.join().unwrap();
                }
                started.elapsed()
            });
            eprintln!("sharded-cache-reader threads={threads} distributed={distributed} reads={} elapsed_ms={:.3} reads_per_second={:.0}", threads * READS_PER_THREAD, elapsed.as_secs_f64() * 1000.0, (threads * READS_PER_THREAD) as f64 / elapsed.as_secs_f64());
        }
    }
    assert_eq!(cache.snapshot().pinned_bytes, 0);
    drop(readers);
    for path in paths {
        remove_mapping_fixture(&path, &[1]);
    }
}