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
use crate::store::state_machine::memory::kv_handler::CacheRequestHandler;
use crate::store::state_machine::memory::state_machine::{CacheResponse, StateMachineData};
use chrono::Utc;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{RwLock, oneshot};
use tokio::{task, time};
use tracing::{debug, warn};
#[derive(Debug)]
pub enum TtlRequest {
Ttl((i64, String)),
SnapshotBuild(oneshot::Sender<BTreeMap<i64, String>>),
SnapshotInstall((BTreeMap<i64, String>, oneshot::Sender<()>)),
}
pub fn spawn(tx_kv: flume::Sender<CacheRequestHandler>) -> flume::Sender<TtlRequest> {
let (tx, rx) = flume::unbounded();
task::spawn(ttl_handler(tx_kv, rx));
tx
}
async fn ttl_handler(tx_kv: flume::Sender<CacheRequestHandler>, rx: flume::Receiver<TtlRequest>) {
let mut data: BTreeMap<i64, String> = BTreeMap::new();
loop {
let sleep_exp = {
let first_exp = data
.first_entry()
.map(|e| *e.key() - Utc::now().timestamp());
if let Some(exp) = first_exp {
if exp < 1 {
let key = data.pop_first().unwrap().1;
tx_kv
.send(CacheRequestHandler::Delete(key))
.expect("kv handler to always be running");
continue;
} else {
Duration::from_secs(exp as u64)
}
} else {
Duration::from_secs(u64::MAX)
}
};
tokio::select! {
// req = rx.recv_async() => {
// if let Ok((ttl, key)) = req {
// // TODO currently we use microsecond precision and this could overlap
// // if very unlucky -> use nano's or da an additional step ech time to check if
// // the entry exists already? otherwise a value will not be expired
// data.insert(ttl, key);
// } else {
// break;
// }
// }
req = rx.recv_async() => {
if let Ok(req) = req {
// TODO currently we use microsecond precision and this could overlap
// if very unlucky -> use nano's or da an additional step ech time to check if
// the entry exists already? otherwise a value will not be expired
match req {
TtlRequest::Ttl((ttl, key)) => {
data.insert(ttl, key);
}
TtlRequest::SnapshotBuild(ack) => {
ack.send(data.clone()).unwrap();
}
TtlRequest::SnapshotInstall((snap, ack)) => {
data = snap;
ack.send(()).unwrap();
}
}
} else {
break;
}
}
_ = time::sleep(sleep_exp) => {
debug!("Timeout reached - first entry in map expires");
}
}
}
debug!("cache::ttl_handler exiting");
}