logdb 0.3.0

Embedded, append-only, crash-recoverable, optionally tamper-proof local log database
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

//! Commit trigger logic and backoff/wait strategies.
//!
//! The [`CommitTrigger`] decides when the Committer should flush a batch of
//! records to disk based on byte count, record count, and time thresholds.
//!
//! Types are re-exported from [`crate::config`] for use by the pipeline.

use std::hint;
use std::thread;
use std::time::Duration;

// Re-export config types used in this module (CommitTrigger is imported by
// lib.rs via this path).
pub use crate::config::{CommitTrigger, WaitStrategy};

/// Stateful backoff helper for a single thread.
///
/// Tracks spin/yield/sleep phase across iterations.
pub struct Backoff {
    spins: u32,
    config: WaitStrategy,
}

impl Backoff {
    /// Create a new backoff with the given strategy.
    pub fn new(config: WaitStrategy) -> Self {
        Self { spins: 0, config }
    }

    /// Execute one backoff step.
    ///
    /// Progresses through spin → yield → park phases based on call count.
    /// Call `reset()` when work is found to restart the backoff progression.
    pub fn step(&mut self) {
        self.spins = self.spins.saturating_add(1);
        if self.spins <= self.config.spin_count {
            hint::spin_loop();
        } else if self.spins <= self.config.spin_count + self.config.yield_count {
            thread::yield_now();
        } else {
            thread::sleep(self.config.park_duration);
        }
    }

    /// Reset the backoff state (call when work is found).
    pub fn reset(&mut self) {
        self.spins = 0;
    }
}

/// Producer-side backoff for when the ring is full.
#[inline]
pub fn producer_backoff(spins: &mut u32) {
    *spins = spins.saturating_add(1);
    if *spins <= 64 {
        hint::spin_loop();
    } else if *spins <= 256 {
        thread::yield_now();
    } else {
        thread::sleep(Duration::from_micros(100));
        *spins = 128;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn backoff_resets() {
        let mut bo = Backoff::new(WaitStrategy::default());
        // Spin a few times
        for _ in 0..10 {
            bo.step();
        }
        assert!(bo.spins > 0);
        bo.reset();
        assert_eq!(bo.spins, 0);
    }

    #[test]
    fn commit_trigger_defaults() {
        let t = CommitTrigger::default();
        assert_eq!(t.bytes, 256 * 1024);
        assert_eq!(t.records, 1024);
        assert_eq!(t.interval, Duration::from_millis(10));
    }
}