dualcache_ff/componant/
workers.rs

1use std::vec::Vec;
2
3use core::cell::UnsafeCell;
4use core::mem::MaybeUninit;
5
6/// Zero-allocation batch buffer: fixed-size MaybeUninit array, reused in-place.
7/// No Mutex, no Vec, no heap allocation on the hot path.
8///
9/// Cache-line aligned to prevent false sharing between worker slots.
10#[cfg_attr(any(target_arch = "aarch64", target_arch = "arm"), repr(C, align(128)))]
11#[cfg_attr(not(any(target_arch = "aarch64", target_arch = "arm")), repr(C, align(64)))]
12pub struct BatchBuf<K, V> {
13    items: [MaybeUninit<(K, V, u64)>; 32],
14    len: usize,
15}
16
17impl<K, V> BatchBuf<K, V> {
18    pub fn new() -> Self {
19        Self {
20            items: unsafe { MaybeUninit::uninit().assume_init() },
21            len: 0,
22        }
23    }
24}
25
26impl<K, V> Default for BatchBuf<K, V> {
27    fn default() -> Self {
28        Self::new()
29    }
30}
31
32impl<K, V> BatchBuf<K, V> {
33    /// Returns `true` when the buffer is full (32 items) and should be flushed.
34    #[inline(always)]
35    pub fn push(&mut self, item: (K, V, u64)) -> bool {
36        self.items[self.len] = MaybeUninit::new(item);
37        self.len += 1;
38        self.len == 32
39    }
40
41    pub fn len(&self) -> usize {
42        self.len
43    }
44
45    pub fn is_empty(&self) -> bool {
46        self.len == 0
47    }
48
49    /// Drains all items into a `Vec`, resetting the buffer.
50    pub fn drain_to_vec(&mut self) -> Vec<(K, V, u64)> {
51        let mut batch = Vec::with_capacity(self.len);
52        for i in 0..self.len {
53            batch.push(unsafe { self.items[i].assume_init_read() });
54        }
55        self.len = 0;
56        batch
57    }
58}
59
60impl<K, V> Drop for BatchBuf<K, V> {
61    fn drop(&mut self) {
62        for i in 0..self.len {
63            unsafe {
64                self.items[i].assume_init_drop();
65            }
66        }
67    }
68}
69
70unsafe impl<K: Send, V: Send> Send for BatchBuf<K, V> {}
71unsafe impl<K: Sync, V: Sync> Sync for BatchBuf<K, V> {}
72
73/// Per-worker exclusive slot holding a `BatchBuf` inside an `UnsafeCell`.
74///
75/// The WORKER_ID TLS invariant guarantees that only one thread ever accesses
76/// any given slot, eliminating the need for any synchronisation primitive on
77/// the insert hot-path (zero atomics, zero locks, pure memory write).
78#[cfg_attr(any(target_arch = "aarch64", target_arch = "arm"), repr(C, align(128)))]
79#[cfg_attr(not(any(target_arch = "aarch64", target_arch = "arm")), repr(C, align(64)))]
80pub struct WorkerSlot<K, V> {
81    inner: UnsafeCell<BatchBuf<K, V>>,
82}
83
84impl<K, V> WorkerSlot<K, V> {
85    pub fn new() -> Self {
86        Self {
87            inner: UnsafeCell::new(BatchBuf::new()),
88        }
89    }
90}
91
92impl<K, V> Default for WorkerSlot<K, V> {
93    fn default() -> Self {
94        Self::new()
95    }
96}
97
98impl<K, V> WorkerSlot<K, V> {
99    /// Provides exclusive access to the underlying buffer.
100    ///
101    /// # Safety
102    /// The caller must guarantee that only one thread accesses this slot at a time.
103    /// In DualCache-FF this is enforced by the WORKER_ID TLS invariant.
104    #[inline(always)]
105    #[allow(clippy::mut_from_ref)]
106    pub unsafe fn get_mut_unchecked(&self) -> &mut BatchBuf<K, V> {
107        unsafe { &mut *self.inner.get() }
108    }
109}
110
111unsafe impl<K: Send, V: Send> Send for WorkerSlot<K, V> {}
112unsafe impl<K: Send + Sync, V: Send + Sync> Sync for WorkerSlot<K, V> {}
dualcache_ff/componant/workers.rs

dualcache_ff/componant/
workers.rs
