use crate::{
multi_gpu::{
DeviceId, GpuDevice, IntelligentLoadBalancer, LoadBalancingStrategy, Workload,
WorkloadCoordinator,
},
GpuError,
};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use thiserror::Error;
use tokio::sync::RwLock;
use wgpu::util::DeviceExt;
#[derive(Error, Debug)]
pub enum UnifiedGpuError {
#[error("GPU error: {0}")]
Gpu(#[from] GpuError),
#[error("Shader compilation failed: {0}")]
ShaderCompilation(String),
#[error("Buffer size mismatch: expected {expected}, got {actual}")]
BufferSizeMismatch { expected: usize, actual: usize },
#[error("Invalid operation: {0}")]
InvalidOperation(String),
#[error("Memory allocation failed: {0}")]
MemoryAllocation(String),
}
pub type UnifiedGpuResult<T> = Result<T, UnifiedGpuError>;
pub trait GpuAccelerated<T> {
fn to_gpu_buffer(&self, context: &GpuContext) -> UnifiedGpuResult<wgpu::Buffer>;
fn from_gpu_buffer(buffer: &wgpu::Buffer, context: &GpuContext) -> UnifiedGpuResult<T>;
fn gpu_operation(
&self,
operation: &str,
context: &GpuContext,
params: &GpuOperationParams,
) -> UnifiedGpuResult<T>;
}
#[derive(Debug, Clone)]
pub struct GpuOperationParams {
pub params: HashMap<String, GpuParam>,
pub batch_size: usize,
pub workgroup_size: (u32, u32, u32),
}
#[derive(Debug, Clone)]
pub enum GpuParam {
Float(f32),
Double(f64),
Integer(i32),
UnsignedInteger(u32),
Buffer(String), Array(Vec<f32>),
}
impl Default for GpuOperationParams {
fn default() -> Self {
Self {
params: HashMap::new(),
batch_size: 1,
workgroup_size: (1, 1, 1),
}
}
}
pub struct GpuContext {
pub device: wgpu::Device,
pub queue: wgpu::Queue,
shader_cache: HashMap<String, wgpu::ComputePipeline>,
#[allow(dead_code)]
buffer_pool: GpuBufferPool,
}
impl GpuContext {
pub async fn new() -> UnifiedGpuResult<Self> {
let instance = wgpu::Instance::default();
let adapter = instance
.request_adapter(&wgpu::RequestAdapterOptions {
power_preference: wgpu::PowerPreference::HighPerformance,
compatible_surface: None,
force_fallback_adapter: false,
})
.await
.ok_or_else(|| {
UnifiedGpuError::Gpu(GpuError::InitializationError(
"No GPU adapter found".to_string(),
))
})?;
let (device, queue) = adapter
.request_device(
&wgpu::DeviceDescriptor {
label: Some("Amari Unified GPU Device"),
required_features: wgpu::Features::empty(),
required_limits: wgpu::Limits::default(),
},
None,
)
.await
.map_err(|e| UnifiedGpuError::Gpu(GpuError::InitializationError(e.to_string())))?;
Ok(Self {
device,
queue,
shader_cache: HashMap::new(),
buffer_pool: GpuBufferPool::new(),
})
}
pub fn get_compute_pipeline(
&mut self,
shader_key: &str,
shader_source: &str,
bind_group_layout: &wgpu::BindGroupLayout,
) -> UnifiedGpuResult<&wgpu::ComputePipeline> {
if !self.shader_cache.contains_key(shader_key) {
let shader_module = self
.device
.create_shader_module(wgpu::ShaderModuleDescriptor {
label: Some(&format!("{} Shader", shader_key)),
source: wgpu::ShaderSource::Wgsl(shader_source.into()),
});
let pipeline_layout =
self.device
.create_pipeline_layout(&wgpu::PipelineLayoutDescriptor {
label: Some(&format!("{} Pipeline Layout", shader_key)),
bind_group_layouts: &[bind_group_layout],
push_constant_ranges: &[],
});
let compute_pipeline =
self.device
.create_compute_pipeline(&wgpu::ComputePipelineDescriptor {
label: Some(&format!("{} Pipeline", shader_key)),
layout: Some(&pipeline_layout),
module: &shader_module,
entry_point: "main",
});
self.shader_cache
.insert(shader_key.to_string(), compute_pipeline);
}
Ok(self
.shader_cache
.get(shader_key)
.expect("Pipeline should exist"))
}
pub fn create_buffer_with_data<T: bytemuck::Pod>(
&self,
label: &str,
data: &[T],
usage: wgpu::BufferUsages,
) -> wgpu::Buffer {
self.device
.create_buffer_init(&wgpu::util::BufferInitDescriptor {
label: Some(label),
contents: bytemuck::cast_slice(data),
usage,
})
}
pub fn create_buffer(&self, label: &str, size: u64, usage: wgpu::BufferUsages) -> wgpu::Buffer {
self.device.create_buffer(&wgpu::BufferDescriptor {
label: Some(label),
size,
usage,
mapped_at_creation: false,
})
}
pub fn execute_compute(
&self,
pipeline: &wgpu::ComputePipeline,
bind_group: &wgpu::BindGroup,
workgroup_count: (u32, u32, u32),
) {
let mut encoder = self
.device
.create_command_encoder(&wgpu::CommandEncoderDescriptor {
label: Some("Compute Encoder"),
});
{
let mut compute_pass = encoder.begin_compute_pass(&wgpu::ComputePassDescriptor {
label: Some("Compute Pass"),
timestamp_writes: None,
});
compute_pass.set_pipeline(pipeline);
compute_pass.set_bind_group(0, bind_group, &[]);
compute_pass.dispatch_workgroups(
workgroup_count.0,
workgroup_count.1,
workgroup_count.2,
);
}
self.queue.submit([encoder.finish()]);
}
pub async fn read_buffer<T: bytemuck::Pod + Clone>(
&self,
buffer: &wgpu::Buffer,
size: u64,
) -> UnifiedGpuResult<Vec<T>> {
let staging_buffer = self.device.create_buffer(&wgpu::BufferDescriptor {
label: Some("Staging Buffer"),
size,
usage: wgpu::BufferUsages::COPY_DST | wgpu::BufferUsages::MAP_READ,
mapped_at_creation: false,
});
let mut encoder = self
.device
.create_command_encoder(&wgpu::CommandEncoderDescriptor {
label: Some("Copy Encoder"),
});
encoder.copy_buffer_to_buffer(buffer, 0, &staging_buffer, 0, size);
self.queue.submit([encoder.finish()]);
let buffer_slice = staging_buffer.slice(..);
let (tx, rx) = futures::channel::oneshot::channel();
buffer_slice.map_async(wgpu::MapMode::Read, move |result| {
tx.send(result).ok();
});
self.device.poll(wgpu::Maintain::Wait);
rx.await
.map_err(|_| UnifiedGpuError::InvalidOperation("Buffer read timeout".to_string()))?
.map_err(|e| UnifiedGpuError::InvalidOperation(format!("Buffer map failed: {}", e)))?;
let data = buffer_slice.get_mapped_range();
let result: Vec<T> = bytemuck::cast_slice(&data).to_vec();
drop(data);
staging_buffer.unmap();
Ok(result)
}
}
pub struct GpuBufferPool {
_pools: HashMap<String, Vec<wgpu::Buffer>>, }
impl GpuBufferPool {
pub fn new() -> Self {
Self {
_pools: HashMap::new(),
}
}
}
impl Default for GpuBufferPool {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone)]
pub struct SharedGpuContext {
device: Arc<wgpu::Device>,
queue: Arc<wgpu::Queue>,
adapter_info: wgpu::AdapterInfo,
buffer_pool: Arc<std::sync::Mutex<EnhancedGpuBufferPool>>,
shader_cache: Arc<std::sync::Mutex<HashMap<String, Arc<wgpu::ComputePipeline>>>>,
creation_time: Instant,
multi_gpu_enabled: bool,
gpu_devices: Arc<RwLock<HashMap<DeviceId, Arc<GpuDevice>>>>,
load_balancer: Arc<IntelligentLoadBalancer>,
workload_coordinator: Arc<WorkloadCoordinator>,
primary_device_id: DeviceId,
}
impl SharedGpuContext {
pub async fn global() -> UnifiedGpuResult<&'static Self> {
let context = Self::new().await?;
Ok(Box::leak(Box::new(context)))
}
async fn new() -> UnifiedGpuResult<Self> {
let instance = wgpu::Instance::new(wgpu::InstanceDescriptor {
backends: wgpu::Backends::all(),
flags: wgpu::InstanceFlags::default(),
dx12_shader_compiler: wgpu::Dx12Compiler::default(),
gles_minor_version: wgpu::Gles3MinorVersion::Automatic,
});
let adapter = instance
.request_adapter(&wgpu::RequestAdapterOptions {
power_preference: wgpu::PowerPreference::HighPerformance,
compatible_surface: None,
force_fallback_adapter: false,
})
.await
.ok_or_else(|| {
UnifiedGpuError::InvalidOperation("No suitable GPU adapter found".into())
})?;
let adapter_info = adapter.get_info();
let (device, queue) = adapter
.request_device(
&wgpu::DeviceDescriptor {
label: Some("Shared Amari GPU Device"),
required_features: wgpu::Features::TIMESTAMP_QUERY,
required_limits: wgpu::Limits::default(),
},
None,
)
.await
.map_err(|e| {
UnifiedGpuError::InvalidOperation(format!("Device request failed: {:?}", e))
})?;
let primary_device_id = DeviceId(0);
let gpu_device = Arc::new(
GpuDevice::new(primary_device_id, &adapter, device, queue)
.await
.map_err(|_| {
UnifiedGpuError::InvalidOperation("Failed to create GPU device".into())
})?,
);
let device_arc = Arc::clone(&gpu_device.device);
let queue_arc = Arc::clone(&gpu_device.queue);
let mut gpu_devices = HashMap::new();
gpu_devices.insert(primary_device_id, gpu_device);
Ok(Self {
device: device_arc,
queue: queue_arc,
adapter_info,
buffer_pool: Arc::new(std::sync::Mutex::new(EnhancedGpuBufferPool::new())),
shader_cache: Arc::new(std::sync::Mutex::new(HashMap::new())),
creation_time: Instant::now(),
multi_gpu_enabled: false,
gpu_devices: Arc::new(RwLock::new(gpu_devices)),
load_balancer: Arc::new(IntelligentLoadBalancer::new(
LoadBalancingStrategy::Balanced,
)),
workload_coordinator: Arc::new(WorkloadCoordinator::new()),
primary_device_id,
})
}
pub async fn with_multi_gpu() -> UnifiedGpuResult<Self> {
let instance = wgpu::Instance::new(wgpu::InstanceDescriptor {
backends: wgpu::Backends::all(),
flags: wgpu::InstanceFlags::default(),
dx12_shader_compiler: wgpu::Dx12Compiler::default(),
gles_minor_version: wgpu::Gles3MinorVersion::Automatic,
});
let adapters: Vec<_> = instance.enumerate_adapters(wgpu::Backends::all());
if adapters.is_empty() {
return Err(UnifiedGpuError::InvalidOperation(
"No GPU adapters found".into(),
));
}
let mut gpu_devices = HashMap::new();
let mut primary_device = None;
let mut primary_queue = None;
let mut primary_adapter_info = None;
for (i, adapter) in adapters.iter().enumerate() {
let device_id = DeviceId(i);
if let Ok((device, queue)) = adapter
.request_device(
&wgpu::DeviceDescriptor {
label: Some(&format!("Amari Multi-GPU Device {}", i)),
required_features: wgpu::Features::TIMESTAMP_QUERY,
required_limits: wgpu::Limits::default(),
},
None,
)
.await
{
if let Ok(gpu_device) = GpuDevice::new(device_id, adapter, device, queue).await {
if primary_device.is_none() {
primary_device = Some(Arc::clone(&gpu_device.device));
primary_queue = Some(Arc::clone(&gpu_device.queue));
primary_adapter_info = Some(adapter.get_info());
}
gpu_devices.insert(device_id, Arc::new(gpu_device));
}
}
}
if gpu_devices.is_empty() {
return Err(UnifiedGpuError::InvalidOperation(
"No usable GPU devices found".into(),
));
}
let primary_device_id = DeviceId(0);
let load_balancer = Arc::new(IntelligentLoadBalancer::new(
LoadBalancingStrategy::CapabilityAware,
));
for device in gpu_devices.values() {
load_balancer.add_device(Arc::clone(device)).await;
}
Ok(Self {
device: primary_device.unwrap(),
queue: primary_queue.unwrap(),
adapter_info: primary_adapter_info.unwrap(),
buffer_pool: Arc::new(std::sync::Mutex::new(EnhancedGpuBufferPool::new())),
shader_cache: Arc::new(std::sync::Mutex::new(HashMap::new())),
creation_time: Instant::now(),
multi_gpu_enabled: true,
gpu_devices: Arc::new(RwLock::new(gpu_devices)),
load_balancer,
workload_coordinator: Arc::new(WorkloadCoordinator::new()),
primary_device_id,
})
}
pub fn device(&self) -> &wgpu::Device {
&self.device
}
pub fn queue(&self) -> &wgpu::Queue {
&self.queue
}
pub fn adapter_info(&self) -> &wgpu::AdapterInfo {
&self.adapter_info
}
pub fn get_buffer(
&self,
size: u64,
usage: wgpu::BufferUsages,
label: Option<&str>,
) -> wgpu::Buffer {
if let Ok(mut pool) = self.buffer_pool.lock() {
pool.get_or_create(&self.device, size, usage, label)
} else {
self.device.create_buffer(&wgpu::BufferDescriptor {
label,
size,
usage,
mapped_at_creation: false,
})
}
}
pub fn return_buffer(&self, buffer: wgpu::Buffer, size: u64, usage: wgpu::BufferUsages) {
if let Ok(mut pool) = self.buffer_pool.lock() {
pool.return_buffer(buffer, size, usage);
}
}
pub fn get_compute_pipeline(
&self,
shader_key: &str,
shader_source: &str,
entry_point: &str,
) -> UnifiedGpuResult<Arc<wgpu::ComputePipeline>> {
let cache_key = format!("{}:{}", shader_key, entry_point);
if let Ok(mut cache) = self.shader_cache.lock() {
if let Some(pipeline) = cache.get(&cache_key) {
return Ok(Arc::clone(pipeline));
}
let shader_module = self
.device
.create_shader_module(wgpu::ShaderModuleDescriptor {
label: Some(&format!("{} Shader", shader_key)),
source: wgpu::ShaderSource::Wgsl(shader_source.into()),
});
let bind_group_layout =
self.device
.create_bind_group_layout(&wgpu::BindGroupLayoutDescriptor {
label: Some(&format!("{} Bind Group Layout", shader_key)),
entries: &[
wgpu::BindGroupLayoutEntry {
binding: 0,
visibility: wgpu::ShaderStages::COMPUTE,
ty: wgpu::BindingType::Buffer {
ty: wgpu::BufferBindingType::Storage { read_only: true },
has_dynamic_offset: false,
min_binding_size: None,
},
count: None,
},
wgpu::BindGroupLayoutEntry {
binding: 1,
visibility: wgpu::ShaderStages::COMPUTE,
ty: wgpu::BindingType::Buffer {
ty: wgpu::BufferBindingType::Storage { read_only: false },
has_dynamic_offset: false,
min_binding_size: None,
},
count: None,
},
],
});
let pipeline_layout =
self.device
.create_pipeline_layout(&wgpu::PipelineLayoutDescriptor {
label: Some(&format!("{} Pipeline Layout", shader_key)),
bind_group_layouts: &[&bind_group_layout],
push_constant_ranges: &[],
});
let pipeline = self
.device
.create_compute_pipeline(&wgpu::ComputePipelineDescriptor {
label: Some(&format!("{} Pipeline", shader_key)),
layout: Some(&pipeline_layout),
module: &shader_module,
entry_point,
});
let pipeline_arc = Arc::new(pipeline);
cache.insert(cache_key, Arc::clone(&pipeline_arc));
Ok(pipeline_arc)
} else {
Err(UnifiedGpuError::InvalidOperation(
"Failed to access shader cache".into(),
))
}
}
pub fn buffer_pool_stats(&self) -> BufferPoolStats {
if let Ok(pool) = self.buffer_pool.lock() {
pool.get_stats()
} else {
BufferPoolStats::default()
}
}
pub fn uptime(&self) -> std::time::Duration {
self.creation_time.elapsed()
}
pub fn get_optimal_workgroup(&self, operation: &str, data_size: usize) -> (u32, u32, u32) {
match operation {
"matrix_multiply" | "matrix_operation" => {
(16, 16, 1)
}
"vector_operation" | "reduce" | "scan" => {
let workgroup_size = if data_size > 10000 {
256 } else if data_size > 1000 {
128 } else {
64 };
(workgroup_size, 1, 1)
}
"geometric_algebra" | "clifford_algebra" => {
(128, 1, 1)
}
"cellular_automata" | "ca_evolution" => {
(16, 16, 1)
}
"neural_network" | "batch_processing" => {
(256, 1, 1)
}
"information_geometry" | "fisher_information" | "bregman_divergence" => {
(256, 1, 1)
}
"tropical_algebra" | "tropical_matrix" => {
(128, 1, 1)
}
"dual_number" | "automatic_differentiation" => {
(128, 1, 1)
}
"fusion_system" | "llm_evaluation" => {
(256, 1, 1)
}
"enumerative_geometry" | "intersection_theory" => {
(64, 1, 1)
}
_ => (64, 1, 1), }
}
pub fn get_workgroup_declaration(&self, operation: &str, data_size: usize) -> String {
let (x, y, z) = self.get_optimal_workgroup(operation, data_size);
if y == 1 && z == 1 {
format!("@compute @workgroup_size({})", x)
} else if z == 1 {
format!("@compute @workgroup_size({}, {})", x, y)
} else {
format!("@compute @workgroup_size({}, {}, {})", x, y, z)
}
}
pub fn is_multi_gpu_enabled(&self) -> bool {
self.multi_gpu_enabled
}
pub async fn device_count(&self) -> usize {
self.gpu_devices.read().await.len()
}
pub async fn get_device_info(&self) -> Vec<(DeviceId, String, String)> {
let devices = self.gpu_devices.read().await;
devices
.iter()
.map(|(id, device)| {
(
*id,
device.adapter_info.name.clone(),
format!("{:?}", device.capabilities.architecture),
)
})
.collect()
}
pub async fn get_device(&self, device_id: DeviceId) -> Option<Arc<GpuDevice>> {
let devices = self.gpu_devices.read().await;
devices.get(&device_id).cloned()
}
pub async fn optimal_device_for_operation(
&self,
operation: &str,
_data_size: usize,
) -> DeviceId {
if !self.multi_gpu_enabled {
return self.primary_device_id;
}
let devices = self.gpu_devices.read().await;
let available_devices: Vec<_> = devices
.values()
.filter(|device| device.is_available())
.collect();
if available_devices.is_empty() {
return self.primary_device_id;
}
available_devices
.iter()
.max_by(|a, b| {
a.performance_score(operation)
.partial_cmp(&b.performance_score(operation))
.unwrap_or(std::cmp::Ordering::Equal)
})
.map(|device| device.id)
.unwrap_or(self.primary_device_id)
}
pub async fn distribute_workload(
&self,
workload: Workload,
) -> UnifiedGpuResult<Vec<crate::multi_gpu::DeviceWorkload>> {
if !self.multi_gpu_enabled {
return Ok(vec![crate::multi_gpu::DeviceWorkload {
device_id: self.primary_device_id,
workload_fraction: 1.0,
data_range: (0, workload.data_size),
estimated_completion_ms: 100.0,
memory_requirement_mb: workload.memory_requirement_mb,
}]);
}
self.load_balancer
.distribute_workload(&workload)
.await
.map_err(|e| {
UnifiedGpuError::InvalidOperation(format!("Workload distribution failed: {:?}", e))
})
}
pub async fn execute_multi_gpu_workload(
&self,
workload_id: String,
workload: Workload,
) -> UnifiedGpuResult<Vec<Vec<u8>>> {
if !self.multi_gpu_enabled {
return Err(UnifiedGpuError::InvalidOperation(
"Multi-GPU mode not enabled".into(),
));
}
let assignments = self.distribute_workload(workload).await?;
self.workload_coordinator
.submit_workload(workload_id.clone(), assignments)
.await
.map_err(|e| {
UnifiedGpuError::InvalidOperation(format!("Workload submission failed: {:?}", e))
})?;
let timeout = std::time::Duration::from_secs(30);
self.workload_coordinator
.wait_for_completion(&workload_id, timeout)
.await
.map_err(|e| {
UnifiedGpuError::InvalidOperation(format!("Workload execution failed: {:?}", e))
})
}
pub async fn get_gpu_utilization(&self) -> HashMap<DeviceId, f32> {
let devices: tokio::sync::RwLockReadGuard<HashMap<DeviceId, Arc<GpuDevice>>> =
self.gpu_devices.read().await;
devices
.iter()
.map(|(id, device): (&DeviceId, &Arc<GpuDevice>)| (*id, device.current_load()))
.collect()
}
pub async fn get_multi_gpu_stats(&self) -> MultiGpuStats {
let devices: tokio::sync::RwLockReadGuard<HashMap<DeviceId, Arc<GpuDevice>>> =
self.gpu_devices.read().await;
let device_count = devices.len();
let total_operations: usize = devices
.values()
.map(|device| {
device
.total_operations
.load(std::sync::atomic::Ordering::Relaxed)
})
.sum();
let total_errors: usize = devices
.values()
.map(|device| {
device
.error_count
.load(std::sync::atomic::Ordering::Relaxed)
})
.sum();
let avg_utilization = if !devices.is_empty() {
devices
.values()
.map(|device: &Arc<GpuDevice>| device.current_load())
.sum::<f32>()
/ devices.len() as f32
} else {
0.0
};
MultiGpuStats {
device_count,
total_operations,
total_errors,
avg_utilization_percent: avg_utilization,
uptime: self.creation_time.elapsed(),
}
}
pub async fn set_load_balancing_strategy(
&self,
_strategy: LoadBalancingStrategy,
) -> UnifiedGpuResult<()> {
if !self.multi_gpu_enabled {
return Err(UnifiedGpuError::InvalidOperation(
"Multi-GPU mode not enabled".into(),
));
}
Ok(())
}
pub async fn add_gpu_device(&self, device: Arc<GpuDevice>) -> UnifiedGpuResult<()> {
if !self.multi_gpu_enabled {
return Err(UnifiedGpuError::InvalidOperation(
"Multi-GPU mode not enabled".into(),
));
}
let mut devices: tokio::sync::RwLockWriteGuard<HashMap<DeviceId, Arc<GpuDevice>>> =
self.gpu_devices.write().await;
devices.insert(device.id, Arc::clone(&device));
self.load_balancer.add_device(device).await;
Ok(())
}
pub async fn remove_gpu_device(&self, device_id: DeviceId) -> UnifiedGpuResult<()> {
if !self.multi_gpu_enabled {
return Err(UnifiedGpuError::InvalidOperation(
"Multi-GPU mode not enabled".into(),
));
}
let mut devices: tokio::sync::RwLockWriteGuard<HashMap<DeviceId, Arc<GpuDevice>>> =
self.gpu_devices.write().await;
devices.remove(&device_id);
self.load_balancer.remove_device(device_id).await;
Ok(())
}
}
pub struct EnhancedGpuBufferPool {
pools: HashMap<(u64, wgpu::BufferUsages), Vec<wgpu::Buffer>>,
stats: HashMap<(u64, wgpu::BufferUsages), PoolEntryStats>,
total_created: u64,
total_reused: u64,
last_cleanup: Instant,
}
#[derive(Debug, Clone, Default)]
pub struct PoolEntryStats {
pub created_count: u64,
pub reused_count: u64,
pub last_used: Option<Instant>,
pub total_size_bytes: u64,
}
#[derive(Debug, Clone, Default)]
pub struct BufferPoolStats {
pub total_buffers_created: u64,
pub total_buffers_reused: u64,
pub current_pooled_count: usize,
pub total_pooled_memory_mb: f32,
pub hit_rate_percent: f32,
}
impl EnhancedGpuBufferPool {
pub fn new() -> Self {
Self {
pools: HashMap::new(),
stats: HashMap::new(),
total_created: 0,
total_reused: 0,
last_cleanup: Instant::now(),
}
}
}
impl Default for EnhancedGpuBufferPool {
fn default() -> Self {
Self::new()
}
}
impl EnhancedGpuBufferPool {
pub fn get_or_create(
&mut self,
device: &wgpu::Device,
size: u64,
usage: wgpu::BufferUsages,
label: Option<&str>,
) -> wgpu::Buffer {
let key = (size, usage);
if let Some(buffers) = self.pools.get_mut(&key) {
if let Some(buffer) = buffers.pop() {
self.total_reused += 1;
self.stats.entry(key).or_default().reused_count += 1;
self.stats.get_mut(&key).unwrap().last_used = Some(Instant::now());
return buffer;
}
}
let buffer = device.create_buffer(&wgpu::BufferDescriptor {
label,
size,
usage,
mapped_at_creation: false,
});
self.total_created += 1;
let stats = self.stats.entry(key).or_default();
stats.created_count += 1;
stats.total_size_bytes += size;
stats.last_used = Some(Instant::now());
if self.last_cleanup.elapsed().as_secs() > 30 {
self.cleanup_old_buffers();
}
buffer
}
pub fn return_buffer(&mut self, buffer: wgpu::Buffer, size: u64, usage: wgpu::BufferUsages) {
let key = (size, usage);
self.pools.entry(key).or_default().push(buffer);
}
pub fn get_stats(&self) -> BufferPoolStats {
let total_ops = self.total_created + self.total_reused;
let hit_rate = if total_ops > 0 {
(self.total_reused as f32 / total_ops as f32) * 100.0
} else {
0.0
};
let current_pooled_count = self.pools.values().map(|v| v.len()).sum();
let total_pooled_memory_mb: f32 = self
.pools
.iter()
.map(|((size, _usage), buffers)| {
(*size as f32 * buffers.len() as f32) / 1024.0 / 1024.0
})
.sum();
BufferPoolStats {
total_buffers_created: self.total_created,
total_buffers_reused: self.total_reused,
current_pooled_count,
total_pooled_memory_mb,
hit_rate_percent: hit_rate,
}
}
fn cleanup_old_buffers(&mut self) {
let now = Instant::now();
let cleanup_threshold = std::time::Duration::from_secs(300);
self.pools.retain(|&key, buffers| {
if let Some(stats) = self.stats.get(&key) {
if let Some(last_used) = stats.last_used {
if now.duration_since(last_used) > cleanup_threshold {
buffers.clear();
return false;
}
}
}
true
});
self.last_cleanup = now;
}
}
pub struct GpuDispatcher {
gpu_context: Option<GpuContext>,
cpu_threshold: usize,
gpu_threshold: usize,
}
impl GpuDispatcher {
pub async fn new() -> UnifiedGpuResult<Self> {
let gpu_context = (GpuContext::new().await).ok();
Ok(Self {
gpu_context,
cpu_threshold: 100, gpu_threshold: 1000, })
}
pub fn should_use_gpu(&self, workload_size: usize) -> bool {
self.gpu_context.is_some()
&& workload_size >= self.cpu_threshold
&& workload_size >= self.gpu_threshold
}
pub async fn execute<T, F, G>(&mut self, workload_size: usize, gpu_op: G, cpu_op: F) -> T
where
F: FnOnce() -> T,
G: FnOnce(&mut GpuContext) -> UnifiedGpuResult<T>,
{
if self.should_use_gpu(workload_size) {
if let Some(ref mut ctx) = self.gpu_context {
if let Ok(result) = gpu_op(ctx) {
return result;
}
}
}
cpu_op()
}
}
#[derive(Debug, Clone)]
pub struct MultiGpuStats {
pub device_count: usize,
pub total_operations: usize,
pub total_errors: usize,
pub avg_utilization_percent: f32,
pub uptime: std::time::Duration,
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
#[ignore = "GPU hardware required, may fail in CI/CD environments"]
async fn test_gpu_context_creation() {
let _result = GpuContext::new().await;
}
#[tokio::test]
#[ignore = "GPU hardware required, may fail in CI/CD environments"]
async fn test_gpu_dispatcher() {
let dispatcher = GpuDispatcher::new().await;
assert!(dispatcher.is_ok());
}
#[test]
fn test_gpu_operation_params() {
let mut params = GpuOperationParams::default();
params
.params
.insert("scale".to_string(), GpuParam::Float(2.0));
params.batch_size = 100;
assert_eq!(params.batch_size, 100);
match params.params.get("scale") {
Some(GpuParam::Float(val)) => assert_eq!(*val, 2.0),
_ => panic!("Expected float parameter"),
}
}
}