rkvs 0.4.0

A high-performance, namespace-based key-value storage system with async operations and batch processing
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

//! Contains internal helper functions for the `Namespace` module.
use super::{Namespace, Result, RkvsError};
use crate::data_table::DataTable;
use std::sync::Arc;
use tokio::sync::RwLock;

/// Jump consistent hash implementation
/// Returns a shard index in the range [0, num_shards)
pub(super) fn jump_consistent_hash(key: &[u8], num_shards: usize) -> usize {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    std::hash::Hash::hash(key, &mut hasher);
    let mut hash = std::hash::Hasher::finish(&hasher);

    let mut b: i64 = -1;
    let mut j: i64 = 0;

    while j < num_shards as i64 {
        b = j;
        hash = hash.wrapping_mul(2862933555777941757).wrapping_add(1);
        j = ((b.wrapping_add(1) as f64) * ((1i64 << 31) as f64)
            / (((hash >> 33).wrapping_add(1)) as f64)) as i64;
    }

    b as usize
}

impl Namespace {
    /// Helper to acquire a read lock with a configured timeout.
    pub(super) async fn timeout_read_lock<'a, T>(
        &self,
        lock: &'a RwLock<T>,
        context: &str,
    ) -> Result<tokio::sync::RwLockReadGuard<'a, T>> {
        tokio::time::timeout(self.config.lock_timeout(), lock.read())
            .await
            .map_err(|_| {
                RkvsError::Storage(format!(
                    "Timeout while waiting to acquire read lock for {}",
                    context
                ))
            })
    }

    /// Helper to acquire a write lock with a configured timeout.
    pub(super) async fn timeout_write_lock<'a, T>(
        &self,
        lock: &'a RwLock<T>,
        context: &str,
    ) -> Result<tokio::sync::RwLockWriteGuard<'a, T>> {
        tokio::time::timeout(self.config.lock_timeout(), lock.write())
            .await
            .map_err(|_| {
                RkvsError::Storage(format!(
                    "Timeout while waiting to acquire write lock for {}",
                    context
                ))
            })
    }

    /// Get the shard for a given key
    pub(super) async fn get_shard(&self, key: &[u8]) -> Result<(usize, Arc<RwLock<DataTable>>)> {
        let shards_guard = self.timeout_read_lock(&self.shards, "shard list").await?;
        let shard_count = shards_guard.len();
        let shard_idx = jump_consistent_hash(key, shard_count);
        let shard = Arc::clone(&shards_guard[shard_idx]);
        Ok((shard_idx, shard))
    }
}