kv_storage/
sled_config.rs

1use super::Result;
2use super::sled_storage::{CleanupFun, SledStorageDB};
3use anyhow::Error;
4use convert::Bytesize;
5use serde::{Deserialize, Serialize};
6use tokio::time::sleep;
7/// Configuration for Sled storage backend
8#[derive(Debug, Clone, Serialize, Deserialize)]
9pub struct Config {
10    /// Path to database directory
11    pub path: String,
12    /// Cache capacity in bytes
13    pub cache_capacity: Bytesize,
14    /// Cleanup function for expired keys
15    #[serde(skip, default = "Config::cleanup_f_default")]
16    pub cleanup_f: CleanupFun,
17}
18
19impl Default for Config {
20    fn default() -> Self {
21        Config {
22            path: String::default(),
23            cache_capacity: Bytesize::from(1024 * 1024 * 1024),
24            cleanup_f: def_cleanup,
25        }
26    }
27}
28
29impl Config {
30    /// Converts to Sled's native configuration
31    #[inline]
32    pub fn to_sled_config(&self) -> Result<sled::Config> {
33        if self.path.trim().is_empty() {
34            return Err(Error::msg("storage dir is empty"));
35        }
36        let sled_cfg = sled::Config::default()
37            .path(self.path.trim())
38            .cache_capacity(self.cache_capacity.as_u64())
39            .flush_every_ms(Some(3000))
40            .mode(sled::Mode::LowSpace);
41        Ok(sled_cfg)
42    }
43
44    /// Returns default cleanup function
45    #[inline]
46    fn cleanup_f_default() -> CleanupFun {
47        def_cleanup
48    }
49}
50
51/// Default cleanup function that runs in background thread
52fn def_cleanup(_db: &SledStorageDB) {
53    #[cfg(feature = "ttl")]
54    {
55        let db = _db.clone();
56
57        tokio::spawn(async move {
58            let limit = 200;
59            loop {
60                sleep(std::time::Duration::from_secs(10)).await;
61                let mut total_cleanups = 0;
62                let now = std::time::Instant::now();
63                loop {
64                    let now = std::time::Instant::now();
65                    let count = db.cleanup(limit);
66                    total_cleanups += count;
67                    if count > 0 {
68                        log::debug!(
69                            "def_cleanup: {}, total cleanups: {}, active_count(): {}, cost time: {:?}",
70                            count,
71                            total_cleanups,
72                            db.active_count(),
73                            now.elapsed()
74                        );
75                    }
76                    if count < limit {
77                        break;
78                    }
79                    if db.active_count() > 50 {
80                        sleep(std::time::Duration::from_millis(500)).await;
81                    } else {
82                        sleep(std::time::Duration::from_millis(0)).await;
83                    }
84                }
85                if now.elapsed().as_secs() > 3 {
86                    log::info!(
87                        "total cleanups: {}, cost time: {:?}",
88                        total_cleanups,
89                        now.elapsed()
90                    );
91                }
92            }
93        });
94    }
95}
kv_storage/sled_config.rs

kv_storage/
sled_config.rs
