dualcache_ff/componant/daemon/
mod.rs

1use std::thread::{self, JoinHandle};
2use ::core::hash::Hash;
3
4
5
6#[allow(clippy::large_enum_variant)]
7pub enum DaemonMessage<K, V> {
8    Hit(usize, u8),           // hash, weight
9    HitBatch([(usize, u8); 32], u8), // batch of hits
10    Promote(usize, K, V, u8), // hash, key, value, tier (0=T0, 2=T2)
11    /// Dynamically adjust Daemon poll interval (Power-Saving Mode) - missing from v0.5.0
12    SetPollInterval(u64),
13    /// Zero-cost callbacks / blocking maintenance flush - missing from v0.5.0
14    Sync(std::sync::Arc<OneshotAck>),
15    /// Graceful shutdown - missing from v0.5.0
16    Shutdown,
17}
18
19/// The Daemon manages background tasks like TLS-to-Core promotion
20/// and QSBR memory reclamation.
21pub struct Daemon {
22    handle: Option<JoinHandle<()>>,
23}
24
25impl Daemon {
26    pub fn join(&mut self) {
27        if let Some(handle) = self.handle.take() {
28            let _ = handle.join();
29        }
30    }
31
32    fn compress_and_push<K, V>(batch: &mut std::vec::Vec<DaemonMessage<K, V>>, msg: DaemonMessage<K, V>) {
33        match msg {
34            DaemonMessage::Hit(hash, weight) => {
35                if let Some(DaemonMessage::Hit(last_hash, last_weight)) = batch.last_mut()
36                    && *last_hash == hash {
37                        *last_weight = last_weight.saturating_add(weight);
38                        return;
39                    }
40                batch.push(DaemonMessage::Hit(hash, weight));
41            }
42            DaemonMessage::HitBatch(arr, len) => {
43                for &(hash, weight) in arr[..(len as usize)].iter() {
44                    let mut found = false;
45                    if let Some(DaemonMessage::Hit(last_hash, last_weight)) = batch.last_mut()
46                        && *last_hash == hash {
47                            *last_weight = last_weight.saturating_add(weight);
48                            found = true;
49                        }
50                    if !found {
51                        batch.push(DaemonMessage::Hit(hash, weight));
52                    }
53                }
54            }
55            DaemonMessage::Promote(hash, key, val, tier) => {
56                batch.push(DaemonMessage::Promote(hash, key, val, tier));
57            }
58            DaemonMessage::SetPollInterval(ms) => {
59                batch.push(DaemonMessage::SetPollInterval(ms));
60            }
61            DaemonMessage::Sync(ack) => {
62                batch.push(DaemonMessage::Sync(ack));
63            }
64            DaemonMessage::Shutdown => {
65                batch.push(DaemonMessage::Shutdown);
66            }
67        }
68    }
69
70    /// Spawn the daemon thread. Returns the Daemon handle.
71    #[inline(never)]
72    pub fn spawn<K, V, P, const CAP2: usize, const CAP1: usize, const CAP0: usize, const TOTAL_CAP: usize>(
73        core: &'static crate::core::DualCacheCore<K, V, P, CAP2, CAP1, CAP0, TOTAL_CAP>, 
74        rx: std::sync::Arc<no_std_tool::collections::mpsc_queue::BoundedQueue<DaemonMessage<K, V>, 65536>>,
75        broadcast_txs: std::vec::Vec<std::sync::Arc<no_std_tool::collections::mpsc_queue::BoundedQueue<(usize, u8), 1024>>>,
76        daemon_node: *mut crate::componant::qsbr::ThreadStateNode
77    ) -> Self
78    where
79        K: Clone + Eq + Hash + Send + Sync + 'static,
80        V: Clone + Send + Sync + 'static,
81        P: crate::componant::config::CachePolicy + Send + Sync + 'static,
82    {
83        let daemon_node_ptr = daemon_node as usize;
84        let handle = thread::spawn(move || {
85            let daemon_node = daemon_node_ptr as *mut crate::componant::qsbr::ThreadStateNode;
86            let mut batch = std::vec::Vec::with_capacity(65536);
87            let mut _poll_ms = 10;
88            let mut empty_spins = 0u32;
89            loop {
90                let mut disconnected = false;
91
92                // 1. Log Compaction (High Fidelity Compression)
93                let msg_opt = rx.pop();
94                
95                match msg_opt {
96                    Some(msg) => {
97                        empty_spins = 0;
98                        Self::compress_and_push(&mut batch, msg);
99                        while batch.len() < 65536 {
100                            if let Some(next_msg) = rx.pop() {
101                                Self::compress_and_push(&mut batch, next_msg);
102                            } else {
103                                break;
104                            }
105                        }
106                    }
107                    None => {
108                        empty_spins = empty_spins.saturating_add(1);
109                        if std::sync::Arc::strong_count(&rx) == 1 {
110                            disconnected = true;
111                        } else {
112                            if empty_spins < 100 {
113                                core::hint::spin_loop();
114                            } else if empty_spins < 200 {
115                                std::thread::yield_now();
116                            } else {
117                                let shift = core::cmp::min((empty_spins - 200) / 10, 6);
118                                let backoff = 1u64 << shift;
119                                let sleep_ms = core::cmp::min(backoff, _poll_ms);
120                                thread::sleep(std::time::Duration::from_millis(sleep_ms));
121                            }
122                        }
123                    }
124                }
125
126                // 2. Process batch and Broadcast
127                let mut last_hash: Option<usize> = None;
128                for msg in batch.drain(..) {
129                    match msg {
130                        DaemonMessage::Hit(hash, weight) => {
131                            core.record_remote_hit(hash, weight);
132                            if let Some(prev) = last_hash {
133                                core.set_prefetch_hint(prev, hash);
134                            }
135                            last_hash = Some(hash);
136                            for tx in &broadcast_txs {
137                                let _ = tx.push((hash, weight));
138                            }
139                        }
140                        DaemonMessage::HitBatch(_, _) => unreachable!(),
141                        DaemonMessage::Promote(_hash, key, value, tier) => {
142                            if tier == 0 {
143                                core.put_t0(key, value, daemon_node);
144                            } else {
145                                core.put(key, value, daemon_node);
146                            }
147                        }
148                        DaemonMessage::SetPollInterval(ms) => {
149                            _poll_ms = ms;
150                        }
151                        DaemonMessage::Sync(ack) => {
152                            ack.signal();
153                        }
154                        DaemonMessage::Shutdown => {
155                            disconnected = true;
156                        }
157                    }
158                }
159
160                // Pin daemon node so it updates its QSBR epoch and participates in GC
161                let _guard = crate::componant::qsbr::pin(daemon_node);
162                
163                // GC: Move safe nodes from thread-local garbage queues to the Arena directly
164                crate::componant::qsbr::daemon_reclaim(|batch| {
165                    if batch.is_empty() { return; }
166                    unsafe {
167                        for i in 0..batch.len() {
168                            let idx = batch[i];
169                            core.arena.drop_node(idx as usize);
170                            if i < batch.len() - 1 {
171                                core.arena.set_next_free(idx, batch[i + 1]);
172                            }
173                        }
174                        core.arena.free_batch(batch[0], batch[batch.len() - 1]);
175                    }
176                });
177
178                if disconnected {
179                    let node_ref = unsafe { &*daemon_node };
180                    node_ref.active.store(false, Ordering::Release);
181                    break;
182                }
183            }
184        });
185        Self { handle: Some(handle) }
186    }
187}
188
189
190
191use core::sync::atomic::{AtomicBool, Ordering};
192use std::sync::Arc;
193
194pub struct OneshotAck {
195    ready: AtomicBool,
196}
197
198impl OneshotAck {
199    pub fn new() -> Arc<Self> {
200        Arc::new(Self {
201            ready: AtomicBool::new(false),
202        })
203    }
204
205    #[inline(always)]
206    pub fn signal(&self) {
207        self.ready.store(true, Ordering::Release);
208    }
209
210    #[inline(always)]
211    pub fn wait(&self) {
212        let mut spins = 0;
213        while !self.ready.load(Ordering::Acquire) {
214            if spins < 100 {
215                core::hint::spin_loop();
216                spins += 1;
217            } else {
218                std::thread::yield_now();
219            }
220        }
221    }
222}
223
224#[cfg(test)]
225mod tests {
226    use super::*;
227
228    #[test]
229    fn test_daemon_compress_and_push() {
230        let mut batch: std::vec::Vec<DaemonMessage<u64, u64>> = std::vec::Vec::new();
231        Daemon::compress_and_push(&mut batch, DaemonMessage::Hit(1, 10));
232        Daemon::compress_and_push(&mut batch, DaemonMessage::Hit(1, 5));
233        Daemon::compress_and_push(&mut batch, DaemonMessage::Hit(2, 5));
234        
235        let mut arr = [(0usize, 0u8); 32];
236        arr[0] = (2, 5);
237        arr[1] = (3, 10);
238        Daemon::compress_and_push(&mut batch, DaemonMessage::HitBatch(arr, 2));
239    }
240}
dualcache_ff/componant/daemon/mod.rs

dualcache_ff/componant/daemon/
mod.rs
