#![allow(non_snake_case)]
use serde::{Serialize, Deserialize};
use std::collections::HashMap as FxHashMap;
use chrono::{DateTime, Utc};
use crate::device::RiResourceAllocation;
use tokio::sync::RwLock;
use super::core::{RiDevice, RiDeviceType, RiDeviceCapabilities};
use super::pool::RiResourcePoolManager;
use std::sync::Arc;
#[allow(dead_code)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiResourceScheduler {
allocations: Arc<RwLock<FxHashMap<String, RiResourceAllocation>>>,
allocation_history: Arc<RwLock<Vec<RiResourceAllocation>>>,
}
#[cfg_attr(feature = "pyo3", pyo3::prelude::pymethods)]
impl RiResourceScheduler {
#[cfg(feature = "pyo3")]
#[new]
fn new() -> Self {
Self {
allocations: Arc::new(RwLock::new(FxHashMap::default())),
allocation_history: Arc::new(RwLock::new(Vec::new())),
}
}
}
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiDeviceScheduler {
scheduling_policies: FxHashMap<RiDeviceType, RiSchedulingPolicy>,
allocation_history: Arc<RwLock<Vec<RiAllocationRecord>>>,
resource_pool_manager: Arc<RwLock<RiResourcePoolManager>>,
round_robin_counters: Arc<RwLock<FxHashMap<RiDeviceType, usize>>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub enum RiSchedulingPolicy {
FirstFit,
BestFit,
WorstFit,
RoundRobin,
PriorityBased,
LoadBalanced,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiAllocationRecord {
pub allocation_id: String,
pub device_id: String,
pub device_type: RiDeviceType,
pub allocated_at: DateTime<Utc>,
pub released_at: Option<DateTime<Utc>>,
pub duration_seconds: Option<f64>,
pub success: bool,
pub capabilities_required: RiDeviceCapabilities,
}
#[derive(Debug, Clone)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiAllocationRequest {
pub device_type: RiDeviceType,
pub capabilities: RiDeviceCapabilities,
pub priority: u32,
pub timeout_secs: u64,
pub sla_class: Option<super::RiRequestSlaClass>,
pub resource_weights: Option<super::RiResourceWeights>,
pub affinity: Option<super::RiAffinityRules>,
pub anti_affinity: Option<super::RiAffinityRules>,
}
impl RiDeviceScheduler {
pub fn new(resource_pool_manager: Arc<RwLock<RiResourcePoolManager>>) -> Self {
let mut scheduling_policies = FxHashMap::default();
scheduling_policies.insert(RiDeviceType::CPU, RiSchedulingPolicy::LoadBalanced);
scheduling_policies.insert(RiDeviceType::GPU, RiSchedulingPolicy::PriorityBased);
scheduling_policies.insert(RiDeviceType::Memory, RiSchedulingPolicy::BestFit);
scheduling_policies.insert(RiDeviceType::Storage, RiSchedulingPolicy::FirstFit);
scheduling_policies.insert(RiDeviceType::Network, RiSchedulingPolicy::RoundRobin);
scheduling_policies.insert(RiDeviceType::Sensor, RiSchedulingPolicy::FirstFit);
scheduling_policies.insert(RiDeviceType::Actuator, RiSchedulingPolicy::PriorityBased);
scheduling_policies.insert(RiDeviceType::Custom, RiSchedulingPolicy::LoadBalanced);
Self {
scheduling_policies,
allocation_history: Arc::new(RwLock::new(Vec::new())),
resource_pool_manager,
round_robin_counters: Arc::new(RwLock::new(FxHashMap::default())),
}
}
pub fn get_policy(&self, device_type: &RiDeviceType) -> &RiSchedulingPolicy {
self.scheduling_policies.get(device_type).unwrap_or(&RiSchedulingPolicy::FirstFit)
}
pub fn set_policy(&mut self, device_type: RiDeviceType, policy: RiSchedulingPolicy) {
self.scheduling_policies.insert(device_type, policy);
}
pub async fn select_device(&self, request: &RiAllocationRequest) -> Option<Arc<RiDevice>> {
let policy = self.get_policy(&request.device_type);
let available_devices = {
let pool_manager = self.resource_pool_manager.read().await;
let pools = pool_manager.get_pools_by_type(request.device_type);
if pools.is_empty() {
return None;
}
let mut devices: Vec<Arc<RiDevice>> = Vec::new();
for pool in &pools {
let pool_devices = pool.get_available_devices();
devices.extend(pool_devices.into_iter());
}
devices
};
let filtered = self.filter_candidates(available_devices, request);
if filtered.is_empty() {
return None;
}
let scored = self.score_candidates(&filtered, request);
match policy {
RiSchedulingPolicy::FirstFit => self.first_fit(&scored),
RiSchedulingPolicy::BestFit => self.best_fit(&scored, &request.capabilities),
RiSchedulingPolicy::WorstFit => self.worst_fit(&scored, &request.capabilities),
RiSchedulingPolicy::RoundRobin => self.round_robin(&scored, request.device_type).await,
RiSchedulingPolicy::PriorityBased => self.priority_based(&scored, request.priority),
RiSchedulingPolicy::LoadBalanced => self.load_balanced(&scored),
}
}
fn filter_candidates(
&self,
devices: Vec<Arc<RiDevice>>,
request: &RiAllocationRequest,
) -> Vec<Arc<RiDevice>> {
devices
.into_iter()
.filter(|device| device.capabilities().meets_requirements(&request.capabilities))
.filter(|device| {
if let Some(rules) = &request.affinity {
for (key, val) in &rules.required_labels {
match device.metadata().get(key) {
Some(v) if v == val => {}
_ => return false,
}
}
}
if let Some(rules) = &request.anti_affinity {
for (key, val) in &rules.forbidden_labels {
if let Some(v) = device.metadata().get(key) {
if v == val {
return false;
}
}
}
}
true
})
.collect()
}
fn score_candidates(
&self,
candidates: &[Arc<RiDevice>],
request: &RiAllocationRequest,
) -> Vec<Arc<RiDevice>> {
if candidates.is_empty() {
return Vec::new();
}
let sla_multiplier: f64 = match request.sla_class {
Some(super::RiRequestSlaClass::Critical) => 1.5,
Some(super::RiRequestSlaClass::High) => 1.2,
Some(super::RiRequestSlaClass::Medium) => 1.0,
Some(super::RiRequestSlaClass::Low) => 0.8,
None => 1.0,
};
let (compute_weight, memory_weight, storage_weight, bandwidth_weight) =
match &request.resource_weights {
Some(w) => (w.compute_weight, w.memory_weight, w.storage_weight, w.bandwidth_weight),
None => (1.0, 1.0, 1.0, 1.0),
};
let mut scored: Vec<(Arc<RiDevice>, f64)> = candidates
.iter()
.cloned()
.map(|device| {
let caps = device.capabilities();
let fitness = self.calculate_fitness_score(caps, &request.capabilities);
let fitness_score = 1.0 / (1.0 + fitness.max(0.0));
let remaining = self.calculate_remaining_capacity_score(caps);
let health = device.health_score() as f64 / 100.0;
let mut dim_score = 0.0;
if let (Some(req), Some(avail)) = (request.capabilities.compute_units, caps.compute_units) {
if avail > 0 {
let ratio = req as f64 / avail as f64;
dim_score += compute_weight * (1.0 / (1.0 + ratio.max(0.0)));
}
}
if let (Some(req), Some(avail)) = (request.capabilities.memory_gb, caps.memory_gb) {
if avail > 0.0 {
let ratio = req / avail;
dim_score += memory_weight * (1.0 / (1.0 + ratio.max(0.0)));
}
}
if let (Some(req), Some(avail)) = (request.capabilities.storage_gb, caps.storage_gb) {
if avail > 0.0 {
let ratio = req / avail;
dim_score += storage_weight * (1.0 / (1.0 + ratio.max(0.0)));
}
}
if let (Some(req), Some(avail)) = (request.capabilities.bandwidth_gbps, caps.bandwidth_gbps) {
if avail > 0.0 {
let ratio = req / avail;
dim_score += bandwidth_weight * (1.0 / (1.0 + ratio.max(0.0)));
}
}
let remaining_score = (remaining / (1.0 + remaining.abs())).max(0.0);
let mut affinity_bonus = 0.0;
if let Some(rules) = &request.affinity {
for (key, val) in &rules.preferred_labels {
if let Some(v) = device.metadata().get(key) {
if v == val {
affinity_bonus += 0.05;
}
}
}
}
let score = sla_multiplier
* (
0.4 * fitness_score + 0.3 * dim_score + 0.2 * remaining_score
+ 0.1 * health
+ affinity_bonus
);
(device, score)
})
.collect();
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
scored.into_iter().map(|(device, _)| device).collect()
}
fn first_fit(&self, devices: &[Arc<RiDevice>]) -> Option<Arc<RiDevice>> {
devices.first().cloned()
}
fn best_fit(&self, devices: &[Arc<RiDevice>], requirements: &RiDeviceCapabilities) -> Option<Arc<RiDevice>> {
devices.iter()
.filter(|device| device.capabilities().meets_requirements(requirements))
.min_by(|a, b| {
let a_fitness = self.calculate_fitness_score(a.capabilities(), requirements);
let b_fitness = self.calculate_fitness_score(b.capabilities(), requirements);
a_fitness.partial_cmp(&b_fitness).unwrap_or(std::cmp::Ordering::Equal)
})
.cloned()
}
fn worst_fit(&self, devices: &[Arc<RiDevice>], requirements: &RiDeviceCapabilities) -> Option<Arc<RiDevice>> {
devices.iter()
.filter(|device| device.capabilities().meets_requirements(requirements))
.max_by(|a, b| {
let a_score = self.calculate_remaining_capacity_score(a.capabilities());
let b_score = self.calculate_remaining_capacity_score(b.capabilities());
a_score.partial_cmp(&b_score).unwrap_or(std::cmp::Ordering::Equal)
})
.cloned()
}
async fn round_robin(&self, devices: &[Arc<RiDevice>], device_type: RiDeviceType) -> Option<Arc<RiDevice>> {
let mut counters = self.round_robin_counters.write().await;
let counter = counters.entry(device_type)
.or_insert(0);
let index = *counter % devices.len();
*counter += 1;
devices.get(index).cloned()
}
fn priority_based(&self, devices: &[Arc<RiDevice>], priority: u32) -> Option<Arc<RiDevice>> {
devices.iter()
.max_by(|a, b| {
let a_score = a.health_score() as u32 * priority;
let b_score = b.health_score() as u32 * priority;
a_score.cmp(&b_score)
})
.cloned()
}
fn load_balanced(&self, devices: &[Arc<RiDevice>]) -> Option<Arc<RiDevice>> {
devices.iter()
.min_by(|a, b| {
a.health_score().cmp(&b.health_score()).reverse()
})
.cloned()
}
fn calculate_fitness_score(&self, device_cap: &RiDeviceCapabilities, requirements: &RiDeviceCapabilities) -> f64 {
let mut score = 0.0;
if let (Some(req_mem), Some(avail_mem)) = (requirements.memory_gb, device_cap.memory_gb) {
let used_ratio = req_mem / avail_mem;
score += used_ratio;
}
if let (Some(req_units), Some(avail_units)) = (requirements.compute_units, device_cap.compute_units) {
let used_ratio = req_units as f64 / avail_units as f64;
score += used_ratio;
}
if let (Some(req_storage), Some(avail_storage)) = (requirements.storage_gb, device_cap.storage_gb) {
let used_ratio = req_storage / avail_storage;
score += used_ratio;
}
score
}
fn calculate_remaining_capacity_score(&self, device_cap: &RiDeviceCapabilities) -> f64 {
let mut score = 0.0;
if let Some(mem) = device_cap.memory_gb {
score += mem;
}
if let Some(units) = device_cap.compute_units {
score += units as f64 * 100.0;
}
if let Some(storage) = device_cap.storage_gb {
score += storage;
}
score
}
pub async fn allocate(&self, request: &RiAllocationRequest) -> Option<String> {
if let Some(device) = self.select_device(request).await {
let allocation_id = uuid::Uuid::new_v4().to_string();
self.record_allocation(allocation_id.clone(), device.id().to_string(), device.device_type(), request.capabilities.clone()).await;
Some(allocation_id)
} else {
None
}
}
pub async fn record_allocation(&self, allocation_id: String, device_id: String, device_type: RiDeviceType, capabilities_required: RiDeviceCapabilities) {
let record = RiAllocationRecord {
allocation_id,
device_id,
device_type,
allocated_at: Utc::now(),
released_at: None,
duration_seconds: None,
success: true,
capabilities_required,
};
let mut history = self.allocation_history.write().await;
history.push(record);
if history.len() > 1000 {
history.remove(0);
}
}
pub async fn record_release(&self, allocation_id: &str) {
let mut history = self.allocation_history.write().await;
if let Some(record) = history.iter_mut().find(|r| r.allocation_id == allocation_id) {
record.released_at = Some(Utc::now());
if let Some(released_at) = record.released_at {
if let Ok(duration) = released_at.signed_duration_since(record.allocated_at).to_std() {
record.duration_seconds = Some(duration.as_secs_f64());
}
}
}
}
pub async fn release_device(&self, device_id: &str) -> Result<(), ()> {
let history = self.allocation_history.read().await;
if let Some(record) = history.iter().find(|r| r.device_id == device_id && r.released_at.is_none()) {
let allocation_id = record.allocation_id.clone();
drop(history);
self.record_release(&allocation_id).await;
Ok(())
} else {
Err(())
}
}
pub async fn get_statistics(&self) -> RiAllocationStatistics {
let history = self.allocation_history.read().await;
let total_allocations = history.len();
let successful_allocations = history.iter().filter(|r| r.success).count();
let failed_allocations = total_allocations - successful_allocations;
let completed_allocations: Vec<&RiAllocationRecord> = history.iter()
.filter(|r| r.released_at.is_some())
.collect();
let total_duration_seconds: f64 = completed_allocations.iter()
.filter_map(|r| r.duration_seconds)
.sum();
let average_duration_seconds = if !completed_allocations.is_empty() {
total_duration_seconds / completed_allocations.len() as f64
} else {
0.0
};
let mut by_device_type = FxHashMap::default();
for device_type in [RiDeviceType::CPU, RiDeviceType::GPU, RiDeviceType::Memory,
RiDeviceType::Storage, RiDeviceType::Network, RiDeviceType::Sensor,
RiDeviceType::Actuator, RiDeviceType::Custom] {
let type_allocations = history.iter()
.filter(|r| r.device_type == device_type)
.count();
if type_allocations > 0 {
let type_completed = history.iter()
.filter(|r| r.device_type == device_type && r.released_at.is_some())
.count();
let type_duration: f64 = history.iter()
.filter(|r| r.device_type == device_type)
.filter_map(|r| r.duration_seconds)
.sum();
let type_avg_duration = if type_completed > 0 {
type_duration / type_completed as f64
} else {
0.0
};
by_device_type.insert(device_type, RiDeviceTypeStatistics {
total_allocations: type_allocations,
completed_allocations: type_completed,
average_duration_seconds: type_avg_duration,
});
}
}
RiAllocationStatistics {
total_allocations,
successful_allocations,
failed_allocations,
success_rate: if total_allocations > 0 {
(successful_allocations as f64 / total_allocations as f64) * 100.0
} else {
0.0
},
average_duration_seconds,
by_device_type,
}
}
pub async fn get_recommendations(&self, device_type: &RiDeviceType) -> Vec<RiSchedulingRecommendation> {
let mut recommendations = Vec::with_capacity(4);
let history = self.allocation_history.read().await;
let recent_allocations: Vec<&RiAllocationRecord> = history.iter()
.filter(|r| r.device_type == *device_type)
.rev()
.take(100)
.collect();
if recent_allocations.is_empty() {
recommendations.push(RiSchedulingRecommendation {
recommendation_type: RiSchedulingRecommendationType::UseDefaultPolicy,
description: format!("No recent allocation data for {device_type:?}, using default policy"),
priority: 1,
confidence: 0.5,
});
return recommendations;
}
let avg_duration = recent_allocations.iter()
.filter_map(|r| r.duration_seconds)
.sum::<f64>() / recent_allocations.len() as f64;
let success_rate = recent_allocations.iter().filter(|r| r.success).count() as f64
/ recent_allocations.len() as f64;
if success_rate < 0.8 {
recommendations.push(RiSchedulingRecommendation {
recommendation_type: RiSchedulingRecommendationType::ConsiderPolicyChange,
description: format!("Low success rate ({:.1}%) for {:?}, consider changing scheduling policy",
success_rate * 100.0, device_type),
priority: 3,
confidence: 0.8,
});
}
if avg_duration > 300.0 {
recommendations.push(RiSchedulingRecommendation {
recommendation_type: RiSchedulingRecommendationType::OptimizeForLongRunning,
description: format!("Average allocation duration is {avg_duration:.1} seconds for {device_type:?}, consider load balancing"),
priority: 2,
confidence: 0.7,
});
}
if recent_allocations.len() > 50 && avg_duration < 60.0 {
recommendations.push(RiSchedulingRecommendation {
recommendation_type: RiSchedulingRecommendationType::OptimizeForShortRunning,
description: format!("High frequency of short allocations for {device_type:?}, consider round-robin scheduling"),
priority: 2,
confidence: 0.6,
});
}
recommendations.push(RiSchedulingRecommendation {
recommendation_type: RiSchedulingRecommendationType::ContinueCurrentPolicy,
description: format!("Current scheduling policy appears effective for {device_type:?}"),
priority: 1,
confidence: 0.9,
});
recommendations.sort_by(|a, b| b.priority.cmp(&a.priority));
recommendations
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiAllocationStatistics {
pub total_allocations: usize,
pub successful_allocations: usize,
pub failed_allocations: usize,
pub success_rate: f64,
pub average_duration_seconds: f64,
pub by_device_type: FxHashMap<RiDeviceType, RiDeviceTypeStatistics>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiDeviceTypeStatistics {
pub total_allocations: usize,
pub completed_allocations: usize,
pub average_duration_seconds: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub enum RiSchedulingRecommendationType {
UseDefaultPolicy,
ContinueCurrentPolicy,
ConsiderPolicyChange,
OptimizeForLongRunning,
OptimizeForShortRunning,
LoadBalance,
Prioritize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiSchedulingRecommendation {
pub recommendation_type: RiSchedulingRecommendationType,
pub description: String,
pub priority: u8,
pub confidence: f64,
}