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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
//! Background periodic-flush worker.
//!
//! Mirrors [`Compactor`](crate::compaction::Compactor): a thread that wakes on a
//! fixed interval and runs a flush closure, wired to the shared
//! [`ShutdownSignal`](crate::shutdown::ShutdownSignal) so it stops together with
//! the rest of the database's background workers. Unlike compaction, a failing
//! flush is logged and retried on the next tick rather than ending the loop —
//! a missed flush degrades durability but is not a reason to kill the worker.
use std::time::Duration;
use crate::error::DbResult;
/// Background periodic-flush handle. Stops and joins its thread on `stop()` or
/// `Drop`.
pub struct Flusher {
stop: crate::shutdown::ShutdownSignal,
handle: Option<std::thread::JoinHandle<()>>,
}
impl Flusher {
/// Start a background flush thread with its own shutdown signal.
pub fn start(flush_fn: impl Fn() -> DbResult<()> + Send + 'static, interval: Duration) -> Self {
Self::start_with_signal(flush_fn, interval, crate::shutdown::ShutdownSignal::new())
}
/// Start a background flush thread controlled by an external shutdown
/// signal. When the signal fires the thread wakes immediately instead of
/// waiting for the full interval.
pub fn start_with_signal(
flush_fn: impl Fn() -> DbResult<()> + Send + 'static,
interval: Duration,
signal: crate::shutdown::ShutdownSignal,
) -> Self {
let stop = signal.clone();
let handle = std::thread::spawn(move || {
while !stop.is_shutdown() {
if stop.wait_timeout(interval) {
break;
}
if let Err(e) = flush_fn() {
tracing::error!(error = %e, "flush error");
}
}
});
Self {
stop: signal,
handle: Some(handle),
}
}
/// Signal the worker to stop and join its thread.
pub fn stop(&mut self) {
self.stop.shutdown();
if let Some(h) = self.handle.take() {
let _ = h.join();
}
}
}
impl Drop for Flusher {
fn drop(&mut self) {
self.stop();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
fn wait_until(pred: impl Fn() -> bool, timeout: Duration) -> bool {
let start = std::time::Instant::now();
while start.elapsed() < timeout {
if pred() {
return true;
}
std::thread::sleep(Duration::from_millis(5));
}
pred()
}
#[test]
fn flusher_ticks_and_stop_joins() {
let n = Arc::new(AtomicUsize::new(0));
let n2 = n.clone();
let mut f = Flusher::start(
move || {
n2.fetch_add(1, Ordering::SeqCst);
Ok(())
},
Duration::from_millis(10),
);
assert!(
wait_until(|| n.load(Ordering::SeqCst) >= 2, Duration::from_secs(2)),
"flusher did not tick at least twice"
);
f.stop();
let after = n.load(Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(50));
assert_eq!(
n.load(Ordering::SeqCst),
after,
"flusher kept ticking after stop()"
);
}
#[test]
fn flusher_survives_error() {
let n = Arc::new(AtomicUsize::new(0));
let n2 = n.clone();
let mut f = Flusher::start(
move || {
let c = n2.fetch_add(1, Ordering::SeqCst);
if c % 2 == 0 {
Err(crate::error::DbError::Config("boom"))
} else {
Ok(())
}
},
Duration::from_millis(10),
);
assert!(
wait_until(|| n.load(Ordering::SeqCst) >= 4, Duration::from_secs(2)),
"flusher stopped after an error"
);
f.stop();
}
}