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    free_list: no_std_tool::sync::SpinMutex<no_std_tool::collections::Vec<usize, MAX_THREADS>>,
173}
174
175unsafe 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> {}
176unsafe 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> {}
177
178impl<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> {
179    #[cfg(feature = "std")]
180    pub fn clear_channels(&self) {
181        for i in 0..MAX_THREADS {
182            let block = unsafe { &mut *self.blocks[i].get() };
183            block.value.tx = None;
184            block.value.hit_rx = None;
185        }
186    }
187}
188
189impl<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> {
190    fn default() -> Self {
191        Self::new()
192    }
193}
194
195impl<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> {
196    pub const fn new() -> Self {
197        Self {
198            blocks: [const { UnsafeCell::new(no_std_tool::sync::CachePadded { value: TlsBlock::new() }) }; MAX_THREADS],
199            next_id: AtomicUsize::new(0),
200            free_list: no_std_tool::sync::SpinMutex::new(no_std_tool::collections::Vec::new()),
201        }
202    }
203
204    pub fn max_threads(&self) -> usize {
205        MAX_THREADS
206    }
207
208    pub fn get_metrics(&self) -> (u64, u64) {
209        let mut total_ops = 0;
210        let mut total_hits = 0;
211        let active = self.next_id.load(Ordering::Relaxed);
212        for i in 0..active {
213            let block = unsafe { &*self.blocks[i].get() };
214            total_ops += block.value.op_count;
215            total_hits += block.value.hit_count;
216        }
217        (total_ops, total_hits)
218    }
219
220    pub fn register_thread(&self) -> TlsHandle {
221        let mut id = usize::MAX;
222        if let Ok(Some(free_id)) = self.free_list.lock().map(|mut f| f.pop()) {
223            id = free_id;
224        }
225        
226        let mut is_new = false;
227        if id == usize::MAX {
228            id = self.next_id.fetch_add(1, Ordering::Relaxed);
229            if id >= MAX_THREADS {
230                panic!("Exceeded max thread capacity in TlsRegistry");
231            }
232            is_new = true;
233        }
234
235        let block = unsafe { &mut (*self.blocks[id].get()).value };
236        let qsbr_node = &mut block.qsbr_node as *mut _;
237        
238        unsafe fn free_id_trampoline<K, V, const MAX_THREADS: usize, const TLS_CAP: usize, const TLS_INDEX_CAP: usize>(
239            registry: *const core::ffi::c_void,
240            id: usize,
241        ) {
242            let registry = unsafe { &*(registry as *const TlsRegistry<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP>) };
243            if let Ok(mut free_list) = registry.free_list.lock() {
244                let _ = free_list.push(id);
245            }
246        }
247        let trampoline = Some(free_id_trampoline::<K, V, MAX_THREADS, TLS_CAP, TLS_INDEX_CAP> as unsafe fn(*const core::ffi::c_void, usize));
248        let registry_ptr = self as *const _ as *const core::ffi::c_void;
249        
250        if is_new {
251            crate::componant::qsbr::register_node(qsbr_node, id, registry_ptr, trampoline);
252        } else {
253            // Already registered, just reactivate it
254            unsafe { (*qsbr_node).active.store(true, Ordering::Release) };
255            
256            #[cfg(feature = "std")]
257            crate::componant::qsbr::reregister_cleanup(qsbr_node, id, registry_ptr, trampoline);
258        }
259        
260        TlsHandle { id, qsbr_node }
261    }
262
263    pub fn deregister_thread(&self, handle: &TlsHandle) {
264        if let Ok(mut free_list) = self.free_list.lock() {
265            let _ = free_list.push(handle.id);
266        }
267    }
268
269    #[inline]
270    #[allow(clippy::mut_from_ref)]
271    pub fn get_block_mut(&self, handle: &TlsHandle) -> &mut TlsBlock<K, V, TLS_CAP, TLS_INDEX_CAP> {
272        let block_ptr = self.blocks[handle.id].get();
273        unsafe { &mut (*block_ptr).value }
274    }
275}
dualcache_ff/componant/tls/mod.rs

dualcache_ff/componant/tls/
mod.rs
