sz_orm_core/
cache_coherence.rs

1//! 缓存一致性协议模块（v4.1.0，`cache-coherence` feature gate）
2//!
3//! 实现 MESI 风格缓存一致性状态机，支持 WriteThrough/WriteBehind 策略，
4//! 通过 trait-based 广播器实现跨实例失效广播。
5
6use std::collections::HashMap;
7use std::sync::{Arc, RwLock};
8use std::time::{SystemTime, UNIX_EPOCH};
9
10/// MESI 缓存行状态
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum MesiState {
13    /// 已修改（本地修改，其他实例无副本）
14    Modified,
15    /// 独占（本地独有，未修改，其他实例无副本）
16    Exclusive,
17    /// 共享（多实例共享只读副本）
18    Shared,
19    /// 无效（缓存行不存在或已失效）
20    Invalid,
21}
22
23/// 写策略
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum ConsistencyStrategy {
26    /// 写穿透（同步写缓存+DB）
27    WriteThrough,
28    /// 写后行（异步写 DB，先写缓存）
29    WriteBehind,
30}
31
32/// 失效操作类型
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub enum InvalidationOp {
35    /// 修改操作
36    Modify,
37    /// 删除操作
38    Delete,
39}
40
41/// 失效广播事件
42#[derive(Debug, Clone)]
43pub struct InvalidationEvent {
44    /// 缓存键
45    pub key: String,
46    /// 发送实例 ID
47    pub instance_id: String,
48    /// 时间戳（Unix 毫秒）
49    pub timestamp: u64,
50    /// 操作类型
51    pub op: InvalidationOp,
52}
53
54/// 失效广播器 trait（抽象消息队列，避免硬依赖具体 MQ 实现）
55pub trait InvalidationBroadcaster: Send + Sync {
56    /// 广播失效事件
57    fn broadcast(&self, event: &InvalidationEvent) -> Result<(), CoherenceError>;
58}
59
60/// 一致性指标
61#[derive(Debug, Clone, Default)]
62pub struct CoherenceMetrics {
63    /// Modified 状态计数
64    pub modified_count: u64,
65    /// Exclusive 状态计数
66    pub exclusive_count: u64,
67    /// Shared 状态计数
68    pub shared_count: u64,
69    /// Invalid 状态计数
70    pub invalid_count: u64,
71    /// 失效广播次数
72    pub invalidation_broadcasts: u64,
73    /// 一致性违规次数
74    pub coherence_violations: u64,
75    /// Write-behind 回滚次数
76    pub write_behind_rollbacks: u64,
77}
78
79/// 一致性错误
80#[derive(Debug, Clone, thiserror::Error)]
81pub enum CoherenceError {
82    /// 广播失败
83    #[error("broadcast failed: {0}")]
84    BroadcastFailed(String),
85    /// Write-behind 失败
86    #[error("write-behind failed for key: {key}")]
87    WriteBehindFailed {
88        /// 失败的缓存键
89        key: String,
90    },
91    /// 脑裂检测
92    #[error("split-brain detected for key: {key}")]
93    SplitBrain {
94        /// 发生脑裂的缓存键
95        key: String,
96    },
97    /// 缓存未命中
98    #[error("cache miss for key: {0}")]
99    CacheMiss(String),
100}
101
102/// 缓存一致性协议
103pub struct CacheCoherenceProtocol {
104    /// 缓存行状态表（key → MESI 状态）
105    states: RwLock<HashMap<String, MesiState>>,
106    /// 广播器
107    broadcaster: Arc<dyn InvalidationBroadcaster>,
108    /// 实例 ID
109    instance_id: String,
110    /// 写策略
111    strategy: ConsistencyStrategy,
112    /// 指标
113    metrics: Arc<RwLock<CoherenceMetrics>>,
114}
115
116impl CacheCoherenceProtocol {
117    /// 创建新的一致性协议实例
118    pub fn new(
119        instance_id: String,
120        strategy: ConsistencyStrategy,
121        broadcaster: Arc<dyn InvalidationBroadcaster>,
122    ) -> Self {
123        Self {
124            states: RwLock::new(HashMap::new()),
125            broadcaster,
126            instance_id,
127            strategy,
128            metrics: Arc::new(RwLock::new(CoherenceMetrics::default())),
129        }
130    }
131
132    /// 获取缓存行状态
133    pub fn state(&self, key: &str) -> MesiState {
134        self.states
135            .read()
136            .unwrap()
137            .get(key)
138            .copied()
139            .unwrap_or(MesiState::Invalid)
140    }
141
142    /// 读操作（触发状态转换：Invalid→Exclusive/Shared）
143    pub fn read(&self, key: &str, other_instances_have: bool) -> MesiState {
144        let mut states = self.states.write().unwrap();
145        let mut metrics = self.metrics.write().unwrap();
146        let current = states.get(key).copied().unwrap_or(MesiState::Invalid);
147        let new_state = match current {
148            MesiState::Invalid => {
149                if other_instances_have {
150                    MesiState::Shared
151                } else {
152                    MesiState::Exclusive
153                }
154            }
155            other => other,
156        };
157        states.insert(key.to_string(), new_state);
158        Self::update_metrics(&mut metrics, &new_state);
159        new_state
160    }
161
162    /// 写操作（触发状态转换：→Modified，广播失效）
163    pub fn write(&self, key: &str) -> Result<MesiState, CoherenceError> {
164        let event = InvalidationEvent {
165            key: key.to_string(),
166            instance_id: self.instance_id.clone(),
167            timestamp: SystemTime::now()
168                .duration_since(UNIX_EPOCH)
169                .unwrap_or_default()
170                .as_millis() as u64,
171            op: InvalidationOp::Modify,
172        };
173        self.broadcaster.broadcast(&event)?;
174
175        let mut states = self.states.write().unwrap();
176        let mut metrics = self.metrics.write().unwrap();
177        states.insert(key.to_string(), MesiState::Modified);
178        metrics.invalidation_broadcasts += 1;
179        Self::update_metrics(&mut metrics, &MesiState::Modified);
180        Ok(MesiState::Modified)
181    }
182
183    /// 处理收到的失效广播（→Invalid）
184    pub fn handle_invalidation(&self, event: &InvalidationEvent) {
185        if event.instance_id == self.instance_id {
186            return;
187        }
188        let mut states = self.states.write().unwrap();
189        let mut metrics = self.metrics.write().unwrap();
190        states.insert(event.key.clone(), MesiState::Invalid);
191        metrics.invalid_count += 1;
192    }
193
194    /// 获取指标快照
195    pub fn metrics(&self) -> CoherenceMetrics {
196        self.metrics.read().unwrap().clone()
197    }
198
199    /// 获取写策略
200    pub fn strategy(&self) -> ConsistencyStrategy {
201        self.strategy
202    }
203
204    fn update_metrics(metrics: &mut CoherenceMetrics, state: &MesiState) {
205        match state {
206            MesiState::Modified => metrics.modified_count += 1,
207            MesiState::Exclusive => metrics.exclusive_count += 1,
208            MesiState::Shared => metrics.shared_count += 1,
209            MesiState::Invalid => metrics.invalid_count += 1,
210        }
211    }
212}
213
214/// 空广播器（单实例模式，不广播）
215pub struct NoopBroadcaster;
216
217impl InvalidationBroadcaster for NoopBroadcaster {
218    fn broadcast(&self, _event: &InvalidationEvent) -> Result<(), CoherenceError> {
219        Ok(())
220    }
221}
222
223/// 本地广播器（收集事件用于测试）
224pub struct LocalBroadcaster {
225    events: RwLock<Vec<InvalidationEvent>>,
226}
227
228impl LocalBroadcaster {
229    /// 创建本地广播器
230    pub fn new() -> Self {
231        Self {
232            events: RwLock::new(Vec::new()),
233        }
234    }
235
236    /// 获取已收集的事件
237    pub fn events(&self) -> Vec<InvalidationEvent> {
238        self.events.read().unwrap().clone()
239    }
240}
241
242impl Default for LocalBroadcaster {
243    fn default() -> Self {
244        Self::new()
245    }
246}
247
248impl InvalidationBroadcaster for LocalBroadcaster {
249    fn broadcast(&self, event: &InvalidationEvent) -> Result<(), CoherenceError> {
250        self.events.write().unwrap().push(event.clone());
251        Ok(())
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    #[test]
260    fn test_mesi_state_transitions() {
261        let broadcaster = Arc::new(LocalBroadcaster::new());
262        let protocol = CacheCoherenceProtocol::new(
263            "instance-A".to_string(),
264            ConsistencyStrategy::WriteThrough,
265            broadcaster,
266        );
267
268        assert_eq!(protocol.state("key1"), MesiState::Invalid);
269
270        let s = protocol.read("key1", false);
271        assert_eq!(s, MesiState::Exclusive);
272
273        let s = protocol.read("key1", true);
274        assert_eq!(s, MesiState::Exclusive);
275
276        let s = protocol.write("key1").unwrap();
277        assert_eq!(s, MesiState::Modified);
278        assert_eq!(protocol.state("key1"), MesiState::Modified);
279    }
280
281    #[test]
282    fn test_invalid_to_shared() {
283        let broadcaster = Arc::new(LocalBroadcaster::new());
284        let protocol = CacheCoherenceProtocol::new(
285            "instance-A".to_string(),
286            ConsistencyStrategy::WriteThrough,
287            broadcaster,
288        );
289
290        let s = protocol.read("key1", true);
291        assert_eq!(s, MesiState::Shared);
292    }
293
294    #[test]
295    fn test_invalid_to_exclusive() {
296        let broadcaster = Arc::new(LocalBroadcaster::new());
297        let protocol = CacheCoherenceProtocol::new(
298            "instance-A".to_string(),
299            ConsistencyStrategy::WriteThrough,
300            broadcaster,
301        );
302
303        let s = protocol.read("key1", false);
304        assert_eq!(s, MesiState::Exclusive);
305    }
306
307    #[test]
308    fn test_write_broadcasts_invalidation() {
309        let broadcaster = Arc::new(LocalBroadcaster::new());
310        let protocol = CacheCoherenceProtocol::new(
311            "instance-A".to_string(),
312            ConsistencyStrategy::WriteThrough,
313            broadcaster.clone(),
314        );
315
316        protocol.write("key1").unwrap();
317        let events = broadcaster.events();
318        assert_eq!(events.len(), 1);
319        assert_eq!(events[0].key, "key1");
320        assert_eq!(events[0].op, InvalidationOp::Modify);
321    }
322
323    #[test]
324    fn test_handle_invalidation_sets_invalid() {
325        let broadcaster = Arc::new(LocalBroadcaster::new());
326        let protocol = CacheCoherenceProtocol::new(
327            "instance-A".to_string(),
328            ConsistencyStrategy::WriteThrough,
329            broadcaster,
330        );
331
332        protocol.read("key1", false);
333        assert_eq!(protocol.state("key1"), MesiState::Exclusive);
334
335        let event = InvalidationEvent {
336            key: "key1".to_string(),
337            instance_id: "instance-B".to_string(),
338            timestamp: 0,
339            op: InvalidationOp::Modify,
340        };
341        protocol.handle_invalidation(&event);
342        assert_eq!(protocol.state("key1"), MesiState::Invalid);
343    }
344
345    #[test]
346    fn test_ignore_self_invalidation() {
347        let broadcaster = Arc::new(LocalBroadcaster::new());
348        let protocol = CacheCoherenceProtocol::new(
349            "instance-A".to_string(),
350            ConsistencyStrategy::WriteThrough,
351            broadcaster,
352        );
353
354        protocol.read("key1", false);
355        assert_eq!(protocol.state("key1"), MesiState::Exclusive);
356
357        let event = InvalidationEvent {
358            key: "key1".to_string(),
359            instance_id: "instance-A".to_string(),
360            timestamp: 0,
361            op: InvalidationOp::Modify,
362        };
363        protocol.handle_invalidation(&event);
364        assert_eq!(protocol.state("key1"), MesiState::Exclusive);
365    }
366
367    #[test]
368    fn test_metrics_tracking() {
369        let broadcaster = Arc::new(LocalBroadcaster::new());
370        let protocol = CacheCoherenceProtocol::new(
371            "instance-A".to_string(),
372            ConsistencyStrategy::WriteThrough,
373            broadcaster,
374        );
375
376        protocol.read("key1", false);
377        protocol.read("key2", true);
378        protocol.write("key1").unwrap();
379
380        let metrics = protocol.metrics();
381        assert!(metrics.exclusive_count > 0);
382        assert!(metrics.shared_count > 0);
383        assert!(metrics.modified_count > 0);
384        assert!(metrics.invalidation_broadcasts > 0);
385    }
386
387    #[test]
388    fn test_noop_broadcaster() {
389        let broadcaster = Arc::new(NoopBroadcaster);
390        let protocol = CacheCoherenceProtocol::new(
391            "instance-A".to_string(),
392            ConsistencyStrategy::WriteBehind,
393            broadcaster,
394        );
395
396        let result = protocol.write("key1");
397        assert!(result.is_ok());
398    }
399
400    #[test]
401    fn test_shared_to_modified_on_write() {
402        let broadcaster = Arc::new(LocalBroadcaster::new());
403        let protocol = CacheCoherenceProtocol::new(
404            "instance-A".to_string(),
405            ConsistencyStrategy::WriteThrough,
406            broadcaster,
407        );
408
409        let s = protocol.read("key1", true);
410        assert_eq!(s, MesiState::Shared);
411
412        let s = protocol.write("key1").unwrap();
413        assert_eq!(s, MesiState::Modified);
414    }
415
416    #[test]
417    fn test_strategy_access() {
418        let broadcaster = Arc::new(NoopBroadcaster);
419        let protocol = CacheCoherenceProtocol::new(
420            "instance-A".to_string(),
421            ConsistencyStrategy::WriteBehind,
422            broadcaster,
423        );
424        assert_eq!(protocol.strategy(), ConsistencyStrategy::WriteBehind);
425    }
426}
sz_orm_core/cache_coherence.rs

sz_orm_core/
cache_coherence.rs
