use anyhow::{Context, Result, anyhow};
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex, atomic::Ordering, RwLock};
use std::fs::File;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use std::time::{Duration, Instant};
use std::sync::atomic::Ordering;
use std::sync::atomic::AtomicU64;
use std::path::PathBuf;
use std::hash::Hasher;
Arc, RwLock, Mutex, atomic::{AtomicU64, Ordering},
};
#[derive(Debug)]
struct HighPerfCacheEntry {
wasm_bytes: Arc<Vec<u8>>,
precompiled_module: Option<wasmtime::Module>,
cached_at: Instant,
last_access: Instant,
access_count: AtomicU64,
file_path: Option<PathBuf>,
size: usize,
hash: String,
}
impl HighPerfCacheEntry {
fn new(wasm_bytes: Vec<u8>, hash: String) -> Self {
let now: _ = Instant::now();
let size: _ = wasm_bytes.len();
HighPerfCacheEntry {
wasm_bytes: Arc::new(Mutex::new(wasm_bytes)))
precompiled_module: None,
cached_at: now,
last_access: now,
access_count: AtomicU64::new(0),
file_path: None,
size,
hash,
}
}
fn update_access(_entry: &Arc<Self>) {
}
fn access_count(&self) -> u64 {
self.access_count.load(Ordering::Relaxed)
}
fn usage_score(&self) -> f64 {
let age_secs: _ = self.cached_at.elapsed().as_secs_f64();
if age_secs == 0.0 {
self.access_count() as f64
} else {
self.access_count() as f64 / age_secs
}
}
}
#[derive(Debug, Clone)]
pub struct HighPerfCacheConfig {
pub max_l1_entries: usize,
pub max_l1_size: usize,
pub l2_cache_dir: PathBuf,
pub enable_l2: bool,
pub enable_precompile: bool,
pub batch_size: usize,
pub async_io_concurrency: usize,
}
impl Default for HighPerfCacheConfig {
fn default() -> Self {
Self {
max_l1_entries: 5000, max_l1_size: 500 * 1024 * 1024, l2_cache_dir: PathBuf::from("benchmarks/fixtures/wasm_cache_high_perf"),
enable_l2: true,
enable_precompile: true,
batch_size: 100,
async_io_concurrency: 16,
}
}
}
#[derive(Debug, Default)]
pub struct HighPerfCacheStats {
pub l1_entries: usize,
pub l2_entries: usize,
pub hits: AtomicU64,
pub misses: AtomicU64,
pub total_modules: usize,
pub total_size: usize,
pub precompiled_modules: usize,
pub zero_copy_operations: AtomicU64,
pub async_io_operations: AtomicU64,
pub avg_load_time_ns: AtomicU64,
pub load_operations: AtomicU64,
}
impl HighPerfCacheStats {
pub fn hit_ratio(&self) -> f64 {
let hits: _ = self.hits.load(Ordering::Relaxed);
let total: _ = hits + self.misses.load(Ordering::Relaxed);
if total > 0 {
hits as f64 / total as f64
} else {
0.0
}
}
pub fn avg_load_time(&self) -> Duration {
let total_ns: _ = self.avg_load_time_ns.load(Ordering::Relaxed);
let operations: _ = self.load_operations.load(Ordering::Relaxed);
if operations > 0 {
Duration::from_nanos(total_ns / operations)
} else {
Duration::default()
}
}
}
pub struct HighPerformanceWasmCache {
l1_cache: Arc<RwLock<HashMap<String, Arc<HighPerfCacheEntry>>>>,
l2_cache: Arc<Mutex<HashMap<String, PathBuf>>>,
config: HighPerfCacheConfig,
stats: Arc<HighPerfCacheStats>,
}
impl HighPerformanceWasmCache {
pub fn new() -> Result<Self> {
Self::new_with_config(HighPerfCacheConfig::default())
}
pub fn new_with_config(config: HighPerfCacheConfig) -> Result<Self> {
if config.enable_l2 {
std::fs::create_dir_all(&config.l2_cache_dir)
.context("Failed to create L2 cache directory")?;
}
Ok(HighPerformanceWasmCache {
l1_cache: Arc::new(Mutex::new(HashMap::new()))
l2_cache: Arc::new(Mutex::new(HashMap::new()))
config,
stats: Arc::new(Mutex::new(HighPerfCacheStats::default()))
})
}
pub async fn store_module(&self, module_hash: String, wasm_bytes: Vec<u8>) -> Result<()> {
let start: _ = Instant::now();
let hash: _ = if module_hash.is_empty() {
self.calculate_hash(&wasm_bytes)
} else {
module_hash
};
let entry: _ = Arc::new(Mutex::new(HighPerfCacheEntry::new(wasm_bytes, hash.clone()),;
{
let mut l1 = self.l1_cache.write().unwrap();
l1.insert(hash.clone(), Arc::clone(entry));
}
if self.config.enable_l2 {
let hash_clone: _ = hash.clone();
let entry_clone: _ = Arc::clone(entry);
tokio::spawn(async move {
if let Err(e) = Self::store_to_l2_async(&hash_clone, &entry_clone).await {
eprintln!("Warning: Failed to store to L2 cache: {}", e);
}
});
}
self.record_load_time(start.elapsed());
Ok(())
}
pub async fn load_module(&self, module_hash: &str) -> Result<Arc<Vec<u8> {
let start: _ = Instant::now();
{
let l1: _ = self.l1_cache.read().unwrap();
if let Some(entry) = l1.get(module_hash) {
let _: _ = entry.clone(); self.stats.zero_copy_operations.fetch_add(1, Ordering::Relaxed);
let load_time: _ = start.elapsed();
self.record_load_time(load_time);
self.stats.hits.fetch_add(1, Ordering::Relaxed);
return Ok(Arc::clone(&entry.wasm_bytes));
}
}
if self.config.enable_l2 {
if let Ok(wasm_bytes) = self.load_from_l2_async(module_hash).await {
if !wasm_bytes.is_empty() {
let cache: _ = self.clone();
let hash: _ = module_hash.to_string();
let bytes: _ = wasm_bytes.clone();
tokio::spawn(async move {
let _: _ = cache.store_module(hash, bytes).await;
});
let load_time: _ = start.elapsed();
self.record_load_time(load_time);
self.stats.hits.fetch_add(1, Ordering::Relaxed);
return Ok(Arc::new(Mutex::new(wasm_bytes)),;
}
}
}
self.stats.misses.fetch_add(1, Ordering::Relaxed);
Err(anyhow!("Module not found in cache: {}", module_hash))
}
async fn store_to_l2_async(
hash: &str,
entry: &Arc<HighPerfCacheEntry>,
) -> Result<()> {
let wasm_bytes: _ = Arc::clone(&entry.wasm_bytes);
let file_path: _ = format!("benchmarks/fixtures/wasm_cache_high_perf/{}.wasm", hash);
let mut file = fs::File::create(&file_path).await?;
file.write_all(&wasm_bytes).await?;
Ok(())
}
async fn load_from_l2_async(&self, module_hash: &str) -> Result<Vec<u8> {
let file_path: _ = {
let l2: _ = self.l2_cache.lock().unwrap();
l2.get(module_hash).cloned()
};
if let Some(file_path) = file_path {
let mut file = fs::File::open(&file_path).await?;
let mut buffer = Vec::new();
file.read_to_end(&mut buffer).await?;
self.stats.async_io_operations.fetch_add(1, Ordering::Relaxed);
Ok(buffer)
} else {
Err(anyhow!("Module not found in L2 cache"))
}
}
pub async fn load_modules_batch(
&self,
module_hashes: Vec<String>,
) -> Result<Vec<Result<Arc<Vec<u8>, anyhow::Error>> {
let mut results = Vec::with_capacity(module_hashes.len());
let mut handles = Vec::new();
for hash in module_hashes {
let cache: _ = self.clone();
let hash_clone: _ = hash.clone();
let handle: _ = tokio::spawn(async move {
match cache.load_module(&hash_clone).await {
Ok(bytes) => Ok(bytes),
Err(e) => Err(e),
}
});
handles.push(handle);
}
for handle in handles {
results.push(handle.await.map_err(|_| anyhow!("Task panicked"))?);
}
Ok(results)
}
pub async fn warmup_cache_batch(&self, modules: Vec<(String, Vec<u8>)>) -> Result<()> {
let mut handles = Vec::new();
for chunk in modules.chunks(self.config.batch_size) {
for (hash, wasm_bytes) in chunk {
let cache: _ = self.clone();
let hash_clone: _ = hash.clone();
let bytes_clone: _ = wasm_bytes.clone();
let handle: _ = tokio::spawn(async move {
cache.store_module(hash_clone, bytes_clone).await
});
handles.push(handle);
}
for handle in handles.drain(..) {
handle.await.map_err(|_| anyhow!("Warmup task failed"))??;
}
}
Ok(())
}
pub fn calculate_hash(&self, wasm_bytes: &[u8]) -> String {
let mut hasher = Hasher::new();
hasher.update(wasm_bytes);
hasher.finalize().to_hex().to_string()
}
fn clone(&self) -> Arc<Self> {
Arc::new(Mutex::new(HighPerformanceWasmCache {)),
l1_cache: Arc::clone(&self.l1_cache))
l2_cache: Arc::clone(&self.l2_cache),
config: self.config.clone(),
stats: Arc::clone(&self.stats),
})
}
fn record_load_time(&self, load_time: Duration) {
self.stats.load_operations.fetch_add(1, Ordering::Relaxed);
self.stats.avg_load_time_ns.fetch_add(load_time.as_nanos() as u64, Ordering::Relaxed);
}
pub fn get_stats(&self) -> HighPerfCacheStats {
let mut stats = HighPerfCacheStats::default();
stats.hits = AtomicU64::new(self.stats.hits.load(Ordering::Relaxed));
stats.misses = AtomicU64::new(self.stats.misses.load(Ordering::Relaxed));
stats.zero_copy_operations = AtomicU64::new(self.stats.zero_copy_operations.load(Ordering::Relaxed));
stats.async_io_operations = AtomicU64::new(self.stats.async_io_operations.load(Ordering::Relaxed));
stats.avg_load_time_ns = AtomicU64::new(self.stats.avg_load_time_ns.load(Ordering::Relaxed));
stats.load_operations = AtomicU64::new(self.stats.load_operations.load(Ordering::Relaxed));
{
let l1: _ = self.l1_cache.read().unwrap();
stats.l1_entries = l1.len();
stats.precompiled_modules = 0; }
if self.config.enable_l2 {
let l2: _ = self.l2_cache.lock().unwrap();
stats.l2_entries = l2.len();
}
stats
}
}
#[cfg(test)]
mod tests {
#[tokio::test]
async fn test_high_performance_cache_creation() {
let cache: _ = HighPerformanceWasmCache::new().unwrap();
let stats: _ = cache.get_stats();
assert_eq!(stats.l1_entries, 0);
assert_eq!(stats.l2_entries, 0);
println!("✅ 高性能缓存创建成功");
}
#[tokio::test]
async fn test_zero_copy_operations() {
let cache: _ = HighPerformanceWasmCache::new().unwrap();
let wasm_bytes: _ = vec![0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00];
let hash: _ = cache.calculate_hash(&wasm_bytes);
cache.store_module(hash.clone(), wasm_bytes.clone()).await.unwrap();
let loaded_bytes: _ = cache.load_module(&hash).await.unwrap();
let wasm_bytes_ref: _ = &wasm_bytes;
assert!(Arc::ptr_eq(&loaded_bytes, &Arc::new(Mutex::new(wasm_bytes_ref.to_vec()), || std::ptr::eq(loaded_bytes.as_ptr(), wasm_bytes_ref.as_ptr());
let stats: _ = cache.get_stats();
assert_eq!(stats.zero_copy_operations.load(Ordering::Relaxed), 1);
println!("✅ 零拷贝操作测试通过");
}
#[tokio::test]
async fn test_async_l2_cache() {
let cache: _ = HighPerformanceWasmCache::new().unwrap();
let wasm_bytes: _ = wat::parse_str(r#"
(module
(func $_start (export "_start") nop)
)
"#).expect("创建WASM字节码失败");
let hash: _ = cache.calculate_hash(&wasm_bytes);
cache.store_module(hash.clone(), wasm_bytes.clone()).await.unwrap();
tokio::time::sleep(Duration::from_millis(100)).await;
let loaded_bytes: _ = cache.load_module(&hash).await.unwrap();
assert_eq!(loaded_bytes.len(), wasm_bytes.len());
let stats: _ = cache.get_stats();
assert!(stats.async_io_operations.load(Ordering::Relaxed) > 0);
println!("✅ 异步 L2 缓存测试通过");
}
#[tokio::test]
async fn test_batch_operations() {
let cache: _ = HighPerformanceWasmCache::new().unwrap();
let mut modules = Vec::new();
for i in 0..10 {
let wasm_bytes: _ = wat::parse_str(&format!(r#"
(module
(func $_start (export "_start") nop)
(data (i32.const 0) "{}")
)
"#, i)).expect("创建WASM字节码失败");
let hash: _ = cache.calculate_hash(&wasm_bytes);
modules.push((hash, wasm_bytes));
}
cache.warmup_cache_batch(modules.clone()).await.unwrap();
let hashes: Vec<String> = modules.iter().map(|(h, _)| h.clone()).collect();
let results: _ = cache.load_modules_batch(hashes).await.unwrap();
assert_eq!(results.len(), 10);
let success_count: _ = results.iter().filter(|r| r.is_ok()).count();
assert_eq!(success_count, 10);
println!("✅ 批量操作测试通过");
}
}