tycho_core/storage/block_handle/
meta.rs

1use std::sync::atomic::{AtomicU64, Ordering};
2
3use bytes::Buf;
4use tycho_storage::kv::{StoredValue, StoredValueBuffer};
5
6#[derive(Debug, Copy, Clone)]
7pub struct NewBlockMeta {
8    pub is_key_block: bool,
9    pub gen_utime: u32,
10    pub ref_by_mc_seqno: u32,
11}
12
13impl NewBlockMeta {
14    pub fn zero_state(gen_utime: u32, is_key_block: bool) -> Self {
15        Self {
16            is_key_block,
17            gen_utime,
18            ref_by_mc_seqno: 0,
19        }
20    }
21}
22
23#[derive(Debug, Default)]
24pub struct BlockMeta {
25    flags: AtomicU64,
26    gen_utime: u32,
27}
28
29#[derive(Debug, Clone, Copy)]
30pub struct LoadedBlockMeta {
31    pub flags: BlockFlags,
32    pub mc_ref_seqno: u32,
33    pub gen_utime: u32,
34}
35
36impl BlockMeta {
37    pub fn with_data(data: NewBlockMeta) -> Self {
38        const IS_KEY_BLOCK_MASK: u64 =
39            (BlockFlags::IS_KEY_BLOCK.bits() as u64) << BLOCK_FLAGS_OFFSET;
40
41        Self {
42            flags: AtomicU64::new(
43                if data.is_key_block {
44                    IS_KEY_BLOCK_MASK
45                } else {
46                    0
47                } | data.ref_by_mc_seqno as u64,
48            ),
49            gen_utime: data.gen_utime,
50        }
51    }
52
53    pub fn load(&self) -> LoadedBlockMeta {
54        let flags = self.flags.load(Ordering::Acquire);
55        LoadedBlockMeta {
56            flags: BlockFlags::from_bits_retain((flags >> BLOCK_FLAGS_OFFSET) as u32),
57            mc_ref_seqno: flags as u32,
58            gen_utime: self.gen_utime,
59        }
60    }
61
62    pub fn flags(&self) -> BlockFlags {
63        let flags = self.flags.load(Ordering::Acquire) >> BLOCK_FLAGS_OFFSET;
64        BlockFlags::from_bits_retain(flags as u32)
65    }
66
67    pub fn ref_by_mc_seqno(&self) -> u32 {
68        self.flags.load(Ordering::Acquire) as u32
69    }
70
71    pub fn gen_utime(&self) -> u32 {
72        self.gen_utime
73    }
74
75    pub(crate) fn add_flags(&self, flags: BlockFlags) -> bool {
76        let flags = (flags.bits() as u64) << BLOCK_FLAGS_OFFSET;
77        self.flags.fetch_or(flags, Ordering::Release) & flags != flags
78    }
79}
80
81impl StoredValue for BlockMeta {
82    /// 8 bytes flags
83    /// 4 bytes `gen_utime`
84    const SIZE_HINT: usize = 8 + 4;
85
86    type OnStackSlice = [u8; Self::SIZE_HINT];
87
88    fn serialize<T: StoredValueBuffer>(&self, buffer: &mut T) {
89        let flags = self.flags.load(Ordering::Acquire);
90        buffer.write_raw_slice(&flags.to_le_bytes());
91        buffer.write_raw_slice(&self.gen_utime.to_le_bytes());
92    }
93
94    fn deserialize(reader: &mut &[u8]) -> Self
95    where
96        Self: Sized,
97    {
98        let flags = reader.get_u64_le();
99        let gen_utime = reader.get_u32_le();
100
101        Self {
102            flags: AtomicU64::new(flags),
103            gen_utime,
104        }
105    }
106}
107
108bitflags::bitflags! {
109    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
110    pub struct BlockFlags: u32 {
111        // Persistent flags
112        const HAS_DATA = 1 << 0;
113        const HAS_PROOF = 1 << 1;
114        const HAS_QUEUE_DIFF = 1 << 2;
115
116        const HAS_STATE = 1 << 3;
117        const HAS_PERSISTENT_SHARD_STATE = 1 << 4;
118        const HAS_PERSISTENT_QUEUE_STATE = 1 << 5;
119
120        const HAS_NEXT_1 = 1 << 6;
121        const HAS_NEXT_2 = 1 << 7;
122        const HAS_PREV_1 = 1 << 8;
123        const HAS_PREV_2 = 1 << 9;
124        const IS_COMMITTED = 1 << 10;
125        const IS_KEY_BLOCK = 1 << 11;
126        const IS_PERSISTENT = 1 << 12;
127
128        const IS_REMOVED = 1 << 15;
129
130        // Composite flags
131        const HAS_ALL_BLOCK_PARTS =
132            Self::HAS_DATA.bits() | Self::HAS_PROOF.bits() | Self::HAS_QUEUE_DIFF.bits();
133    }
134}
135
136const BLOCK_FLAGS_OFFSET: usize = 32;
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn meta_store_load() {
144        let meta = BlockMeta::with_data(NewBlockMeta {
145            is_key_block: true,
146            gen_utime: 123456789,
147            ref_by_mc_seqno: 4311231,
148        });
149        assert_eq!(meta.flags(), BlockFlags::IS_KEY_BLOCK);
150        assert_eq!(meta.ref_by_mc_seqno(), 4311231);
151        assert_eq!(meta.gen_utime(), 123456789);
152
153        let stored = meta.to_vec();
154        assert_eq!(stored.len(), BlockMeta::SIZE_HINT);
155
156        let loaded = BlockMeta::from_slice(&stored);
157        assert_eq!(loaded.flags(), BlockFlags::IS_KEY_BLOCK);
158        assert_eq!(loaded.ref_by_mc_seqno(), 4311231);
159        assert_eq!(loaded.gen_utime(), 123456789);
160
161        let updated = meta.add_flags(BlockFlags::HAS_ALL_BLOCK_PARTS);
162        assert!(updated);
163        assert_eq!(
164            meta.flags(),
165            BlockFlags::HAS_ALL_BLOCK_PARTS | BlockFlags::IS_KEY_BLOCK
166        );
167
168        meta.add_flags(BlockFlags::IS_REMOVED);
169        assert_eq!(
170            meta.flags(),
171            BlockFlags::IS_KEY_BLOCK | BlockFlags::HAS_ALL_BLOCK_PARTS | BlockFlags::IS_REMOVED
172        );
173    }
174}
tycho_core/storage/block_handle/meta.rs

tycho_core/storage/block_handle/
meta.rs
