pagesift_store/
engine.rs

1//! 段文件管理、读写路径、压缩和淘汰。
2
3use std::collections::{BTreeMap, HashMap};
4use std::fs::{self, File, OpenOptions};
5use std::path::{Path, PathBuf};
6use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
7use std::sync::{Arc, Mutex, MutexGuard, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard};
8use std::time::{Instant, SystemTime};
9
10use crate::index::{Index, Loc, Slot};
11use crate::record::{
12    self, MAX_KEY, MAX_RECORD, Op, REC_HEADER, SEG_HEADER, SEG_MAGIC, read_exact_at, write_all_at,
13};
14use crate::{
15    Cache, Config, Error, GcReport, Items, OpenReport, Result, Snapshots, Stats, SyncPolicy, Tasks,
16    now_secs, to_ms,
17};
18
19/// 命名空间编号，写进每条记录，不能改。
20pub(crate) mod ns {
21    pub const CACHE: u8 = 1;
22    pub const SNAP_BLOB: u8 = 2;
23    pub const SNAP_ITEM: u8 = 3;
24    pub const KV: u8 = 4;
25    pub const LIST_ITEM: u8 = 5;
26    pub const SET_MEMBER: u8 = 6;
27    pub const ITEM: u8 = 7;
28}
29
30fn evictable(n: u8) -> bool {
31    matches!(n, ns::CACHE | ns::SNAP_BLOB)
32}
33
34/// 分组键：`[name_len u16][name][rest]`，列表的 rest 是大端序号，集合的 rest 是成员。
35pub(crate) fn group_key(name: &[u8], rest: &[u8]) -> Result<Vec<u8>> {
36    if name.len() + rest.len() + 2 > MAX_KEY {
37        return Err(Error::KeyTooLarge(name.len() + rest.len() + 2));
38    }
39    let mut k = Vec::with_capacity(2 + name.len() + rest.len());
40    k.extend_from_slice(&(name.len() as u16).to_le_bytes());
41    k.extend_from_slice(name);
42    k.extend_from_slice(rest);
43    Ok(k)
44}
45
46pub(crate) fn split_group(key: &[u8]) -> Option<(&[u8], &[u8])> {
47    let n = u16::from_le_bytes(key.get(..2)?.try_into().ok()?) as usize;
48    Some((key.get(2..2 + n)?, &key[2 + n..]))
49}
50
51/// 列表和集合的内存统计，由索引变化推出来，重启后重建。
52#[derive(Debug, Clone, Copy)]
53pub(crate) struct Group {
54    pub live: u64,
55    /// 列表最小的可能存活序号，弹出时从这里往后探测。
56    pub head: u64,
57    pub next: u64,
58}
59
60struct Seg {
61    file: Arc<File>,
62    total: u64,
63    live: u64,
64    tomb: u64,
65}
66
67impl Seg {
68    fn new(file: Arc<File>) -> Self {
69        Self {
70            file,
71            total: SEG_HEADER,
72            live: 0,
73            tomb: 0,
74        }
75    }
76    fn garbage(&self) -> u64 {
77        self.total - SEG_HEADER - self.live
78    }
79    fn dead_puts(&self) -> u64 {
80        self.garbage() - self.tomb
81    }
82}
83
84struct State {
85    index: Index,
86    segs: BTreeMap<u32, Seg>,
87    groups: HashMap<(u8, Box<[u8]>), Group>,
88    disk: u64,
89    live: u64,
90}
91
92impl State {
93    fn read_raw(&self, loc: &Loc, len: usize) -> Result<Vec<u8>> {
94        let seg = self
95            .segs
96            .get(&loc.seg)
97            .ok_or_else(|| Error::Corrupt(format!("segment {} missing", loc.seg)))?;
98        let mut buf = vec![0; len];
99        read_exact_at(&seg.file, &mut buf, u64::from(loc.off))?;
100        Ok(buf)
101    }
102
103    /// 找到键的位置（已经核对过键）。
104    fn find(&self, ns: u8, key: &[u8]) -> Result<Option<(Slot, &Loc)>> {
105        match self.index.candidate(ns, key) {
106            None => Ok(None),
107            Some((slot, loc, true)) => Ok(Some((slot, loc))),
108            Some((slot, loc, false)) => {
109                let n = record::key_probe_len(key).min(loc.len() as usize);
110                let prefix = self.read_raw(loc, n)?;
111                Ok(record::key_matches(&prefix, ns, key).then_some((slot, loc)))
112            }
113        }
114    }
115
116    /// 读出完整记录，返回 (位置, 写入时间, 值)。
117    fn get_full(&self, ns: u8, key: &[u8]) -> Result<Option<(&Loc, u64, Vec<u8>)>> {
118        let Some((_, loc, _)) = self.index.candidate(ns, key) else {
119            return Ok(None);
120        };
121        let mut raw = self.read_raw(loc, loc.len() as usize)?;
122        let rec = record::decode(&raw).ok_or_else(|| {
123            Error::Corrupt(format!("checksum mismatch at {}:{}", loc.seg, loc.off))
124        })?;
125        if rec.ns != ns || rec.key != key {
126            return Ok(None);
127        }
128        let ts = rec.ts_ms;
129        let start = raw.len() - rec.value.len();
130        raw.drain(..start);
131        Ok(Some((loc, ts, raw)))
132    }
133
134    fn read_key(&self, loc: &Loc) -> Result<Vec<u8>> {
135        let head = self.read_raw(loc, REC_HEADER + 12)?;
136        let klen = u16::from_le_bytes([head[REC_HEADER + 10], head[REC_HEADER + 11]]) as usize;
137        let mut key = vec![0; klen];
138        let seg = &self.segs[&loc.seg];
139        read_exact_at(
140            &seg.file,
141            &mut key,
142            u64::from(loc.off) + (REC_HEADER + 12) as u64,
143        )?;
144        Ok(key)
145    }
146
147    fn account_append(&mut self, seg: u32, len: u64, tomb: bool) {
148        let s = self.segs.get_mut(&seg).expect("active segment registered");
149        s.total += len;
150        if tomb {
151            s.tomb += len;
152        }
153        self.disk += len;
154    }
155
156    /// 让键指向新位置（`Some`）或删除（`None`），返回之前是否存在。
157    fn apply(&mut self, ns: u8, key: &[u8], new: Option<Loc>) -> Result<bool> {
158        let found = self
159            .find(ns, key)?
160            .map(|(slot, loc)| (slot, loc.seg, u64::from(loc.len())));
161        let existed = found.is_some();
162        if let Some((_, seg, len)) = &found {
163            if let Some(s) = self.segs.get_mut(seg) {
164                s.live -= len;
165            }
166            self.live -= len;
167        }
168        let now = new.is_some();
169        if let Some(loc) = &new {
170            let len = u64::from(loc.len());
171            self.segs.get_mut(&loc.seg).expect("segment").live += len;
172            self.live += len;
173        }
174        match (found, new) {
175            (Some((slot, ..)), Some(loc)) => {
176                *self.index.get_mut(&slot, ns, key).expect("slot") = loc;
177            }
178            (None, Some(loc)) => self.index.insert_new(ns, key, loc),
179            (Some((slot, ..)), None) => {
180                self.index.remove(&slot, ns, key);
181            }
182            (None, None) => {}
183        }
184        if existed != now {
185            self.update_group(ns, key, now);
186        }
187        Ok(existed)
188    }
189
190    fn update_group(&mut self, ns: u8, key: &[u8], added: bool) {
191        let is_list = match ns {
192            ns::LIST_ITEM | ns::SNAP_ITEM => true,
193            ns::SET_MEMBER => false,
194            _ => return,
195        };
196        let Some((name, rest)) = split_group(key) else {
197            return;
198        };
199        let seq = match (is_list, <[u8; 8]>::try_from(rest)) {
200            (true, Ok(b)) => u64::from_be_bytes(b),
201            _ => 0,
202        };
203        let gk = (ns, Box::<[u8]>::from(name));
204        if added {
205            let g = self.groups.entry(gk).or_insert(Group {
206                live: 0,
207                head: seq,
208                next: seq,
209            });
210            g.live += 1;
211            g.head = g.head.min(seq);
212            g.next = g.next.max(seq + 1);
213        } else if let Some(g) = self.groups.get_mut(&gk) {
214            g.live -= 1;
215            if g.live == 0 {
216                self.groups.remove(&gk);
217            } else if seq == g.head {
218                g.head = seq + 1;
219            }
220        }
221    }
222
223    /// 压缩时把存活记录挪到新位置，保留访问时间。
224    fn relocate(&mut self, ns: u8, key: &[u8], seg: u32, off: u32) {
225        let (slot, _, _) = self.index.candidate(ns, key).expect("live record");
226        let loc = self.index.get_mut(&slot, ns, key).expect("slot");
227        let (old, len) = (loc.seg, u64::from(loc.len()));
228        loc.seg = seg;
229        loc.off = off;
230        if let Some(s) = self.segs.get_mut(&old) {
231            s.live -= len;
232        }
233        self.segs.get_mut(&seg).expect("segment").live += len;
234    }
235
236    fn is_at(&self, ns: u8, key: &[u8], seg: u32, off: u32) -> bool {
237        self.index
238            .candidate(ns, key)
239            .is_some_and(|(_, loc, _)| loc.seg == seg && loc.off == off)
240    }
241}
242
243struct Writer {
244    active: u32,
245    file: Arc<File>,
246    len: u64,
247    last_sync: Instant,
248}
249
250pub(crate) struct Inner {
251    cfg: Config,
252    writer: Mutex<Writer>,
253    state: RwLock<State>,
254    /// 同一时间只跑一个压缩或 gc。
255    maint: Mutex<()>,
256    busy: AtomicBool,
257    last_maint_ms: AtomicU64,
258    last_error: Mutex<Option<String>>,
259    open: OpenReport,
260    /// 持有期间其他进程打不开同一个目录；随存储一起释放。
261    _lock: File,
262}
263
264impl Drop for Inner {
265    fn drop(&mut self) {
266        if self.cfg.sync != SyncPolicy::Never {
267            let w = self
268                .writer
269                .get_mut()
270                .unwrap_or_else(PoisonError::into_inner);
271            let _ = w.file.sync_data();
272        }
273    }
274}
275
276/// 存储句柄，克隆很便宜（内部是 `Arc`）。
277#[derive(Clone)]
278pub struct Store(Arc<Inner>);
279
280fn seg_path(dir: &Path, id: u32) -> PathBuf {
281    dir.join(format!("{id:08}.seg"))
282}
283
284fn sync_dir(dir: &Path) -> std::io::Result<()> {
285    #[cfg(unix)]
286    {
287        File::open(dir)?.sync_all()
288    }
289    #[cfg(not(unix))]
290    {
291        let _ = dir;
292        Ok(())
293    }
294}
295
296fn create_segment(dir: &Path, id: u32) -> Result<Arc<File>> {
297    let f = OpenOptions::new()
298        .read(true)
299        .write(true)
300        .create_new(true)
301        .open(seg_path(dir, id))?;
302    write_all_at(&f, SEG_MAGIC, 0)?;
303    f.sync_all()?;
304    sync_dir(dir)?;
305    Ok(Arc::new(f))
306}
307
308impl Store {
309    pub fn open(cfg: Config) -> Result<Store> {
310        if !(SEG_HEADER + 64..=2 << 30).contains(&cfg.max_segment_bytes) {
311            return Err(Error::Corrupt(
312                "max_segment_bytes must be within 72B..=2GiB".into(),
313            ));
314        }
315        let start = Instant::now();
316        fs::create_dir_all(&cfg.dir)?;
317        // 段文件没有跨进程协调：两个进程同时追加会互相覆盖。用目录里的锁文件保证同一时间只有一个
318        // 进程打开这个存储；拿不到锁返回 `WouldBlock`，调用方可以决定不用缓存继续运行。
319        let lock = OpenOptions::new()
320            .read(true)
321            .write(true)
322            .create(true)
323            .truncate(false)
324            .open(cfg.dir.join("LOCK"))?;
325        match lock.try_lock() {
326            Ok(()) => {}
327            Err(std::fs::TryLockError::WouldBlock) => {
328                return Err(Error::Io(std::io::Error::new(
329                    std::io::ErrorKind::WouldBlock,
330                    "the directory is in use by another process",
331                )));
332            }
333            Err(std::fs::TryLockError::Error(e)) => return Err(Error::Io(e)),
334        }
335        let mut ids: Vec<u32> = fs::read_dir(&cfg.dir)?
336            .filter_map(|e| {
337                let name = e.ok()?.file_name();
338                name.to_str()?.strip_suffix(".seg")?.parse().ok()
339            })
340            .collect();
341        ids.sort_unstable();
342
343        let mut st = State {
344            index: Index::new(cfg.hash_mask),
345            segs: BTreeMap::new(),
346            groups: HashMap::new(),
347            disk: 0,
348            live: 0,
349        };
350        let mut report = OpenReport {
351            segments: ids.len(),
352            ..OpenReport::default()
353        };
354        let mut last_len = 0;
355        for &id in &ids {
356            let path = seg_path(&cfg.dir, id);
357            let file = OpenOptions::new().read(true).write(true).open(&path)?;
358            let len = file.metadata()?.len();
359            if len < SEG_HEADER {
360                // 新段的魔数没写完就崩溃了。
361                file.set_len(0)?;
362                write_all_at(&file, SEG_MAGIC, 0)?;
363                file.sync_all()?;
364                report.truncated_bytes += len;
365            } else {
366                let mut magic = [0; 8];
367                read_exact_at(&file, &mut magic, 0)?;
368                if &magic != SEG_MAGIC {
369                    return Err(Error::Corrupt(format!("{}: bad magic", path.display())));
370                }
371            }
372            let len = len.max(SEG_HEADER);
373            let file = Arc::new(file);
374            st.segs.insert(id, Seg::new(file.clone()));
375            st.disk += SEG_HEADER;
376            let end = record::scan(&file, len, |off, raw, rec| -> Result<()> {
377                if rec.ns >= 16 {
378                    return Err(Error::Corrupt(format!("unknown namespace {}", rec.ns)));
379                }
380                let rlen = raw.len() as u64;
381                st.account_append(id, rlen, rec.op == Op::Del);
382                let new = (rec.op == Op::Put)
383                    .then(|| Loc::new(id, off, rlen as u32, rec.ns, (rec.ts_ms / 1000) as u32));
384                st.apply(rec.ns, rec.key, new)?;
385                report.records += 1;
386                Ok(())
387            })?;
388            if end < len {
389                file.set_len(end)?;
390                file.sync_all()?;
391                report.truncated_bytes += len - end;
392            }
393            last_len = end;
394        }
395
396        let (active, file, len) = match ids.last() {
397            Some(&id) if last_len < cfg.max_segment_bytes => {
398                (id, st.segs[&id].file.clone(), last_len)
399            }
400            last => {
401                let id = last.map_or(1, |id| id + 1);
402                let file = create_segment(&cfg.dir, id)?;
403                st.segs.insert(id, Seg::new(file.clone()));
404                st.disk += SEG_HEADER;
405                (id, file, SEG_HEADER)
406            }
407        };
408        report.elapsed = start.elapsed();
409        Ok(Store(Arc::new(Inner {
410            cfg,
411            writer: Mutex::new(Writer {
412                active,
413                file,
414                len,
415                last_sync: Instant::now(),
416            }),
417            state: RwLock::new(st),
418            maint: Mutex::new(()),
419            busy: AtomicBool::new(false),
420            last_maint_ms: AtomicU64::new(0),
421            last_error: Mutex::new(None),
422            open: report,
423            _lock: lock,
424        })))
425    }
426
427    /// 用默认配置打开目录。
428    pub fn open_dir(dir: impl Into<PathBuf>) -> Result<Store> {
429        Self::open(Config::new(dir))
430    }
431
432    pub fn config(&self) -> &Config {
433        &self.0.cfg
434    }
435
436    pub fn cache(&self) -> Cache<'_> {
437        Cache::new(self)
438    }
439
440    pub fn snapshots(&self) -> Snapshots<'_> {
441        Snapshots::new(self)
442    }
443
444    pub fn items(&self) -> Items<'_> {
445        Items::new(self)
446    }
447
448    pub fn tasks(&self) -> Tasks<'_> {
449        Tasks::new(self)
450    }
451
452    pub fn stats(&self) -> Stats {
453        let st = self.state_r();
454        let groups = st.groups.capacity() as u64
455            * (std::mem::size_of::<((u8, Box<[u8]>), Group)>() as u64 + 1)
456            + st.groups.keys().map(|k| k.1.len() as u64).sum::<u64>();
457        Stats {
458            entries: st.index.len() as u64,
459            index_bytes: st.index.heap_bytes() + groups,
460            disk_bytes: st.disk,
461            live_bytes: st.live,
462            segments: st.segs.len(),
463            open: self.0.open.clone(),
464            last_error: lock(&self.0.last_error).clone(),
465        }
466    }
467
468    /// 把活动段 fsync 到盘上。
469    pub fn flush(&self) -> Result<()> {
470        let mut w = self.writer();
471        w.file.sync_data()?;
472        w.last_sync = Instant::now();
473        Ok(())
474    }
475
476    /// 重写所有含无效数据的段（包括当前活动段），返回重写的段数。
477    pub fn compact(&self) -> Result<usize> {
478        self.compact_with(0.0, true)
479    }
480
481    /// 供 `pagesift gc`：超过 [`Config::max_bytes`] 时按 LRU 淘汰缓存和快照内容，然后压缩。
482    pub fn gc(&self) -> Result<GcReport> {
483        let mut rep = GcReport {
484            disk_bytes_before: self.state_r().disk,
485            ..GcReport::default()
486        };
487        if let Some(max) = self.0.cfg.max_bytes {
488            let target = (max as f64 * self.0.cfg.low_watermark) as u64;
489            let mut cands: Vec<(u32, u32, u32, u32)> = {
490                let st = self.state_r();
491                if st.live <= target {
492                    Vec::new()
493                } else {
494                    st.index
495                        .iter()
496                        .filter(|l| evictable(l.ns()))
497                        .map(|l| (l.atime(), l.seg, l.off, l.len()))
498                        .collect()
499                }
500            };
501            cands.sort_unstable();
502            let _m = lock(&self.0.maint);
503            for (_, seg, off, len) in cands {
504                if self.state_r().live <= target {
505                    break;
506                }
507                let mut tx = self.tx();
508                let victim = {
509                    let st = self.state_r();
510                    if !st.segs.contains_key(&seg) {
511                        continue;
512                    }
513                    let probe = Loc::new(seg, off, len, 0, 0);
514                    let key = st.read_key(&probe)?;
515                    let n = st.read_raw(&probe, REC_HEADER + 1)?[REC_HEADER];
516                    st.is_at(n, &key, seg, off).then_some((n, key))
517                };
518                if let Some((n, key)) = victim
519                    && tx.delete(n, &key)?
520                {
521                    rep.evicted_entries += 1;
522                    rep.evicted_bytes += u64::from(len);
523                }
524            }
525        }
526        rep.compacted_segments = self.compact()?;
527        rep.disk_bytes_after = self.state_r().disk;
528        Ok(rep)
529    }
530
531    /// 在 tokio 的阻塞线程池里调用存储，例如
532    /// `store.blocking(move |s| s.cache().get(&key, max_age)).await`。
533    #[cfg(feature = "tokio")]
534    pub async fn blocking<F, R>(&self, f: F) -> R
535    where
536        F: FnOnce(&Store) -> R + Send + 'static,
537        R: Send + 'static,
538    {
539        let s = self.clone();
540        match tokio::task::spawn_blocking(move || f(&s)).await {
541            Ok(r) => r,
542            Err(e) => match e.try_into_panic() {
543                Ok(p) => std::panic::resume_unwind(p),
544                Err(e) => panic!("store task cancelled: {e}"),
545            },
546        }
547    }
548
549    // ---- 内部接口 ----
550
551    fn state_r(&self) -> RwLockReadGuard<'_, State> {
552        self.0.state.read().unwrap_or_else(PoisonError::into_inner)
553    }
554
555    fn state_w(&self) -> RwLockWriteGuard<'_, State> {
556        self.0.state.write().unwrap_or_else(PoisonError::into_inner)
557    }
558
559    fn writer(&self) -> MutexGuard<'_, Writer> {
560        lock(&self.0.writer)
561    }
562
563    /// 拿到写锁，期间的读写对其他写者是原子的。
564    pub(crate) fn tx(&self) -> Tx<'_> {
565        Tx {
566            store: self,
567            w: self.writer(),
568        }
569    }
570
571    pub(crate) fn get(&self, ns: u8, key: &[u8]) -> Result<Option<(u64, Vec<u8>)>> {
572        let st = self.state_r();
573        let Some((loc, ts, value)) = st.get_full(ns, key)? else {
574            return Ok(None);
575        };
576        if evictable(ns) {
577            loc.touch(now_secs());
578        }
579        Ok(Some((ts, value)))
580    }
581
582    pub(crate) fn contains(&self, ns: u8, key: &[u8]) -> Result<bool> {
583        Ok(self.state_r().find(ns, key)?.is_some())
584    }
585
586    pub(crate) fn put(&self, ns: u8, key: &[u8], value: &[u8], ts_ms: u64) -> Result<bool> {
587        let r = self.tx().put(ns, key, value, ts_ms);
588        self.maybe_maintain();
589        r
590    }
591
592    pub(crate) fn delete(&self, ns: u8, key: &[u8]) -> Result<bool> {
593        let r = self.tx().delete(ns, key);
594        self.maybe_maintain();
595        r
596    }
597
598    /// 某个命名空间里满足条件的所有键（要逐条读盘，只用于低频操作）。
599    pub(crate) fn keys(&self, ns: u8, mut keep: impl FnMut(&[u8]) -> bool) -> Result<Vec<Vec<u8>>> {
600        let st = self.state_r();
601        let mut out = Vec::new();
602        for loc in st.index.iter().filter(|l| l.ns() == ns) {
603            let key = st.read_key(loc)?;
604            if keep(&key) {
605                out.push(key);
606            }
607        }
608        out.sort_unstable();
609        Ok(out)
610    }
611
612    pub(crate) fn group(&self, ns: u8, name: &[u8]) -> Option<Group> {
613        self.state_r().groups.get(&(ns, Box::from(name))).copied()
614    }
615
616    fn append(&self, w: &mut Writer, raw: &[u8], tomb: bool) -> Result<(u32, u32)> {
617        let len = raw.len() as u64;
618        if w.len > SEG_HEADER && w.len + len > self.0.cfg.max_segment_bytes {
619            self.roll(w)?;
620        }
621        if let Err(e) = write_all_at(&w.file, raw, w.len) {
622            let _ = w.file.set_len(w.len);
623            return Err(e.into());
624        }
625        let off = w.len as u32;
626        w.len += len;
627        match self.0.cfg.sync {
628            SyncPolicy::Always => w.file.sync_data()?,
629            SyncPolicy::Interval(d) if w.last_sync.elapsed() >= d => {
630                w.file.sync_data()?;
631                w.last_sync = Instant::now();
632            }
633            _ => {}
634        }
635        self.state_w().account_append(w.active, len, tomb);
636        Ok((w.active, off))
637    }
638
639    /// 换一个新的活动段。旧段总是先 fsync，压缩依赖这一点。
640    fn roll(&self, w: &mut Writer) -> Result<()> {
641        w.file.sync_data()?;
642        let id = w.active + 1;
643        let file = create_segment(&self.0.cfg.dir, id)?;
644        {
645            let mut st = self.state_w();
646            st.segs.insert(id, Seg::new(file.clone()));
647            st.disk += SEG_HEADER;
648        }
649        w.active = id;
650        w.file = file;
651        w.len = SEG_HEADER;
652        w.last_sync = Instant::now();
653        Ok(())
654    }
655
656    fn needs_compact(&self, st: &State) -> bool {
657        let c = &self.0.cfg;
658        st.segs.len() > 1
659            && st.disk >= c.compact_min_bytes
660            && (st.disk - st.live) as f64 >= c.compact_ratio * st.disk as f64
661    }
662
663    fn maybe_maintain(&self) {
664        let inner = &self.0;
665        if !inner.cfg.background {
666            return;
667        }
668        let (over, compact) = {
669            let st = self.state_r();
670            let over = inner.cfg.max_bytes.is_some_and(|m| st.disk > m);
671            (over, self.needs_compact(&st))
672        };
673        if !(over || compact) {
674            return;
675        }
676        let now = to_ms(SystemTime::now());
677        if now.saturating_sub(inner.last_maint_ms.load(Ordering::Relaxed)) < 1000
678            || inner.busy.swap(true, Ordering::AcqRel)
679        {
680            return;
681        }
682        let s = self.clone();
683        let spawned = std::thread::Builder::new()
684            .name("pagesift-store-gc".into())
685            .spawn(move || {
686                let r = if over {
687                    s.gc().map(drop)
688                } else {
689                    s.compact_with(s.0.cfg.compact_ratio, false).map(drop)
690                };
691                if let Err(e) = r {
692                    *lock(&s.0.last_error) = Some(e.to_string());
693                }
694                s.0.last_maint_ms
695                    .store(to_ms(SystemTime::now()), Ordering::Relaxed);
696                s.0.busy.store(false, Ordering::Release);
697            });
698        if spawned.is_err() {
699            inner.busy.store(false, Ordering::Release);
700        }
701    }
702
703    fn compact_with(&self, min_ratio: f64, roll_active: bool) -> Result<usize> {
704        let _m = lock(&self.0.maint);
705        let targets: Vec<u32> = {
706            let mut w = self.writer();
707            if roll_active {
708                let garbage = self.state_r().segs[&w.active].garbage();
709                if garbage > 0 {
710                    self.roll(&mut w)?;
711                }
712            }
713            let st = self.state_r();
714            st.segs
715                .iter()
716                .filter(|(id, s)| {
717                    **id != w.active
718                        && s.garbage() > 0
719                        && s.garbage() as f64 >= min_ratio * s.total as f64
720                })
721                .map(|(id, _)| *id)
722                .collect()
723        };
724        drop(_m);
725        for &id in &targets {
726            self.compact_segment(id)?;
727        }
728        Ok(targets.len())
729    }
730
731    /// 测试用：除活动段外的段编号。
732    #[cfg(test)]
733    pub(crate) fn sealed_segments(&self) -> Vec<u32> {
734        let active = self.writer().active;
735        self.state_r()
736            .segs
737            .keys()
738            .copied()
739            .filter(|id| *id != active)
740            .collect()
741    }
742
743    /// 把一个段里存活的记录复制到活动段，然后删掉这个段。
744    ///
745    /// 删除标记只有在所有更旧的段都没有失效 put 时才能丢掉，否则重启重放时旧值会复活；
746    /// 这时如果键仍然是删除状态，就把标记一并复制过去。
747    pub(crate) fn compact_segment(&self, id: u32) -> Result<()> {
748        let _m = lock(&self.0.maint);
749        let (file, len, drop_tombs) = {
750            let st = self.state_r();
751            let Some(s) = st.segs.get(&id) else {
752                return Ok(());
753            };
754            let drop_tombs = st.segs.range(..id).all(|(_, s)| s.dead_puts() == 0);
755            (s.file.clone(), s.total, drop_tombs)
756        };
757        record::scan(&file, len, |off, raw, rec| -> Result<()> {
758            let mut w = self.writer();
759            match rec.op {
760                Op::Put => {
761                    if self.state_r().is_at(rec.ns, rec.key, id, off) {
762                        let (seg, noff) = self.append(&mut w, raw, false)?;
763                        self.state_w().relocate(rec.ns, rec.key, seg, noff);
764                    }
765                }
766                Op::Del => {
767                    if !drop_tombs && self.state_r().find(rec.ns, rec.key)?.is_none() {
768                        self.append(&mut w, raw, true)?;
769                    }
770                }
771            }
772            Ok(())
773        })?;
774        // 复制出去的记录落盘之后才能删旧段。
775        self.writer().file.sync_data()?;
776        {
777            let mut st = self.state_w();
778            let seg = st.segs.remove(&id).expect("segment");
779            debug_assert_eq!(seg.live, 0);
780            st.disk -= seg.total;
781            st.live -= seg.live;
782        }
783        fs::remove_file(seg_path(&self.0.cfg.dir, id))?;
784        sync_dir(&self.0.cfg.dir)?;
785        Ok(())
786    }
787}
788
789fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
790    m.lock().unwrap_or_else(PoisonError::into_inner)
791}
792
793pub(crate) struct Tx<'a> {
794    store: &'a Store,
795    w: MutexGuard<'a, Writer>,
796}
797
798impl Tx<'_> {
799    pub fn put(&mut self, ns: u8, key: &[u8], value: &[u8], ts_ms: u64) -> Result<bool> {
800        if key.len() > MAX_KEY {
801            return Err(Error::KeyTooLarge(key.len()));
802        }
803        let len = record::encoded_len(key, value);
804        if len > MAX_RECORD {
805            return Err(Error::ValueTooLarge {
806                size: value.len(),
807                limit: MAX_RECORD - (len - value.len()),
808            });
809        }
810        let raw = record::encode(ns, Op::Put, ts_ms, key, value);
811        let (seg, off) = self.store.append(&mut self.w, &raw, false)?;
812        let loc = Loc::new(seg, off, len as u32, ns, now_secs());
813        self.store.state_w().apply(ns, key, Some(loc))
814    }
815
816    pub fn delete(&mut self, ns: u8, key: &[u8]) -> Result<bool> {
817        if !self.store.contains(ns, key)? {
818            return Ok(false);
819        }
820        let raw = record::encode(ns, Op::Del, to_ms(SystemTime::now()), key, &[]);
821        self.store.append(&mut self.w, &raw, true)?;
822        self.store.state_w().apply(ns, key, None)
823    }
824
825    pub fn get(&self, ns: u8, key: &[u8]) -> Result<Option<(u64, Vec<u8>)>> {
826        self.store.get(ns, key)
827    }
828
829    pub fn contains(&self, ns: u8, key: &[u8]) -> Result<bool> {
830        self.store.contains(ns, key)
831    }
832
833    pub fn group(&self, ns: u8, name: &[u8]) -> Option<Group> {
834        self.store.group(ns, name)
835    }
836
837    /// 记下探测到的列表头，下次不用从头探测。
838    pub fn set_head(&self, ns: u8, name: &[u8], head: u64) {
839        if let Some(g) = self.store.state_w().groups.get_mut(&(ns, Box::from(name))) {
840            g.head = head;
841        }
842    }
843}
844
845impl Store {
846    /// 用 [`Store::tx`] 写完之后调用，按需触发后台维护。
847    pub(crate) fn finish(&self) {
848        self.maybe_maintain();
849    }
850}
pagesift_store/engine.rs

pagesift_store/
engine.rs
