use anyhow::{Context, Result};
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};
use tracing::{debug, info};
use std::time::{Duration, Instant};
#[derive(Debug, Clone)]
pub struct MultithreadExecutionResult {
pub thread_count: usize,
pub total_time_ms: f64,
pub time_per_thread_ms: f64,
pub speedup: f64,
pub scaling_efficiency: f64,
pub tasks_completed: usize,
}
#[derive(Debug, Clone)]
pub struct ThreadPoolConfig {
pub max_threads: usize,
pub thread_pool_size: usize,
pub work_stealing: bool,
}
pub struct WasmMultithread {
engine: Arc<Engine>,
thread_pool: rayon::ThreadPool,
config: ThreadPoolConfig,
}
impl WasmMultithread {
pub fn new(config: Option<ThreadPoolConfig>) -> Result<Self> {
let config: _ = config.unwrap_or_else(|| ThreadPoolConfig {
max_threads: num_cpus::get(),
thread_pool_size: num_cpus::get(),
work_stealing: true,
});
info!("🧵 初始化 WASM 多线程执行器 (线程数: {})", config.thread_pool_size);
let thread_pool: _ = rayon::ThreadPoolBuilder::new()
.num_threads(config.thread_pool_size)
.thread_name(|i| format!("wasm-worker-{}", i))
.build()
.context("创建线程池失败")?;
Ok(Self {
engine: Arc::new(Mutex::new(Engine::new(
&wasmtime::Config::new()
.wasm_threads(true)
.wasm_simd(true)
.parallel_compilation(true)
)?)),
thread_pool,
config,
})
}
pub fn compile_modules_parallel(&self, modules: Vec<(String, Vec<u8>)>) -> Result<Vec<String>> {
info!("📦 并行编译 {} 个 WASM 模块", modules.len());
let start_time: _ = std::time::Instant::now();
let results: Vec<String> = self.thread_pool.install(|| {
modules
.par_iter()
.map(|(name, _wasm_bytes)| {
info!("✅ 编译完成: {}", name);
name.clone()
})
.collect()
});
let compile_time: _ = start_time.elapsed().as_secs_f64() * 1000.0;
info!("🚀 并行编译完成 (耗时: {:.2}ms, 线程数: {})", compile_time, self.config.thread_pool_size);
Ok(results)
}
pub fn execute_parallel(&self, name: &str, tasks: Vec<Vec<u8>>) -> Result<MultithreadExecutionResult> {
let thread_count: _ = self.config.thread_pool_size.min(tasks.len());
let start_time: _ = std::time::Instant::now();
info!("⚡ 开始并行执行: {} (线程数: {}, 任务数: {})", name, thread_count, tasks.len());
let chunk_size: _ = (tasks.len() + thread_count - 1) / thread_count;
let task_chunks: Vec<_> = tasks.chunks(chunk_size).collect();
let execution_results: Vec<MultithreadExecutionResult> = self.thread_pool.install(|| {
task_chunks
.par_iter()
.map(|chunk| {
self.execute_task_chunk(name, chunk.to_vec())
})
.collect()
});
let total_time: _ = start_time.elapsed().as_secs_f64() * 1000.0;
let total_tasks: _ = execution_results.iter().map(|r| r.tasks_completed).sum();
let avg_time_per_thread: _ = execution_results.iter().map(|r| r.time_per_thread_ms).sum::<f64>() / thread_count as f64;
let speedup: _ = if execution_results.len() > 1 {
execution_results[0].time_per_thread_ms * thread_count as f64 / total_time
} else {
1.0
};
let scaling_efficiency: _ = speedup / thread_count as f64;
let result: _ = MultithreadExecutionResult {
thread_count,
total_time_ms: total_time,
time_per_thread_ms: avg_time_per_thread,
speedup,
scaling_efficiency,
tasks_completed: total_tasks,
};
info!("✅ 并行执行完成: {} (总耗时: {:.2}ms, 加速比: {:.2}x, 扩展效率: {:.1}%)",
name, total_time, speedup, scaling_efficiency * 100.0);
Ok(result)
}
fn execute_task_chunk(&self, name: &str, tasks: Vec<Vec<u8>>) -> MultithreadExecutionResult {
let chunk_start: _ = std::time::Instant::now();
for _task_input in &tasks {
std::thread::sleep(std::time::Duration::from_millis(10));
}
let chunk_time: _ = chunk_start.elapsed().as_secs_f64() * 1000.0;
MultithreadExecutionResult {
thread_count: 1,
total_time_ms: chunk_time,
time_per_thread_ms: chunk_time,
speedup: 1.0,
scaling_efficiency: 1.0,
tasks_completed: tasks.len(),
}
}
pub fn parallel_matrix_multiply(&self, a: &[f32], b: &[f32], size: usize) -> Result<Vec<f32>> {
info!("🧮 开始并行矩阵乘法 (大小: {}x{})", size, size);
let start_time: _ = std::time::Instant::now();
let mut result = vec![0.0f32; size * size];
self.thread_pool.install(|| {
result
.par_iter_mut()
.enumerate()
.for_each(|(idx, val)| {
let row: _ = idx / size;
let col: _ = idx % size;
let mut sum = 0.0f32;
for k in 0..size {
sum += a[row * size + k] * b[k * size + col];
}
*val = sum;
});
});
let compute_time: _ = start_time.elapsed().as_secs_f64() * 1000.0;
info!("✅ 矩阵乘法完成 (耗时: {:.2}ms, 线程数: {})", compute_time, self.config.thread_pool_size);
Ok(result)
}
pub fn benchmark_scaling(&self, max_threads: usize) -> Result<Vec<MultithreadExecutionResult>> {
info!("📊 开始多线程扩展性基准测试 (最大线程数: {})", max_threads);
let mut results = Vec::new();
let test_tasks: Vec<Vec<u8>> = (0..1000).map(|i| vec![i as u8]).collect();
for thread_count in 1..=max_threads {
let config: _ = ThreadPoolConfig {
max_threads: thread_count,
thread_pool_size: thread_count,
work_stealing: true,
};
let executor: _ = WasmMultithread::new(Some(config))?;
let result: _ = executor.execute_parallel("benchmark", test_tasks.clone())?;
results.push(result);
}
info!("✅ 基准测试完成");
if let (Some(first), Some(last)) = (results.first(), results.last()) {
let total_speedup: _ = last.speedup;
let efficiency: _ = last.scaling_efficiency;
if efficiency >= 0.8 {
info!("🎉 扩展性优秀: 加速比 {:.2}x, 扩展效率 {:.1}%", total_speedup, efficiency * 100.0);
} else if efficiency >= 0.5 {
info!("✅ 扩展性良好: 加速比 {:.2}x, 扩展效率 {:.1}%", total_speedup, efficiency * 100.0);
} else {
info!("⚠️ 扩展性一般: 加速比 {:.2}x, 扩展效率 {:.1}%", total_speedup, efficiency * 100.0);
}
}
Ok(results)
}
pub fn get_config(&self) -> &ThreadPoolConfig {
&self.config
}
pub fn get_thread_count(&self) -> usize {
self.config.thread_pool_size
}
}
impl Default for WasmMultithread {
fn default() -> Self {
Self::new(None).expect("初始化 WasmMultithread 失败")
}
}