dualcache_ff/componant/tls/
mod.rs

1
2use core::cell::UnsafeCell;
3use ::core::sync::atomic::{AtomicUsize, Ordering};
4
5pub struct TlsHandle {
6    pub id: usize,
7    pub qsbr_node: *mut crate::componant::qsbr::ThreadStateNode,
8}
9
10unsafe impl Send for TlsHandle {}
11unsafe impl Sync for TlsHandle {}
12
13impl Drop for TlsHandle {
14    #[inline(always)]
15    fn drop(&mut self) {
16        if !self.qsbr_node.is_null() {
17            unsafe {
18                (*self.qsbr_node).active.store(false, Ordering::Release);
19            }
20        }
21    }
22}
23
24#[derive(Clone)]
25pub struct TlsEntry<K, V> {
26    pub hash: usize,
27    pub key: K,
28    pub value: V,
29    pub hits: u8,
30}
31
32pub struct TlsCache<K, V, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> {
33    slots: [Option<TlsEntry<K, V>>; TLS_CAP],
34    capacity: usize,
35    pub promote_threshold: u8,
36    probation_filter: [u8; 65536],
37    probation_cursor: usize,
38}
39
40impl<K: Clone + Eq, V: Clone, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> Default for TlsCache<K, V, TLS_CAP, TLS_INDEX_CAP> {
41    fn default() -> Self {
42        Self::new()
43    }
44}
45
46impl<K: Clone + Eq, V: Clone, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> TlsCache<K, V, TLS_CAP, TLS_INDEX_CAP> {
47    pub const fn new() -> Self {
48        Self { 
49            slots: [const { None }; TLS_CAP],
50            capacity: TLS_CAP,
51            promote_threshold: 4,
52            probation_filter: [0; 65536],
53            probation_cursor: 0,
54        }
55    }
56
57    #[inline(always)]
58    pub fn get(&mut self, hash: usize, key: &K) -> (Option<&V>, bool, u8) {
59        let idx = hash & (self.capacity - 1);
60        if let Some(entry) = unsafe { self.slots.get_unchecked_mut(idx) }
61            && entry.hash == hash && entry.key == *key {
62                let old_hits = entry.hits;
63                entry.hits = entry.hits.saturating_add(1);
64                
65                let promote = old_hits < self.promote_threshold && entry.hits >= self.promote_threshold;
66                
67                let sync_weight = if entry.hits > self.promote_threshold && (entry.hits & 15) == 0 {
68                    16
69                } else {
70                    0
71                };
72                
73                return (Some(&entry.value), promote, sync_weight);
74        }
75        (None, false, 0)
76    }
77
78    #[inline(always)]
79    pub fn insert(&mut self, hash: usize, key: K, value: V) -> bool {
80        let idx = hash & (self.capacity - 1);
81        if let Some(entry) = unsafe { self.slots.get_unchecked_mut(idx) }
82            && entry.hash == hash && entry.key == key {
83                entry.value = value;
84                return true;
85        }
86        
87        let filter_idx = hash & 4095;
88        
89        // Slower Probation Filter: clear 1 element every 16 inserts
90        self.probation_cursor = (self.probation_cursor + 1) & 65535;
91        if (self.probation_cursor & 15) == 0 {
92            unsafe { *self.probation_filter.get_unchecked_mut(self.probation_cursor >> 4) = 0; }
93        }
94        
95        let count = unsafe { *self.probation_filter.get_unchecked(filter_idx) }.saturating_add(1);
96        unsafe { *self.probation_filter.get_unchecked_mut(filter_idx) = count; }
97        
98        if count < 2 {
99            return false;
100        }
101        
102        unsafe { *self.slots.get_unchecked_mut(idx) = Some(TlsEntry { hash, key, value, hits: 0 }); }
103        true
104    }
105
106    #[inline(always)]
107    pub fn insert_fast_pass(&mut self, hash: usize, key: K, value: V) {
108        let idx = hash & (self.capacity - 1);
109        let hits = self.promote_threshold;
110        unsafe { *self.slots.get_unchecked_mut(idx) = Some(TlsEntry { hash, key, value, hits }); }
111    }
112
113    pub fn record_remote_hit(&mut self, hash: usize, weight: u8) {
114        let idx = hash & (self.capacity - 1);
115        if let Some(entry) = unsafe { self.slots.get_unchecked_mut(idx) }
116            && entry.hash == hash {
117                entry.hits = entry.hits.saturating_add(weight);
118        }
119    }
120}
121
122#[cfg(feature = "std")]
123use crate::componant::daemon::DaemonMessage;
124
125/// A block of TLS data representing the state for a single thread.
126pub struct TlsBlock<K, V, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> {
127    pub cache: TlsCache<K, V, TLS_CAP, TLS_INDEX_CAP>,
128    #[cfg(feature = "std")]
129    pub tx: Option<std::sync::Arc<no_std_tool::collections::mpsc_queue::BoundedQueue<DaemonMessage<K, V>, 65536>>>,
130    #[cfg(feature = "std")]
131    pub hit_rx: Option<std::sync::Arc<no_std_tool::collections::mpsc_queue::BoundedQueue<(usize, u8), 1024>>>,
132    pub op_count: u64,
133    pub hit_count: u64,
134    #[cfg(feature = "std")]
135    pub hit_batch: [(usize, u8); 32],
136    #[cfg(feature = "std")]
137    pub hit_batch_len: u8,
138    pub warmup_state: u16,
139    pub qsbr_node: crate::componant::qsbr::ThreadStateNode,
140}
141
142impl<K: Clone + Eq, V: Clone, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> Default for TlsBlock<K, V, TLS_CAP, TLS_INDEX_CAP> {
143    fn default() -> Self {
144        Self::new()
145    }
146}
147
148impl<K: Clone + Eq, V: Clone, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> TlsBlock<K, V, TLS_CAP, TLS_INDEX_CAP> {
149    pub const fn new() -> Self {
150        Self {
151            cache: TlsCache::new(),
152            #[cfg(feature = "std")]
153            tx: None,
154            #[cfg(feature = "std")]
155            hit_rx: None,
156            op_count: 0,
157            hit_count: 0,
158            #[cfg(feature = "std")]
159            hit_batch: [(0, 0); 32],
160            #[cfg(feature = "std")]
161            hit_batch_len: 0,
162            warmup_state: 0,
163            qsbr_node: crate::componant::qsbr::ThreadStateNode::new(),
164        }
165    }
166}
167
168/// Registry for managing Thread-Local Caches dynamically without OS TLS.
169pub struct TlsRegistry<K, V, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> {
170    blocks: [UnsafeCell<no_std_tool::sync::CachePadded<TlsBlock<K, V, TLS_CAP, TLS_INDEX_CAP>>>; MAX_THREADS],
171    next_id: AtomicUsize,
172}
173
174unsafe impl<K, V, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> Sync for TlsRegistry<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP> {}
175unsafe impl<K, V, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> Send for TlsRegistry<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP> {}
176
177impl<K, V, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> TlsRegistry<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP> {
178    #[cfg(feature = "std")]
179    pub fn clear_channels(&self) {
180        for i in 0..MAX_THREADS {
181            let block = unsafe { &mut *self.blocks[i].get() };
182            block.value.tx = None;
183            block.value.hit_rx = None;
184        }
185    }
186}
187
188impl<K: Clone + Eq, V: Clone, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> Default for TlsRegistry<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP> {
189    fn default() -> Self {
190        Self::new()
191    }
192}
193
194impl<K: Clone + Eq, V: Clone, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize> TlsRegistry<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP> {
195    pub const fn new() -> Self {
196        Self {
197            blocks: [const { UnsafeCell::new(no_std_tool::sync::CachePadded { value: TlsBlock::new() }) }; MAX_THREADS],
198            next_id: AtomicUsize::new(0),
199        }
200    }
201
202    pub fn max_threads(&self) -> usize {
203        MAX_THREADS
204    }
205
206    pub fn get_metrics(&self) -> (u64, u64) {
207        let mut total_ops = 0;
208        let mut total_hits = 0;
209        let active = self.next_id.load(Ordering::Relaxed);
210        for i in 0..active {
211            let block = unsafe { &*self.blocks[i].get() };
212            total_ops += block.value.op_count;
213            total_hits += block.value.hit_count;
214        }
215        (total_ops, total_hits)
216    }
217
218    pub fn register_thread(&self) -> TlsHandle {
219        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
220        if id >= MAX_THREADS {
221            panic!("Exceeded max thread capacity in TlsRegistry");
222        }
223        let block = unsafe { &mut (*self.blocks[id].get()).value };
224        let qsbr_node = &mut block.qsbr_node as *mut _;
225        crate::componant::qsbr::register_node(qsbr_node);
226        TlsHandle { id, qsbr_node }
227    }
228
229    #[inline]
230    #[allow(clippy::mut_from_ref)]
231    pub fn get_block_mut(&self, handle: &TlsHandle) -> &mut TlsBlock<K, V, TLS_CAP, TLS_INDEX_CAP> {
232        let block_ptr = self.blocks[handle.id].get();
233        unsafe { &mut (*block_ptr).value }
234    }
235}
dualcache_ff/componant/tls/mod.rs

dualcache_ff/componant/tls/
mod.rs
