feoxdb/core/
record.rs

1use bytes::Bytes;
2use crossbeam_epoch::{self as epoch, Atomic, Guard, Owned, Shared};
3use std::mem;
4use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
5use std::sync::{Arc, OnceLock, Weak};
6
7use crate::constants::*;
8
9const EXTENT_RETIRED: u32 = 1 << 31;
10const EXTENT_READERS: u32 = !EXTENT_RETIRED;
11
12/// The ordered index stores one slot per key rather than one node per record
13/// generation. An update swaps the slot's pointer under the hash bucket guard
14/// instead of removing and reinserting a skiplist node, so a key is never
15/// briefly absent from a range query, the two indexes cannot drift apart, and a
16/// range scan reads its values here rather than paying a hash lookup per key.
17#[derive(Debug)]
18pub(crate) struct TreeSlot {
19    record: Atomic<Arc<Record>>,
20}
21
22impl TreeSlot {
23    pub(crate) fn new(record: Arc<Record>) -> Self {
24        Self {
25            record: Atomic::new(record),
26        }
27    }
28
29    #[inline]
30    pub(crate) fn load<'g>(&'g self, guard: &'g Guard) -> &'g Arc<Record> {
31        let record = self.record.load(Ordering::Acquire, guard);
32        debug_assert!(!record.is_null());
33        unsafe { record.deref() }
34    }
35
36    #[inline]
37    pub(crate) fn store(&self, record: Arc<Record>) {
38        let guard = &epoch::pin();
39        let previous = self
40            .record
41            .swap(Owned::new(record), Ordering::AcqRel, guard);
42        if !previous.is_null() {
43            unsafe {
44                guard.defer_destroy(previous);
45            }
46        }
47    }
48}
49
50impl Drop for TreeSlot {
51    fn drop(&mut self) {
52        let record = mem::replace(&mut self.record, Atomic::null());
53        unsafe {
54            drop(record.into_owned());
55        }
56    }
57}
58
59pub(crate) struct ExtentReadGuard<'a>(&'a AtomicU32);
60
61impl Drop for ExtentReadGuard<'_> {
62    fn drop(&mut self) {
63        self.0.fetch_sub(1, Ordering::Release);
64    }
65}
66
67#[repr(C)]
68#[derive(Debug)]
69pub struct Record {
70    pub key: Vec<u8>,
71    pub value: parking_lot::RwLock<Option<Bytes>>,
72    pub ttl_expiry: AtomicU64,
73    pub timestamp: u64,
74    pub value_len: usize,
75    pub sector: AtomicU64,
76    pub refcount: AtomicU32,
77    pub key_len: u16,
78    pub hash_link: AtomicLink,
79    pub cache_ref_bit: AtomicU32,
80    pub cache_access_time: AtomicU64,
81    pub(crate) retired_at: AtomicU64,
82    successor: OnceLock<Arc<Record>>,
83    value_source: Option<Weak<Record>>,
84    successor_safe: AtomicBool,
85    extent_state: AtomicU32,
86}
87
88// Custom atomic link for lock-free hash table
89pub struct AtomicLink {
90    pub next: Atomic<Record>,
91}
92
93impl std::fmt::Debug for AtomicLink {
94    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95        f.debug_struct("AtomicLink")
96            .field("next", &"<atomic>")
97            .finish()
98    }
99}
100
101impl Default for AtomicLink {
102    fn default() -> Self {
103        Self::new()
104    }
105}
106
107impl AtomicLink {
108    pub fn new() -> Self {
109        Self {
110            next: Atomic::null(),
111        }
112    }
113
114    pub fn load<'g>(&self, guard: &'g Guard) -> Option<Shared<'g, Record>> {
115        let ptr = self.next.load(Ordering::Acquire, guard);
116        if ptr.is_null() {
117            None
118        } else {
119            Some(ptr)
120        }
121    }
122
123    pub fn store(&self, record: Option<Shared<Record>>, _guard: &Guard) {
124        let ptr = record.unwrap_or(Shared::null());
125        self.next.store(ptr, Ordering::Release);
126    }
127
128    pub fn compare_exchange<'g>(
129        &self,
130        current: Shared<'g, Record>,
131        new: Shared<'g, Record>,
132        guard: &'g Guard,
133    ) -> Result<Shared<'g, Record>, Shared<'g, Record>> {
134        self.next
135            .compare_exchange(current, new, Ordering::AcqRel, Ordering::Acquire, guard)
136            .map_err(|e| e.current)
137    }
138}
139
140unsafe impl Send for Record {}
141unsafe impl Sync for Record {}
142
143impl Record {
144    pub fn new(key: Vec<u8>, value: Vec<u8>, timestamp: u64) -> Self {
145        let key_len = key.len() as u16;
146        let value_len = value.len();
147        let value_bytes = Bytes::from(value);
148
149        Self {
150            key,
151            value: parking_lot::RwLock::new(Some(value_bytes)),
152            ttl_expiry: AtomicU64::new(0),
153            timestamp,
154            value_len,
155            sector: AtomicU64::new(0),
156            refcount: AtomicU32::new(1),
157            key_len,
158            hash_link: AtomicLink::new(),
159            cache_ref_bit: AtomicU32::new(0),
160            cache_access_time: AtomicU64::new(0),
161            retired_at: AtomicU64::new(0),
162            successor: OnceLock::new(),
163            value_source: None,
164            successor_safe: AtomicBool::new(false),
165            extent_state: AtomicU32::new(0),
166        }
167    }
168
169    pub fn new_with_timestamp(key: Vec<u8>, value: Vec<u8>, timestamp: u64) -> Self {
170        Self::new(key, value, timestamp)
171    }
172
173    pub fn new_with_timestamp_ttl(
174        key: Vec<u8>,
175        value: Vec<u8>,
176        timestamp: u64,
177        ttl_expiry: u64,
178    ) -> Self {
179        let record = Self::new(key, value, timestamp);
180        record.ttl_expiry.store(ttl_expiry, Ordering::Release);
181        record
182    }
183
184    /// Create a new record from a Bytes value (zero-copy)
185    pub fn new_from_bytes(key: Vec<u8>, value: Bytes, timestamp: u64) -> Self {
186        let key_len = key.len() as u16;
187        let value_len = value.len();
188
189        Self {
190            key,
191            value: parking_lot::RwLock::new(Some(value)),
192            ttl_expiry: AtomicU64::new(0),
193            timestamp,
194            value_len,
195            sector: AtomicU64::new(0),
196            refcount: AtomicU32::new(1),
197            key_len,
198            hash_link: AtomicLink::new(),
199            cache_ref_bit: AtomicU32::new(0),
200            cache_access_time: AtomicU64::new(0),
201            retired_at: AtomicU64::new(0),
202            successor: OnceLock::new(),
203            value_source: None,
204            successor_safe: AtomicBool::new(false),
205            extent_state: AtomicU32::new(0),
206        }
207    }
208
209    /// Create a new record from Bytes with TTL
210    pub fn new_from_bytes_with_ttl(
211        key: Vec<u8>,
212        value: Bytes,
213        timestamp: u64,
214        ttl_expiry: u64,
215    ) -> Self {
216        let record = Self::new_from_bytes(key, value, timestamp);
217        record.ttl_expiry.store(ttl_expiry, Ordering::Release);
218        record
219    }
220
221    pub(crate) fn new_deferred_with_ttl(
222        predecessor: &Arc<Record>,
223        timestamp: u64,
224        ttl_expiry: u64,
225    ) -> Self {
226        let key = predecessor.key.clone();
227        let key_len = key.len() as u16;
228        Self {
229            key,
230            value: parking_lot::RwLock::new(None),
231            ttl_expiry: AtomicU64::new(ttl_expiry),
232            timestamp,
233            value_len: predecessor.value_len,
234            sector: AtomicU64::new(0),
235            refcount: AtomicU32::new(1),
236            key_len,
237            hash_link: AtomicLink::new(),
238            cache_ref_bit: AtomicU32::new(0),
239            cache_access_time: AtomicU64::new(0),
240            retired_at: AtomicU64::new(0),
241            successor: OnceLock::new(),
242            value_source: Some(Arc::downgrade(predecessor)),
243            successor_safe: AtomicBool::new(false),
244            extent_state: AtomicU32::new(0),
245        }
246    }
247
248    pub fn calculate_size(&self) -> usize {
249        mem::size_of::<Self>() + self.key.capacity() + self.value_len
250    }
251
252    pub fn calculate_disk_size(&self) -> usize {
253        let record_size = SECTOR_HEADER_SIZE
254            + mem::size_of::<u16>()
255            + self.key.len()
256            + mem::size_of::<u64>()
257            + mem::size_of::<u64>()
258            + mem::size_of::<u64>()
259            + self.value_len;
260
261        record_size.div_ceil(FEOX_BLOCK_SIZE) * FEOX_BLOCK_SIZE
262    }
263
264    /// Get value - returns None if value has been offloaded to disk
265    #[inline]
266    pub fn get_value(&self) -> Option<Bytes> {
267        self.value.read().clone()
268    }
269
270    /// Clear value from memory
271    #[inline]
272    pub fn clear_value(&self) {
273        *self.value.write() = None;
274        std::sync::atomic::fence(Ordering::Release);
275    }
276
277    pub(crate) fn value_source(&self) -> Option<Arc<Record>> {
278        self.value_source.as_ref().and_then(Weak::upgrade)
279    }
280
281    pub fn inc_ref(&self) {
282        self.refcount.fetch_add(1, Ordering::AcqRel);
283    }
284
285    pub fn dec_ref(&self) -> u32 {
286        let old = self.refcount.fetch_sub(1, Ordering::AcqRel);
287        debug_assert!(old > 0, "Record refcount underflow");
288        old - 1
289    }
290
291    pub fn ref_count(&self) -> u32 {
292        self.refcount.load(Ordering::Acquire)
293    }
294
295    pub(crate) fn link_successor(&self, successor: &Arc<Record>) {
296        let result = self.successor.set(Arc::clone(successor));
297        debug_assert!(result.is_ok());
298    }
299
300    pub(crate) fn retirement_timestamp(&self) -> u64 {
301        let mut retired_at = self.retired_at.load(Ordering::Acquire);
302        let Some(mut current) = self.successor.get().cloned() else {
303            return retired_at;
304        };
305
306        loop {
307            retired_at = retired_at.max(current.retired_at.load(Ordering::Acquire));
308            let Some(successor) = current.successor.get().cloned() else {
309                return retired_at;
310            };
311            current = successor;
312        }
313    }
314
315    pub(crate) fn successor_is_durable_or_deleted(&self) -> bool {
316        if self.successor_safe.load(Ordering::Acquire) {
317            return true;
318        }
319
320        let Some(mut current) = self.successor.get().cloned() else {
321            return true;
322        };
323        let mut path = Vec::new();
324
325        loop {
326            if current.sector.load(Ordering::Acquire) > 0
327                || current.successor_safe.load(Ordering::Acquire)
328            {
329                break;
330            }
331
332            let Some(successor) = current.successor.get().cloned() else {
333                if current.refcount.load(Ordering::Acquire) != 0 {
334                    return false;
335                }
336                match current.successor.get().cloned() {
337                    Some(successor) => {
338                        path.push(current);
339                        current = successor;
340                        continue;
341                    }
342                    None => break,
343                }
344            };
345
346            path.push(current);
347            current = successor;
348        }
349
350        self.successor_safe.store(true, Ordering::Release);
351        for record in path {
352            record.successor_safe.store(true, Ordering::Release);
353        }
354        true
355    }
356
357    pub(crate) fn acquire_extent(&self) -> Option<ExtentReadGuard<'_>> {
358        let mut state = self.extent_state.load(Ordering::Acquire);
359        loop {
360            if state & EXTENT_RETIRED != 0 {
361                return None;
362            }
363            debug_assert!(state & EXTENT_READERS != EXTENT_READERS);
364            match self.extent_state.compare_exchange_weak(
365                state,
366                state + 1,
367                Ordering::AcqRel,
368                Ordering::Acquire,
369            ) {
370                Ok(_) => return Some(ExtentReadGuard(&self.extent_state)),
371                Err(current) => state = current,
372            }
373        }
374    }
375
376    pub(crate) fn retire_extent(&self) {
377        self.extent_state.fetch_or(EXTENT_RETIRED, Ordering::AcqRel);
378    }
379
380    pub(crate) fn extent_has_readers(&self) -> bool {
381        self.extent_state.load(Ordering::Acquire) & EXTENT_READERS != 0
382    }
383}
384
385impl Drop for Record {
386    fn drop(&mut self) {
387        let mut successor = self.successor.take();
388        while let Some(record) = successor {
389            match Arc::try_unwrap(record) {
390                Ok(mut record) => {
391                    successor = record.successor.take();
392                }
393                Err(record) => {
394                    drop(record);
395                    break;
396                }
397            }
398        }
399    }
400}
feoxdb/core/record.rs

feoxdb/core/
record.rs
