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            loop {
89                let mut disconnected = false;
90
91                // 1. Log Compaction (High Fidelity Compression)
92                let msg_opt = rx.pop();
93                
94                match msg_opt {
95                    Some(msg) => {
96                        Self::compress_and_push(&mut batch, msg);
97                        while batch.len() < 65536 {
98                            if let Some(next_msg) = rx.pop() {
99                                Self::compress_and_push(&mut batch, next_msg);
100                            } else {
101                                break;
102                            }
103                        }
104                    }
105                    None => {
106                        if std::sync::Arc::strong_count(&rx) == 1 {
107                            disconnected = true;
108                        } else {
109                            // If we didn't receive anything, sleep for _poll_ms to avoid 100% CPU spinning
110                            thread::sleep(std::time::Duration::from_millis(_poll_ms));
111                        }
112                    }
113                }
114
115                // 2. Process batch and Broadcast
116                let mut last_hash: Option<usize> = None;
117                for msg in batch.drain(..) {
118                    match msg {
119                        DaemonMessage::Hit(hash, weight) => {
120                            core.record_remote_hit(hash, weight);
121                            if let Some(prev) = last_hash {
122                                core.set_prefetch_hint(prev, hash);
123                            }
124                            last_hash = Some(hash);
125                            for tx in &broadcast_txs {
126                                let _ = tx.push((hash, weight));
127                            }
128                        }
129                        DaemonMessage::HitBatch(_, _) => unreachable!(),
130                        DaemonMessage::Promote(_hash, key, value, tier) => {
131                            if tier == 0 {
132                                core.put_t0(key, value, daemon_node);
133                            } else {
134                                core.put(key, value, daemon_node);
135                            }
136                        }
137                        DaemonMessage::SetPollInterval(ms) => {
138                            _poll_ms = ms;
139                        }
140                        DaemonMessage::Sync(ack) => {
141                            ack.signal();
142                        }
143                        DaemonMessage::Shutdown => {
144                            disconnected = true;
145                        }
146                    }
147                }
148
149                // Pin daemon node so it updates its QSBR epoch and participates in GC
150                let _guard = crate::componant::qsbr::pin(daemon_node);
151                
152                // GC: Move safe nodes from thread-local garbage queues to the Arena directly
153                crate::componant::qsbr::daemon_reclaim(|batch| {
154                    if batch.is_empty() { return; }
155                    unsafe {
156                        for i in 0..batch.len() {
157                            let idx = batch[i];
158                            core.arena.drop_node(idx as usize);
159                            if i < batch.len() - 1 {
160                                core.arena.set_next_free(idx, batch[i + 1]);
161                            }
162                        }
163                        core.arena.free_batch(batch[0], batch[batch.len() - 1]);
164                    }
165                });
166
167                if disconnected {
168                    let node_ref = unsafe { &*daemon_node };
169                    node_ref.active.store(false, Ordering::Release);
170                    break;
171                }
172            }
173        });
174        Self { handle: Some(handle) }
175    }
176}
177
178
179
180use core::sync::atomic::{AtomicBool, Ordering};
181use std::sync::Arc;
182
183pub struct OneshotAck {
184    ready: AtomicBool,
185}
186
187impl OneshotAck {
188    pub fn new() -> Arc<Self> {
189        Arc::new(Self {
190            ready: AtomicBool::new(false),
191        })
192    }
193
194    #[inline(always)]
195    pub fn signal(&self) {
196        self.ready.store(true, Ordering::Release);
197    }
198
199    #[inline(always)]
200    pub fn wait(&self) {
201        let mut spins = 0;
202        while !self.ready.load(Ordering::Acquire) {
203            if spins < 100 {
204                core::hint::spin_loop();
205                spins += 1;
206            } else {
207                std::thread::yield_now();
208            }
209        }
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216
217    #[test]
218    fn test_daemon_compress_and_push() {
219        let mut batch: std::vec::Vec<DaemonMessage<u64, u64>> = std::vec::Vec::new();
220        Daemon::compress_and_push(&mut batch, DaemonMessage::Hit(1, 10));
221        Daemon::compress_and_push(&mut batch, DaemonMessage::Hit(1, 5));
222        Daemon::compress_and_push(&mut batch, DaemonMessage::Hit(2, 5));
223        
224        let mut arr = [(0usize, 0u8); 32];
225        arr[0] = (2, 5);
226        arr[1] = (3, 10);
227        Daemon::compress_and_push(&mut batch, DaemonMessage::HitBatch(arr, 2));
228    }
229}
dualcache_ff/componant/daemon/mod.rs

dualcache_ff/componant/daemon/
mod.rs
