use ipfrs_core::Cid;
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PrefetchStrategy {
None,
ImmediateChildren,
PatternBased,
Subtree,
Adaptive,
}
#[derive(Debug, Clone)]
pub struct PrefetchConfig {
pub strategy: PrefetchStrategy,
pub max_depth: usize,
pub max_concurrent_prefetch: usize,
pub prefetch_buffer_size: usize,
pub min_confidence: f64,
pub pattern_history_size: usize,
pub adaptive_tuning: bool,
pub prefetch_timeout: Duration,
}
impl Default for PrefetchConfig {
fn default() -> Self {
Self {
strategy: PrefetchStrategy::PatternBased,
max_depth: 2,
max_concurrent_prefetch: 16,
prefetch_buffer_size: 128,
min_confidence: 0.6,
pattern_history_size: 1000,
adaptive_tuning: true,
prefetch_timeout: Duration::from_secs(5),
}
}
}
#[derive(Debug, Clone)]
struct AccessPattern {
#[allow(dead_code)]
source: Cid,
target: Cid,
count: usize,
last_access: Instant,
}
#[derive(Debug, Clone)]
struct DagLink {
#[allow(dead_code)]
parent: Cid,
children: Vec<Cid>,
depth: usize,
}
#[derive(Debug, Clone)]
pub struct Prediction {
pub cid: Cid,
pub confidence: f64,
pub depth: usize,
pub reason: PredictionReason,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PredictionReason {
DagChild,
AccessPattern,
Sibling,
Temporal,
}
#[derive(Debug, Clone, Default)]
pub struct PrefetchStats {
pub prefetch_requests: u64,
pub hits: u64,
pub misses: u64,
pub wasted_bytes: u64,
pub saved_latency_ms: u64,
pub hit_rate: f64,
}
impl PrefetchStats {
fn update_hit_rate(&mut self) {
let total = self.hits + self.misses;
if total > 0 {
self.hit_rate = self.hits as f64 / total as f64;
}
}
}
pub struct PrefetchPredictor {
config: PrefetchConfig,
patterns: Arc<RwLock<HashMap<Cid, Vec<AccessPattern>>>>,
dag_links: Arc<RwLock<HashMap<Cid, DagLink>>>,
access_history: Arc<RwLock<VecDeque<(Cid, Instant)>>>,
prefetched: Arc<RwLock<HashMap<Cid, Instant>>>,
stats: Arc<RwLock<PrefetchStats>>,
}
impl PrefetchPredictor {
pub fn new(config: PrefetchConfig) -> Self {
Self {
config,
patterns: Arc::new(RwLock::new(HashMap::new())),
dag_links: Arc::new(RwLock::new(HashMap::new())),
access_history: Arc::new(RwLock::new(VecDeque::new())),
prefetched: Arc::new(RwLock::new(HashMap::new())),
stats: Arc::new(RwLock::new(PrefetchStats::default())),
}
}
pub fn record_access(&self, cid: &Cid) {
let now = Instant::now();
{
let mut prefetched = self.prefetched.write().unwrap_or_else(|e| e.into_inner());
if let Some(prefetch_time) = prefetched.remove(cid) {
let mut stats = self.stats.write().unwrap_or_else(|e| e.into_inner());
stats.hits += 1;
let saved_ms = now.duration_since(prefetch_time).as_millis() as u64;
stats.saved_latency_ms += saved_ms;
stats.update_hit_rate();
}
}
{
let mut history = self
.access_history
.write()
.unwrap_or_else(|e| e.into_inner());
history.push_back((*cid, now));
while history.len() > self.config.pattern_history_size {
history.pop_front();
}
}
self.update_patterns(cid, now);
}
fn update_patterns(&self, current: &Cid, now: Instant) {
let history = self
.access_history
.read()
.unwrap_or_else(|e| e.into_inner());
let mut patterns = self.patterns.write().unwrap_or_else(|e| e.into_inner());
let recent_window = Duration::from_secs(1);
for (prev_cid, prev_time) in history.iter().rev() {
if now.duration_since(*prev_time) > recent_window {
break;
}
if prev_cid == current {
continue;
}
let pattern_list = patterns.entry(*prev_cid).or_default();
if let Some(pattern) = pattern_list.iter_mut().find(|p| p.target == *current) {
pattern.count += 1;
pattern.last_access = now;
} else {
pattern_list.push(AccessPattern {
source: *prev_cid,
target: *current,
count: 1,
last_access: now,
});
}
}
}
pub fn record_dag_links(&self, parent: &Cid, children: Vec<Cid>, depth: usize) {
let mut dag_links = self.dag_links.write().unwrap_or_else(|e| e.into_inner());
dag_links.insert(
*parent,
DagLink {
parent: *parent,
children,
depth,
},
);
}
pub fn predict(&self, current: &Cid) -> Vec<Prediction> {
match self.config.strategy {
PrefetchStrategy::None => Vec::new(),
PrefetchStrategy::ImmediateChildren => self.predict_dag_children(current),
PrefetchStrategy::PatternBased => self.predict_from_patterns(current),
PrefetchStrategy::Subtree => self.predict_subtree(current),
PrefetchStrategy::Adaptive => self.predict_adaptive(current),
}
}
fn predict_dag_children(&self, current: &Cid) -> Vec<Prediction> {
let dag_links = self.dag_links.read().unwrap_or_else(|e| e.into_inner());
if let Some(link) = dag_links.get(current) {
link.children
.iter()
.map(|child| Prediction {
cid: *child,
confidence: 0.95,
depth: link.depth + 1,
reason: PredictionReason::DagChild,
})
.collect()
} else {
Vec::new()
}
}
fn predict_from_patterns(&self, current: &Cid) -> Vec<Prediction> {
let patterns = self.patterns.read().unwrap_or_else(|e| e.into_inner());
if let Some(pattern_list) = patterns.get(current) {
let total_count: usize = pattern_list.iter().map(|p| p.count).sum();
pattern_list
.iter()
.filter_map(|pattern| {
let confidence = pattern.count as f64 / total_count as f64;
if confidence >= self.config.min_confidence {
Some(Prediction {
cid: pattern.target,
confidence,
depth: 1,
reason: PredictionReason::AccessPattern,
})
} else {
None
}
})
.collect()
} else {
self.predict_dag_children(current)
}
}
fn predict_subtree(&self, current: &Cid) -> Vec<Prediction> {
let mut predictions = Vec::new();
let mut visited = HashSet::new();
let mut queue = VecDeque::new();
queue.push_back((*current, 0));
visited.insert(*current);
let dag_links = self.dag_links.read().unwrap_or_else(|e| e.into_inner());
while let Some((cid, depth)) = queue.pop_front() {
if depth >= self.config.max_depth {
continue;
}
if let Some(link) = dag_links.get(&cid) {
for child in &link.children {
if visited.insert(*child) {
predictions.push(Prediction {
cid: *child,
confidence: 0.9 * (0.8_f64).powi(depth as i32),
depth: depth + 1,
reason: PredictionReason::DagChild,
});
queue.push_back((*child, depth + 1));
}
}
}
}
predictions
}
fn predict_adaptive(&self, current: &Cid) -> Vec<Prediction> {
let stats = self.stats.read().unwrap_or_else(|e| e.into_inner());
let hit_rate = stats.hit_rate;
drop(stats);
if hit_rate > 0.5 {
self.predict_from_patterns(current)
} else {
self.predict_dag_children(current)
}
}
pub fn record_prefetch(&self, cid: &Cid) {
let mut prefetched = self.prefetched.write().unwrap_or_else(|e| e.into_inner());
prefetched.insert(*cid, Instant::now());
let mut stats = self.stats.write().unwrap_or_else(|e| e.into_inner());
stats.prefetch_requests += 1;
}
pub fn record_miss(&self, cid: &Cid, bytes: u64) {
let mut prefetched = self.prefetched.write().unwrap_or_else(|e| e.into_inner());
prefetched.remove(cid);
let mut stats = self.stats.write().unwrap_or_else(|e| e.into_inner());
stats.misses += 1;
stats.wasted_bytes += bytes;
stats.update_hit_rate();
}
pub fn cleanup(&self, max_age: Duration) {
let now = Instant::now();
{
let mut prefetched = self.prefetched.write().unwrap_or_else(|e| e.into_inner());
let mut to_remove = Vec::new();
let mut total_missed = 0u64;
for (cid, time) in prefetched.iter() {
if now.duration_since(*time) >= max_age {
to_remove.push(*cid);
total_missed += 1;
}
}
for cid in to_remove {
prefetched.remove(&cid);
}
if total_missed > 0 {
let mut stats = self.stats.write().unwrap_or_else(|e| e.into_inner());
stats.misses += total_missed;
stats.update_hit_rate();
}
}
{
let mut patterns = self.patterns.write().unwrap_or_else(|e| e.into_inner());
let max_pattern_age = Duration::from_secs(300);
for pattern_list in patterns.values_mut() {
pattern_list.retain(|p| now.duration_since(p.last_access) < max_pattern_age);
}
patterns.retain(|_, v| !v.is_empty());
}
}
pub fn stats(&self) -> PrefetchStats {
self.stats.read().unwrap_or_else(|e| e.into_inner()).clone()
}
pub fn update_config(&mut self, config: PrefetchConfig) {
self.config = config;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_prefetch_predictor_creation() {
let config = PrefetchConfig::default();
let _predictor = PrefetchPredictor::new(config);
}
#[test]
fn test_record_access() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let cid = Cid::default();
predictor.record_access(&cid);
let history = predictor
.access_history
.read()
.unwrap_or_else(|e| e.into_inner());
assert_eq!(history.len(), 1);
}
#[test]
fn test_dag_children_prediction() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let parent = Cid::default();
let child1 = Cid::default();
let child2 = Cid::default();
predictor.record_dag_links(&parent, vec![child1, child2], 0);
let predictions = predictor.predict_dag_children(&parent);
assert_eq!(predictions.len(), 2);
}
#[test]
fn test_pattern_based_prediction() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
min_confidence: 0.5,
..Default::default()
});
let cid1 = Cid::default();
let cid2 = Cid::default();
predictor.record_access(&cid1);
std::thread::sleep(Duration::from_millis(10));
predictor.record_access(&cid2);
std::thread::sleep(Duration::from_millis(10));
predictor.record_access(&cid1);
std::thread::sleep(Duration::from_millis(10));
predictor.record_access(&cid2);
let predictions = predictor.predict_from_patterns(&cid1);
assert!(predictions.is_empty());
}
#[test]
fn test_prefetch_stats() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let cid = Cid::default();
predictor.record_prefetch(&cid);
predictor.record_access(&cid);
let stats = predictor.stats();
assert_eq!(stats.hits, 1);
assert_eq!(stats.prefetch_requests, 1);
}
#[test]
fn test_subtree_prediction() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
max_depth: 3,
..Default::default()
});
let root = Cid::default();
let child1 = Cid::default();
let child2 = Cid::default();
let grandchild1 = Cid::default();
predictor.record_dag_links(&root, vec![child1, child2], 0);
predictor.record_dag_links(&child1, vec![grandchild1], 1);
let _predictions = predictor.predict_subtree(&root);
}
#[test]
fn test_adaptive_prediction_switches_strategy() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
strategy: PrefetchStrategy::Adaptive,
..Default::default()
});
let cid = Cid::default();
let child = Cid::default();
predictor.record_dag_links(&cid, vec![child], 0);
let predictions = predictor.predict_adaptive(&cid);
assert!(!predictions.is_empty());
}
#[test]
fn test_prefetch_miss_tracking() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let cid = Cid::default();
predictor.record_prefetch(&cid);
predictor.record_miss(&cid, 1024);
let stats = predictor.stats();
assert_eq!(stats.misses, 1);
assert_eq!(stats.wasted_bytes, 1024);
}
#[test]
fn test_hit_rate_calculation() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let cid1 = Cid::default();
let cid2 = Cid::default();
let cid3 = Cid::default();
predictor.record_prefetch(&cid1);
predictor.record_access(&cid1);
predictor.record_prefetch(&cid2);
predictor.record_access(&cid2);
predictor.record_prefetch(&cid3);
predictor.record_miss(&cid3, 100);
let stats = predictor.stats();
assert_eq!(stats.hits, 2);
assert_eq!(stats.misses, 1);
assert!((stats.hit_rate - 0.666).abs() < 0.01);
}
#[test]
fn test_cleanup_old_prefetches() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let cid = Cid::default();
predictor.record_prefetch(&cid);
predictor.cleanup(Duration::from_secs(0));
let stats = predictor.stats();
assert_eq!(stats.misses, 1); }
#[test]
fn test_multiple_predictions_sorted_by_confidence() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
min_confidence: 0.3,
..Default::default()
});
let cid1 = Cid::default();
let cid2 = Cid::default();
let cid3 = Cid::default();
for _ in 0..3 {
predictor.record_access(&cid1);
std::thread::sleep(Duration::from_millis(10));
predictor.record_access(&cid2);
std::thread::sleep(Duration::from_millis(10));
}
predictor.record_access(&cid1);
std::thread::sleep(Duration::from_millis(10));
predictor.record_access(&cid3);
let predictions = predictor.predict_from_patterns(&cid1);
if !predictions.is_empty() {
let cid2_pred = predictions.iter().find(|p| p.cid == cid2);
let cid3_pred = predictions.iter().find(|p| p.cid == cid3);
if let (Some(p2), Some(p3)) = (cid2_pred, cid3_pred) {
assert!(p2.confidence > p3.confidence);
}
}
}
#[test]
fn test_no_predictions_for_unknown_cid() {
let predictor = PrefetchPredictor::new(PrefetchConfig::default());
let unknown_cid = Cid::default();
let predictions = predictor.predict_dag_children(&unknown_cid);
assert!(predictions.is_empty());
}
#[test]
fn test_prediction_confidence_thresholds() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
min_confidence: 0.8, ..Default::default()
});
let cid1 = Cid::default();
let cid2 = Cid::default();
predictor.record_access(&cid1);
std::thread::sleep(Duration::from_millis(10));
predictor.record_access(&cid2);
let predictions = predictor.predict_from_patterns(&cid1);
assert!(predictions.is_empty() || predictions[0].confidence >= 0.8);
}
#[test]
fn test_prefetch_strategy_none() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
strategy: PrefetchStrategy::None,
..Default::default()
});
let cid = Cid::default();
let predictions = predictor.predict(&cid);
assert!(predictions.is_empty());
}
#[test]
fn test_depth_limited_subtree() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
max_depth: 1,
..Default::default()
});
let root = Cid::default();
let child = Cid::default();
let grandchild = Cid::default();
predictor.record_dag_links(&root, vec![child], 0);
predictor.record_dag_links(&child, vec![grandchild], 1);
let predictions = predictor.predict_subtree(&root);
assert!(predictions.len() <= 1);
}
#[test]
fn test_access_history_limit() {
let predictor = PrefetchPredictor::new(PrefetchConfig {
pattern_history_size: 5,
..Default::default()
});
for _ in 0..10 {
let cid = Cid::default();
predictor.record_access(&cid);
}
let history = predictor
.access_history
.read()
.unwrap_or_else(|e| e.into_inner());
assert!(history.len() <= 5);
}
#[test]
fn test_update_config() {
let mut predictor = PrefetchPredictor::new(PrefetchConfig::default());
let new_config = PrefetchConfig {
strategy: PrefetchStrategy::Subtree,
max_depth: 5,
..Default::default()
};
predictor.update_config(new_config.clone());
assert_eq!(predictor.config.strategy, PrefetchStrategy::Subtree);
assert_eq!(predictor.config.max_depth, 5);
}
}